streamer: adapt thread containment changes

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1216,7 +1216,25 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   # do post-processing on a specific activity
   def contain_activity(%Activity{} = activity, %User{} = user) do
-    contain_broken_threads(activity, user)
+    strategy = Pleroma.Config.get([:instance, :default_reply_visibility])
+
+    case strategy do
+      "public" ->
+        Pleroma.Constants.as_public() in activity.recipients
+
+      "self" ->
+        user.ap_id in activity.recipients
+
+      "following" ->
+        friends = ([user.ap_id] ++ get_friend_ap_ids(user)) |> Enum.into(MapSet.new())
+
+        !(activity.recipients
+          |> Enum.into(MapSet.new())
+          |> MapSet.disjoint?(friends))
+
+      _ ->
+        true
+    end
   end
 
   def fetch_direct_messages_query do

lib/pleroma/web/streamer.ex

@@ -6,7 +6,6 @@ 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
@@ -309,10 +308,6 @@ defmodule Pleroma.Web.Streamer do
   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
+    ActivityPub.contain_activity(activity, user)
-      true
-    else
-      ActivityPub.contain_activity(activity, user)
-    end
   end
 end