Merge branch 'fix/federation-context-issues' into 'develop'

Fix reply context fixing (Pleroma replies to Misskey threads) and removal of context objects

See merge request pleroma/pleroma!3717

lib/pleroma/application.ex

@@ -249,7 +249,8 @@ defmodule Pleroma.Application do
 
   defp background_migrators do
     [
-      Pleroma.Migrators.HashtagsTableMigrator
+      Pleroma.Migrators.HashtagsTableMigrator,
+      Pleroma.Migrators.ContextObjectsDeletionMigrator
     ]
   end
 

lib/pleroma/data_migration.ex

@@ -42,4 +42,5 @@ defmodule Pleroma.DataMigration do
   end
 
   def populate_hashtags_table, do: get_by_name("populate_hashtags_table")
+  def delete_context_objects, do: get_by_name("delete_context_objects")
 end

lib/pleroma/migrators/context_objects_deletion_migrator.ex

@@ -0,0 +1,139 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Migrators.ContextObjectsDeletionMigrator do
+  defmodule State do
+    use Pleroma.Migrators.Support.BaseMigratorState
+
+    @impl Pleroma.Migrators.Support.BaseMigratorState
+    defdelegate data_migration(), to: Pleroma.DataMigration, as: :delete_context_objects
+  end
+
+  use Pleroma.Migrators.Support.BaseMigrator
+
+  alias Pleroma.Migrators.Support.BaseMigrator
+  alias Pleroma.Object
+
+  @doc "This migration removes objects created exclusively for contexts, containing only an `id` field."
+
+  @impl BaseMigrator
+  def feature_config_path, do: [:features, :delete_context_objects]
+
+  @impl BaseMigrator
+  def fault_rate_allowance, do: Config.get([:delete_context_objects, :fault_rate_allowance], 0)
+
+  @impl BaseMigrator
+  def perform do
+    data_migration_id = data_migration_id()
+    max_processed_id = get_stat(:max_processed_id, 0)
+
+    Logger.info("Deleting context objects from `objects` (from oid: #{max_processed_id})...")
+
+    query()
+    |> where([object], object.id > ^max_processed_id)
+    |> Repo.chunk_stream(100, :batches, timeout: :infinity)
+    |> Stream.each(fn objects ->
+      object_ids = Enum.map(objects, & &1.id)
+
+      results = Enum.map(object_ids, &delete_context_object(&1))
+
+      failed_ids =
+        results
+        |> Enum.filter(&(elem(&1, 0) == :error))
+        |> Enum.map(&elem(&1, 1))
+
+      chunk_affected_count =
+        results
+        |> Enum.filter(&(elem(&1, 0) == :ok))
+        |> length()
+
+      for failed_id <- failed_ids do
+        _ =
+          Repo.query(
+            "INSERT INTO data_migration_failed_ids(data_migration_id, record_id) " <>
+              "VALUES ($1, $2) ON CONFLICT DO NOTHING;",
+            [data_migration_id, failed_id]
+          )
+      end
+
+      _ =
+        Repo.query(
+          "DELETE FROM data_migration_failed_ids " <>
+            "WHERE data_migration_id = $1 AND record_id = ANY($2)",
+          [data_migration_id, object_ids -- failed_ids]
+        )
+
+      max_object_id = Enum.at(object_ids, -1)
+
+      put_stat(:max_processed_id, max_object_id)
+      increment_stat(:iteration_processed_count, length(object_ids))
+      increment_stat(:processed_count, length(object_ids))
+      increment_stat(:failed_count, length(failed_ids))
+      increment_stat(:affected_count, chunk_affected_count)
+      put_stat(:records_per_second, records_per_second())
+      persist_state()
+
+      # A quick and dirty approach to controlling the load this background migration imposes
+      sleep_interval = Config.get([:delete_context_objects, :sleep_interval_ms], 0)
+      Process.sleep(sleep_interval)
+    end)
+    |> Stream.run()
+  end
+
+  @impl BaseMigrator
+  def query do
+    # Context objects have no activity type, and only one field, `id`.
+    # Only those context objects are without types.
+    from(
+      object in Object,
+      where: fragment("(?)->'type' IS NULL", object.data),
+      select: %{
+        id: object.id
+      }
+    )
+  end
+
+  @spec delete_context_object(integer()) :: {:ok | :error, integer()}
+  defp delete_context_object(id) do
+    result =
+      %Object{id: id}
+      |> Repo.delete()
+      |> elem(0)
+
+    {result, id}
+  end
+
+  @impl BaseMigrator
+  def retry_failed do
+    data_migration_id = data_migration_id()
+
+    failed_objects_query()
+    |> Repo.chunk_stream(100, :one)
+    |> Stream.each(fn object ->
+      with {res, _} when res != :error <- delete_context_object(object.id) do
+        _ =
+          Repo.query(
+            "DELETE FROM data_migration_failed_ids " <>
+              "WHERE data_migration_id = $1 AND record_id = $2",
+            [data_migration_id, object.id]
+          )
+      end
+    end)
+    |> Stream.run()
+
+    put_stat(:failed_count, failures_count())
+    persist_state()
+
+    force_continue()
+  end
+
+  defp failed_objects_query do
+    from(o in Object)
+    |> join(:inner, [o], dmf in fragment("SELECT * FROM data_migration_failed_ids"),
+      on: dmf.record_id == o.id
+    )
+    |> where([_o, dmf], dmf.data_migration_id == ^data_migration_id())
+    |> order_by([o], asc: o.id)
+  end
+end

lib/pleroma/object.ex

@@ -207,10 +207,6 @@ defmodule Pleroma.Object do
     end
   end
 
-  def context_mapping(context) do
-    Object.change(%Object{}, %{data: %{"id" => context}})
-  end
-
   def make_tombstone(%Object{data: %{"id" => id, "type" => type}}, deleted \\ DateTime.utc_now()) do
     %ObjectTombstone{
       id: id,

lib/pleroma/web/activity_pub/object_validators/article_note_page_validator.ex

@@ -97,7 +97,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.ArticleNotePageValidator do
   defp validate_data(data_cng) do
     data_cng
     |> validate_inclusion(:type, ["Article", "Note", "Page"])
-    |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id])
+    |> validate_required([:id, :actor, :attributedTo, :type, :context])
     |> CommonValidations.validate_any_presence([:cc, :to])
     |> CommonValidations.validate_fields_match([:actor, :attributedTo])
     |> CommonValidations.validate_actor_presence()

lib/pleroma/web/activity_pub/object_validators/common_fields.ex

@@ -52,8 +52,6 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CommonFields do
       field(:summary, :string)
 
       field(:context, :string)
-      # short identifier for PleromaFE to group statuses by context
-      field(:context_id, :integer)
 
       field(:sensitive, :boolean, default: false)
       field(:replies_count, :integer, default: 0)

lib/pleroma/web/activity_pub/object_validators/common_fixes.ex

@@ -22,14 +22,15 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CommonFixes do
   end
 
   def fix_object_defaults(data) do
-    %{data: %{"id" => context}, id: context_id} =
-      Utils.create_context(data["context"] || data["conversation"])
+    context =
+      Utils.maybe_create_context(
+        data["context"] || data["conversation"] || data["inReplyTo"] || data["id"]
+      )
 
     %User{follower_address: follower_collection} = User.get_cached_by_ap_id(data["attributedTo"])
 
     data
     |> Map.put("context", context)
-    |> Map.put("context_id", context_id)
     |> cast_and_filter_recipients("to", follower_collection)
     |> cast_and_filter_recipients("cc", follower_collection)
     |> cast_and_filter_recipients("bto", follower_collection)

lib/pleroma/web/activity_pub/object_validators/create_generic_validator.ex

@@ -75,7 +75,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CreateGenericValidator do
 
     data
     |> CommonFixes.fix_actor()
-    |> Map.put_new("context", object["context"])
+    |> Map.put("context", object["context"])
     |> fix_addressing(object)
   end
 

lib/pleroma/web/activity_pub/object_validators/event_validator.ex

@@ -62,7 +62,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.EventValidator do
   defp validate_data(data_cng) do
     data_cng
     |> validate_inclusion(:type, ["Event"])
-    |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id])
+    |> validate_required([:id, :actor, :attributedTo, :type, :context])
     |> CommonValidations.validate_any_presence([:cc, :to])
     |> CommonValidations.validate_fields_match([:actor, :attributedTo])
     |> CommonValidations.validate_actor_presence()

lib/pleroma/web/activity_pub/object_validators/question_validator.ex

@@ -80,7 +80,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.QuestionValidator do
   defp validate_data(data_cng) do
     data_cng
     |> validate_inclusion(:type, ["Question"])
-    |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id])
+    |> validate_required([:id, :actor, :attributedTo, :type, :context])
     |> CommonValidations.validate_any_presence([:cc, :to])
     |> CommonValidations.validate_fields_match([:actor, :attributedTo])
     |> CommonValidations.validate_actor_presence()

lib/pleroma/web/activity_pub/utils.ex

@@ -154,22 +154,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
     Notification.get_notified_from_activity(%Activity{data: object}, false)
   end
 
-  def create_context(context) do
-    context = context || generate_id("contexts")
-
-    # Ecto has problems accessing the constraint inside the jsonb,
-    # so we explicitly check for the existed object before insert
-    object = Object.get_cached_by_ap_id(context)
-
-    with true <- is_nil(object),
-         changeset <- Object.context_mapping(context),
-         {:ok, inserted_object} <- Repo.insert(changeset) do
-      inserted_object
-    else
-      _ ->
-        object
-    end
-  end
+  def maybe_create_context(context), do: context || generate_id("contexts")
 
   @doc """
   Enqueues an activity for federation if it's local
@@ -201,18 +186,16 @@ defmodule Pleroma.Web.ActivityPub.Utils do
     |> Map.put_new("id", "pleroma:fakeid")
     |> Map.put_new_lazy("published", &make_date/0)
     |> Map.put_new("context", "pleroma:fakecontext")
-    |> Map.put_new("context_id", -1)
     |> lazy_put_object_defaults(true)
   end
 
   def lazy_put_activity_defaults(map, _fake?) do
-    %{data: %{"id" => context}, id: context_id} = create_context(map["context"])
+    context = maybe_create_context(map["context"])
 
     map
     |> Map.put_new_lazy("id", &generate_activity_id/0)
     |> Map.put_new_lazy("published", &make_date/0)
     |> Map.put_new("context", context)
-    |> Map.put_new("context_id", context_id)
     |> lazy_put_object_defaults(false)
   end
 
@@ -226,7 +209,6 @@ defmodule Pleroma.Web.ActivityPub.Utils do
       |> Map.put_new("id", "pleroma:fake_object_id")
       |> Map.put_new_lazy("published", &make_date/0)
       |> Map.put_new("context", activity["context"])
-      |> Map.put_new("context_id", activity["context_id"])
       |> Map.put_new("fake", true)
 
     %{activity | "object" => object}
@@ -239,7 +221,6 @@ defmodule Pleroma.Web.ActivityPub.Utils do
       |> Map.put_new_lazy("id", &generate_object_id/0)
       |> Map.put_new_lazy("published", &make_date/0)
       |> Map.put_new("context", activity["context"])
-      |> Map.put_new("context_id", activity["context_id"])
 
     %{activity | "object" => object}
   end

lib/pleroma/web/api_spec/schemas/status.ex

@@ -148,9 +148,15 @@ defmodule Pleroma.Web.ApiSpec.Schemas.Status do
             description:
               "A map consisting of alternate representations of the `content` property with the key being it's mimetype. Currently the only alternate representation supported is `text/plain`"
           },
+          context: %Schema{
+            type: :string,
+            description: "The thread identifier the status is associated with"
+          },
           conversation_id: %Schema{
             type: :integer,
-            description: "The ID of the AP context the status is associated with (if any)"
+            deprecated: true,
+            description:
+              "The ID of the AP context the status is associated with (if any); deprecated, please use `context` instead"
           },
           direct_conversation_id: %Schema{
             type: :integer,
@@ -325,6 +331,7 @@ defmodule Pleroma.Web.ApiSpec.Schemas.Status do
       "pinned" => false,
       "pleroma" => %{
         "content" => %{"text/plain" => "foobar"},
+        "context" => "http://localhost:4001/objects/8b4c0c80-6a37-4d2a-b1b9-05a19e3875aa",
         "conversation_id" => 345_972,
         "direct_conversation_id" => nil,
         "emoji_reactions" => [],

lib/pleroma/web/common_api/utils.ex

@@ -453,35 +453,6 @@ defmodule Pleroma.Web.CommonAPI.Utils do
 
   def get_report_statuses(_, _), do: {:ok, nil}
 
-  # DEPRECATED mostly, context objects are now created at insertion time.
-  def context_to_conversation_id(context) do
-    with %Object{id: id} <- Object.get_cached_by_ap_id(context) do
-      id
-    else
-      _e ->
-        changeset = Object.context_mapping(context)
-
-        case Repo.insert(changeset) do
-          {:ok, %{id: id}} ->
-            id
-
-          # This should be solved by an upsert, but it seems ecto
-          # has problems accessing the constraint inside the jsonb.
-          {:error, _} ->
-            Object.get_cached_by_ap_id(context).id
-        end
-    end
-  end
-
-  def conversation_id_to_context(id) do
-    with %Object{data: %{"id" => context}} <- Repo.get(Object, id) do
-      context
-    else
-      _e ->
-        {:error, dgettext("errors", "No such conversation")}
-    end
-  end
-
   def validate_character_limit("" = _full_payload, [] = _attachments) do
     {:error, dgettext("errors", "Cannot post an empty status without attachments")}
   end

lib/pleroma/web/mastodon_api/views/status_view.ex

@@ -57,11 +57,19 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do
     end)
   end
 
-  defp get_context_id(%{data: %{"context_id" => context_id}}) when not is_nil(context_id),
-    do: context_id
+  # DEPRECATED This field seems to be a left-over from the StatusNet era.
+  # If your application uses `pleroma.conversation_id`: this field is deprecated.
+  # It is currently stubbed instead by doing a CRC32 of the context, and
+  # clearing the MSB to avoid overflow exceptions with signed integers on the
+  # different clients using this field (Java/Kotlin code, mostly; see Husky.)
+  # This should be removed in a future version of Pleroma. Pleroma-FE currently
+  # depends on this field, as well.
+  defp get_context_id(%{data: %{"context" => context}}) when is_binary(context) do
+    use Bitwise
 
-  defp get_context_id(%{data: %{"context" => context}}) when is_binary(context),
-    do: Utils.context_to_conversation_id(context)
+    :erlang.crc32(context)
+    |> band(bnot(0x8000_0000))
+  end
 
   defp get_context_id(_), do: nil
 
@@ -388,6 +396,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do
       pleroma: %{
         local: activity.local,
         conversation_id: get_context_id(activity),
+        context: object.data["context"],
         in_reply_to_account_acct: reply_to_user && reply_to_user.nickname,
         content: %{"text/plain" => content_plaintext},
         spoiler_text: %{"text/plain" => summary},