changelog

CHANGELOG.md

@@ -101,6 +101,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - Admin API: changed json structure for saving config settings.
 - RichMedia: parsers and their order are configured in `rich_media` config.
 - RichMedia: add the rich media ttl based on image expiration time.
+- Thread Containment has been reworked in a more efficient way with a new `default_reply_visibility` setting.
 
 ### Removed
 - Emoji: Remove longfox emojis.

config/config.exs

@@ -250,7 +250,6 @@ config :pleroma, :instance,
   safe_dm_mentions: false,
   healthcheck: false,
   remote_post_retention_days: 90,
-  skip_thread_containment: true,
   limit_to_local_content: :unauthenticated,
   dynamic_configuration: false,
   user_bio_length: 5000,
@@ -259,7 +258,8 @@ config :pleroma, :instance,
   max_remote_account_fields: 20,
   account_field_name_length: 512,
   account_field_value_length: 512,
-  external_user_synchronization: true
+  external_user_synchronization: true,
+  default_reply_visibility: "public"
 
 config :pleroma, :markup,
   # XXX - unfortunately, inline images must be enabled by default right now, because

docs/config.md

@@ -129,7 +129,6 @@ config :pleroma, Pleroma.Emails.Mailer,
 * `remote_post_retention_days`: The default amount of days to retain remote posts when pruning the database.
 * `user_bio_length`: A user bio maximum length (default: `5000`)
 * `user_name_length`: A user name maximum length (default: `100`)
-* `skip_thread_containment`: Skip filter out broken threads. The default is `false`.
 * `limit_to_local_content`: Limit unauthenticated users to search for local statutes and users only. Possible values: `:unauthenticated`, `:all` and `false`. The default is `:unauthenticated`.
 * `dynamic_configuration`: Allow transferring configuration to DB with the subsequent customization from Admin api.
 * `max_account_fields`: The maximum number of custom fields in the user profile (default: `10`)
@@ -137,7 +136,11 @@ config :pleroma, Pleroma.Emails.Mailer,
 * `account_field_name_length`: An account field name maximum length (default: `512`)
 * `account_field_value_length`: An account field value maximum length (default: `512`)
 * `external_user_synchronization`: Enabling following/followers counters synchronization for external users.
-
+* `default_reply_visibility`: The default reply visibility filter:
+  * "none": show all replies in timelines (like GNU Social)
+  * "public": show public replies in timelines (default)
+  * "following": show replies involving only people the user follows (like Mastodon)
+  * "self": show replies only from the user
 
 
 ## :logger

lib/pleroma/activity.ex

@@ -39,6 +39,7 @@ defmodule Pleroma.Activity do
     field(:local, :boolean, default: true)
     field(:actor, :string)
     field(:recipients, {:array, :string}, default: [])
+    field(:recipient_users, {:array, :string}, default: [])
     field(:thread_muted?, :boolean, virtual: true)
     # This is a fake relation, do not use outside of with_preloaded_bookmark/get_bookmark
     has_one(:bookmark, Bookmark)

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -61,6 +61,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     {recipients, to, cc}
   end
 
+  defp get_recipient_users(recipients, actor),
+    do:
+      Enum.filter(recipients, fn recipient ->
+        actor != recipient && !is_nil(User.get_cached_by_ap_id(recipient))
+      end)
+
   defp check_actor_is_active(actor) do
     if not is_nil(actor) do
       with user <- User.get_cached_by_ap_id(actor),
@@ -126,6 +132,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
          {_, true} <- {:remote_limit_error, check_remote_limit(map)},
          {:ok, map} <- MRF.filter(map),
          {recipients, _, _} = get_recipients(map),
+         recipient_users <- get_recipient_users(recipients, map["actor"]),
          {:fake, false, map, recipients} <- {:fake, fake, map, recipients},
          :ok <- Containment.contain_child(map),
          {:ok, map, object} <- insert_full_object(map) do
@@ -134,7 +141,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
           data: map,
           local: local,
           actor: map["actor"],
-          recipients: recipients
+          recipients: recipients,
+          recipient_users: recipient_users
         })
 
       # Splice in the child object if we have one.
@@ -522,7 +530,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     |> maybe_preload_bookmarks(opts)
     |> maybe_set_thread_muted_field(opts)
     |> restrict_blocked(opts)
-    |> restrict_recipients(recipients, opts["user"])
+    |> restrict_recipients(recipients, opts)
     |> where(
       [activity],
       fragment(
@@ -604,25 +612,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   defp restrict_visibility(query, _visibility), do: query
 
-  defp restrict_thread_visibility(query, _, %{skip_thread_containment: true} = _),
-    do: query
-
-  defp restrict_thread_visibility(
-         query,
-         %{"user" => %User{info: %{skip_thread_containment: true}}},
-         _
-       ),
-       do: query
-
-  defp restrict_thread_visibility(query, %{"user" => %User{ap_id: ap_id}}, _) do
-    from(
-      a in query,
-      where: fragment("thread_visibility(?, (?)->>'id') = true", ^ap_id, a.data)
-    )
-  end
-
-  defp restrict_thread_visibility(query, _, _), do: query
-
   def fetch_user_activities(user, reading_user, params \\ %{}) do
     params =
       params
@@ -710,13 +699,66 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   defp restrict_tag(query, _), do: query
 
-  defp restrict_recipients(query, [], _user), do: query
-
-  defp restrict_recipients(query, recipients, nil) do
-    from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
+  defp get_friend_ap_ids(%User{} = user) do
+    from(u in User.get_friends_query(user), select: u.ap_id)
+    |> Repo.all()
   end
 
-  defp restrict_recipients(query, recipients, user) do
+  defp restrict_recipients(query, [], _opts), do: query
+
+  defp restrict_recipients(query, recipients, %{
+         "user" => %User{} = user,
+         "reply_visibility" => visibility
+       })
+       when visibility in ["self", "following"] do
+    reply_recipients =
+      case visibility do
+        "self" -> [user.ap_id]
+        "following" -> [user.ap_id] ++ get_friend_ap_ids(user)
+      end
+
+    from(
+      [activity, object] in query,
+      where:
+        fragment(
+          "? && ? and (?->>'inReplyTo' is null or ? && ? or ? = ?)",
+          ^recipients,
+          activity.recipients,
+          object.data,
+          activity.recipient_users,
+          ^reply_recipients,
+          activity.actor,
+          ^user.ap_id
+        )
+    )
+  end
+
+  defp restrict_recipients(query, recipients, %{
+         "user" => %User{} = user,
+         "reply_visibility" => "public"
+       }) do
+    reply_recipients = [user.ap_id] ++ get_friend_ap_ids(user)
+    public_recipients = [Pleroma.Constants.as_public()]
+
+    from(
+      [activity, object] in query,
+      where:
+        fragment(
+          "? && ? and (?->>'inReplyTo' is null or ? && ? or ? && ? or ? = ?)",
+          ^recipients,
+          activity.recipients,
+          object.data,
+          activity.recipients,
+          ^public_recipients,
+          activity.recipient_users,
+          ^reply_recipients,
+          activity.actor,
+          ^user.ap_id
+        )
+    )
+  end
+
+  defp restrict_recipients(query, recipients, %{"user" => %User{} = user}) do
     from(
       activity in query,
       where: fragment("? && ?", ^recipients, activity.recipients),
@@ -724,6 +766,10 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     )
   end
 
+  defp restrict_recipients(query, recipients, _opts) do
+    from(activity in query, where: fragment("? && ?", ^recipients, activity.recipients))
+  end
+
   defp restrict_local(query, %{"local_only" => true}) do
     from(activity in query, where: activity.local == true)
   end
@@ -921,16 +967,20 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   defp maybe_order(query, _), do: query
 
   def fetch_activities_query(recipients, opts \\ %{}) do
-    config = %{
-      skip_thread_containment: Config.get([:instance, :skip_thread_containment])
-    }
+    opts =
+      if !Map.has_key?(opts, "user") do
+        opts
+      else
+        default_vis = Pleroma.Config.get([:instance, :default_reply_visibility])
+        Map.put_new(opts, "reply_visibility", default_vis)
+      end
 
     Activity
     |> maybe_preload_objects(opts)
     |> maybe_preload_bookmarks(opts)
     |> maybe_set_thread_muted_field(opts)
     |> maybe_order(opts)
-    |> restrict_recipients(recipients, opts["user"])
+    |> restrict_recipients(recipients, opts)
     |> restrict_tag(opts)
     |> restrict_tag_reject(opts)
     |> restrict_tag_all(opts)
@@ -944,7 +994,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     |> restrict_muted(opts)
     |> restrict_media(opts)
     |> restrict_visibility(opts)
-    |> restrict_thread_visibility(opts, config)
     |> restrict_replies(opts)
     |> restrict_reblogs(opts)
     |> restrict_pinned(opts)
@@ -1174,7 +1223,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/activity_pub/visibility.ex

@@ -5,7 +5,6 @@
 defmodule Pleroma.Web.ActivityPub.Visibility do
   alias Pleroma.Activity
   alias Pleroma.Object
-  alias Pleroma.Repo
   alias Pleroma.User
 
   require Pleroma.Constants
@@ -15,7 +14,10 @@ defmodule Pleroma.Web.ActivityPub.Visibility do
   def is_public?(%Object{data: data}), do: is_public?(data)
   def is_public?(%Activity{data: data}), do: is_public?(data)
   def is_public?(%{"directMessage" => true}), do: false
-  def is_public?(data), do: Pleroma.Constants.as_public() in (data["to"] ++ (data["cc"] || []))
+  def is_public?(%{"to" => to} = data), do: Pleroma.Constants.as_public() in (to ++ (data["cc"] || []))
+  def is_public?(%{"type" => "Create"}), do: false
+  def is_public?(%{"type" => _}), do: true
+  def is_public?(_), do: false
 
   def is_private?(activity) do
     with false <- is_public?(activity),
@@ -54,18 +56,25 @@ defmodule Pleroma.Web.ActivityPub.Visibility do
 
   def visible_for_user?(activity, user) do
     x = [user.ap_id | user.following]
-    y = [activity.actor] ++ activity.data["to"] ++ (activity.data["cc"] || [])
+    y = [activity.actor] ++ activity.recipients
     visible_for_user?(activity, nil) || Enum.any?(x, &(&1 in y))
   end
 
-  def entire_thread_visible_for_user?(%Activity{} = activity, %User{} = user) do
-    {:ok, %{rows: [[result]]}} =
-      Ecto.Adapters.SQL.query(Repo, "SELECT thread_visibility($1, $2)", [
-        user.ap_id,
-        activity.data["id"]
-      ])
+  # XXX: Probably even more inefficient than the previous implementation intended to be a placeholder untill https://git.pleroma.social/pleroma/pleroma/merge_requests/971 is in develop
+  # credo:disable-for-previous-line Credo.Check.Readability.MaxLineLength
+  def entire_thread_visible_for_user?(
+        %Activity{} = tail,
+        # %Activity{data: %{"object" => %{"inReplyTo" => parent_id}}} = tail,
+        user
+      ) do
+    case Object.normalize(tail) do
+      %{data: %{"inReplyTo" => parent_id}} when is_binary(parent_id) ->
+        parent = Activity.get_in_reply_to_activity(tail)
+        visible_for_user?(tail, user) && entire_thread_visible_for_user?(parent, user)
 
-    result
+      _ ->
+        visible_for_user?(tail, user)
+    end
   end
 
   def get_visibility(object) 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
-      true
-    else
-      ActivityPub.contain_activity(activity, user)
-    end
+    ActivityPub.contain_activity(activity, user)
   end
 end

priv/repo/migrations/20190722120237_drop_thread_visibility_function.exs

@@ -0,0 +1,8 @@
+defmodule Pleroma.Repo.Migrations.DropThreadVisibilityFunction do
+  use Ecto.Migration
+  @disable_ddl_transaction true
+
+  def change do
+    execute("drop function if exists thread_visibility(actor varchar, activity_id varchar)")
+  end
+end

priv/repo/migrations/20190825225451_add_recipient_users_to_activities.exs

@@ -0,0 +1,11 @@
+defmodule Pleroma.Repo.Migrations.AddRecipientUsersToActivities do
+  use Ecto.Migration
+
+  def change do
+    alter table(:activities) do
+      add :recipient_users, {:array, :string}
+    end
+
+    create_if_not_exists index(:activities, [:recipient_users], using: :gin)
+  end
+end

priv/repo/migrations/20190825233255_fill_recipient_users_in_activities.exs

@@ -0,0 +1,31 @@
+defmodule Pleroma.Repo.Migrations.FillRecipientUsersInActivities do
+  use Ecto.Migration
+
+  alias Pleroma.RepoStreamer
+  alias Pleroma.User
+
+  import Ecto.Query
+
+  def up do
+    # copy users without as:Public
+    execute("""
+    update activities set recipient_users = array_remove(recipients, 'https://www.w3.org/ns/activitystreams#Public');
+    """)
+
+    # strip followers collections
+    from(
+      u in User,
+      where: not(is_nil(u.follower_address))
+    )
+    |> RepoStreamer.chunk_stream(512)
+    |> Stream.each(fn chunk ->
+      chunk
+      |> Enum.each(fn %User{} = u ->
+        execute("update activities set recipient_users = array_remove(recipient_users, '#{u.follower_address}') where recipient_users && array['#{u.follower_address}'::varchar]")
+      end)
+    end)
+    |> Stream.run()
+  end
+
+  def down, do: :ok
+end

test/web/activity_pub/activity_pub_test.exs

@@ -16,6 +16,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
   import Tesla.Mock
   import Mock
 
+  require Pleroma.Constants
+
   setup do
     mock(fn env -> apply(HttpRequestMock, :request, [env]) end)
     :ok
@@ -278,6 +280,24 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
       assert activity.recipients == ["user1", "user2", user.ap_id]
     end
 
+    test "recipient_users only contains AP IDs of actual users" do
+      user = insert(:user)
+
+      {:ok, activity} =
+        ActivityPub.create(%{
+          to: [Pleroma.Constants.as_public(), user.ap_id],
+          actor: user,
+          context: "",
+          object: %{
+            "to" => [Pleroma.Constants.as_public(), user.ap_id],
+            "type" => "Note",
+            "content" => "testing"
+          }
+        })
+
+      assert activity.recipient_users == [user.ap_id]
+    end
+
     test "increases user note count only for public activities" do
       user = insert(:user)
 

test/web/mastodon_api/mastodon_api_controller_test.exs

@@ -3440,7 +3440,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
     conn3 =
       conn
       |> assign(:user, user3)
-      |> get("api/v1/timelines/home")
+      |> get("api/v1/timelines/home?reply_visibility=none")
 
     [reblogged_activity] = json_response(conn3, 200)