add option skip_thread_containment

config/config.exs

@@ -237,7 +237,8 @@
   max_report_comment_size: 1000,
   safe_dm_mentions: false,
   healthcheck: false,
-  remote_post_retention_days: 90
+  remote_post_retention_days: 90,
+  skip_thread_containment: false
 
 config :pleroma, :app_account_creation, enabled: true, max_requests: 25, interval: 1800
 

docs/config.md

@@ -105,6 +105,7 @@ config :pleroma, Pleroma.Emails.Mailer,
 * `safe_dm_mentions`: If set to true, only mentions at the beginning of a post will be used to address people in direct messages. This is to prevent accidental mentioning of people when talking about them (e.g. "@friend hey i really don't like @enemy"). (Default: `false`)
 * `healthcheck`: if set to true, system data will be shown on ``/api/pleroma/healthcheck``.
 * `remote_post_retention_days`: the default amount of days to retain remote posts when pruning the database
+* `skip_thread_containment`: Skip filter out broken threads. the default is `false`.
 
 ## :app_account_creation
 REST API for creating an account settings

lib/pleroma/user/info.ex

@@ -49,6 +49,8 @@ defmodule Pleroma.User.Info do
       default: %{"remote" => true, "local" => true, "followers" => true, "follows" => true}
     )
 
+    field(:skip_thread_containment, :boolean, default: false)
+
     # Found in the wild
     # ap_id -> Where is this used?
     # bio -> Where is this used?
@@ -208,7 +210,8 @@ def profile_update(info, params) do
       :hide_followers,
       :hide_favorites,
       :background,
-      :show_role
+      :show_role,
+      :skip_thread_containment
     ])
   end
 

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -4,6 +4,7 @@
 
 defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Activity
+  alias Pleroma.Config
   alias Pleroma.Conversation
   alias Pleroma.Notification
   alias Pleroma.Object
@@ -73,7 +74,7 @@ defp check_actor_is_active(actor) do
   end
 
   defp check_remote_limit(%{"object" => %{"content" => content}}) when not is_nil(content) do
-    limit = Pleroma.Config.get([:instance, :remote_limit])
+    limit = Config.get([:instance, :remote_limit])
     String.length(content) <= limit
   end
 
@@ -399,8 +400,8 @@ def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, local \\ tru
   end
 
   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
-    outgoing_blocks = Pleroma.Config.get([:activitypub, :outgoing_blocks])
+    outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
-    unfollow_blocked = Pleroma.Config.get([:activitypub, :unfollow_blocked])
+    unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
 
     if unfollow_blocked do
       follow_activity = fetch_latest_follow(blocker, blocked)

lib/pleroma/web/streamer.ex

@@ -6,6 +6,7 @@ defmodule Pleroma.Web.Streamer do
   use GenServer
   require Logger
   alias Pleroma.Activity
+  alias Pleroma.Config
   alias Pleroma.Conversation.Participation
   alias Pleroma.Notification
   alias Pleroma.Object
@@ -224,11 +225,10 @@ def push_to_socket(topics, topic, %Activity{data: %{"type" => "Announce"}} = ite
         mutes = user.info.mutes || []
         reblog_mutes = user.info.muted_reblogs || []
 
-        parent = Object.normalize(item)
+        with parent when not is_nil(parent) <- Object.normalize(item),
-
+             true <- Enum.all?([blocks, mutes, reblog_mutes], &(item.actor not in &1)),
-        unless is_nil(parent) or item.actor in blocks or item.actor in mutes or
+             true <- Enum.all?([blocks, mutes], &(parent.data["actor"] not in &1)),
-                 item.actor in reblog_mutes or not ActivityPub.contain_activity(item, user) or
+             true <- thread_containment(item, user) do
-                 parent.data["actor"] in blocks or parent.data["actor"] in mutes do
           send(socket.transport_pid, {:text, represent_update(item, user)})
         end
       else
@@ -264,8 +264,8 @@ def push_to_socket(topics, topic, item) do
         blocks = user.info.blocks || []
         mutes = user.info.mutes || []
 
-        unless item.actor in blocks or item.actor in mutes or
+        with true <- Enum.all?([blocks, mutes], &(item.actor not in &1)),
-                 not ActivityPub.contain_activity(item, user) do
+             true <- thread_containment(item, user) do
           send(socket.transport_pid, {:text, represent_update(item, user)})
         end
       else
@@ -279,4 +279,15 @@ defp internal_topic(topic, socket) when topic in ~w[user direct] do
   end
 
   defp internal_topic(topic, _), do: topic
+
+  @spec thread_containment(Activity.t(), User.t()) :: boolean()
+  defp thread_containment(_activity, %User{info: %{skip_thread_containment: true}}), do: true
+
+  defp thread_containment(activity, user) do
+    if Config.get([:instance, :skip_thread_containment]) do
+      true
+    else
+      ActivityPub.contain_activity(activity, user)
+    end
+  end
 end

test/web/activity_pub/visibilty_test.exs

@@ -1,6 +1,7 @@
 defmodule Pleroma.Web.ActivityPub.VisibilityTest do
   use Pleroma.DataCase
 
+  alias Pleroma.Activity
   alias Pleroma.Web.ActivityPub.Visibility
   alias Pleroma.Web.CommonAPI
   import Pleroma.Factory
@@ -121,4 +122,46 @@ test "get_visibility", %{
   test "get_visibility with directMessage flag" do
     assert Visibility.get_visibility(%{data: %{"directMessage" => true}}) == "direct"
   end
+
+  describe "entire_thread_visible_for_user?/2" do
+    test "returns false if not found activity", %{user: user} do
+      refute Visibility.entire_thread_visible_for_user?(%Activity{}, user)
+    end
+
+    test "returns true if activity hasn't 'Create' type", %{user: user} do
+      activity = insert(:like_activity)
+      assert Visibility.entire_thread_visible_for_user?(activity, user)
+    end
+
+    test "returns false when invalid recipients", %{user: user} do
+      author = insert(:user)
+
+      activity =
+        insert(:note_activity,
+          note:
+            insert(:note,
+              user: author,
+              data: %{"to" => ["test-user"]}
+            )
+        )
+
+      refute Visibility.entire_thread_visible_for_user?(activity, user)
+    end
+
+    test "returns true if user following to author" do
+      author = insert(:user)
+      user = insert(:user, following: [author.ap_id])
+
+      activity =
+        insert(:note_activity,
+          note:
+            insert(:note,
+              user: author,
+              data: %{"to" => [user.ap_id]}
+            )
+        )
+
+      assert Visibility.entire_thread_visible_for_user?(activity, user)
+    end
+  end
 end

test/web/streamer_test.exs

@@ -11,6 +11,16 @@ defmodule Pleroma.Web.StreamerTest do
   alias Pleroma.Web.Streamer
   import Pleroma.Factory
 
+  setup do
+    skip_thread_containment = Pleroma.Config.get([:instance, :skip_thread_containment])
+
+    on_exit(fn ->
+      Pleroma.Config.put([:instance, :skip_thread_containment], skip_thread_containment)
+    end)
+
+    :ok
+  end
+
   test "it sends to public" do
     user = insert(:user)
     other_user = insert(:user)
@@ -68,6 +78,74 @@ test "it sends to public" do
     Task.await(task)
   end
 
+  describe "thread_containment" do
+    test "it doesn't send to user if recipients invalid and thread containment is enabled" do
+      Pleroma.Config.put([:instance, :skip_thread_containment], false)
+      author = insert(:user)
+      user = insert(:user, following: [author.ap_id])
+
+      activity =
+        insert(:note_activity,
+          note:
+            insert(:note,
+              user: author,
+              data: %{"to" => ["TEST-FFF"]}
+            )
+        )
+
+      task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
+      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      topics = %{"public" => [fake_socket]}
+      Streamer.push_to_socket(topics, "public", activity)
+
+      Task.await(task)
+    end
+
+    test "it sends message if recipients invalid and thread containment is disabled" do
+      Pleroma.Config.put([:instance, :skip_thread_containment], true)
+      author = insert(:user)
+      user = insert(:user, following: [author.ap_id])
+
+      activity =
+        insert(:note_activity,
+          note:
+            insert(:note,
+              user: author,
+              data: %{"to" => ["TEST-FFF"]}
+            )
+        )
+
+      task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
+      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      topics = %{"public" => [fake_socket]}
+      Streamer.push_to_socket(topics, "public", activity)
+
+      Task.await(task)
+    end
+
+    test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
+      Pleroma.Config.put([:instance, :skip_thread_containment], false)
+      author = insert(:user)
+      user = insert(:user, following: [author.ap_id], info: %{skip_thread_containment: true})
+
+      activity =
+        insert(:note_activity,
+          note:
+            insert(:note,
+              user: author,
+              data: %{"to" => ["TEST-FFF"]}
+            )
+        )
+
+      task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
+      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      topics = %{"public" => [fake_socket]}
+      Streamer.push_to_socket(topics, "public", activity)
+
+      Task.await(task)
+    end
+  end
+
   test "it doesn't send to blocked users" do
     user = insert(:user)
     blocked_user = insert(:user)