Fix deletion by making it possible to insert activities as a deactivated user

An attempt to ensure something like
https://git.pleroma.social/pleroma/pleroma/-/issues/1415 does not happen
or is at least debuggable.

- Deactivate the user before deletion to ensure no new posts/follows
can be made during it
- Run the deletion in a transaction. This should reduce performance
impact of a deletion since it will only use a single connection. Also
makes sure an account cannot get stuck in a weird state between deleted
and active. Made it possible to disable though, in case someone hits
the issue mentioned above.
- Log more errors

config/config.exs

@@ -652,7 +652,9 @@ config :pleroma, :oauth2,
   issue_new_refresh_token: true,
   clean_expired_tokens: false
 
-config :pleroma, :database, rum_enabled: false
+config :pleroma, :database,
+  rum_enabled: false,
+  rollback_on_activity_deletion_errors: true
 
 config :pleroma, :env, Mix.env()
 

config/description.exs

@@ -72,6 +72,20 @@ frontend_options = [
 config :pleroma, :config_description, [
   %{
     group: :pleroma,
+    key: :database,
+    type: :group,
+    description: "Database settings",
+    children: [
+      %{
+        key: :rollback_on_activity_deletion_errors,
+        type: :boolean,
+        description:
+          "Rollback the transaction if Pleroma fails to delete an activity during user deletion. If you need to disable this, please report the issue you were having on the bugtracker."
+      }
+    ]
+  },
+  %{
+    group: :pleroma,
     key: Pleroma.Upload,
     type: :group,
     description: "Upload general settings",

config/test.exs

@@ -133,6 +133,10 @@ config :pleroma, :side_effects,
   ap_streamer: Pleroma.Web.ActivityPub.ActivityPubMock,
   logger: Pleroma.LoggerMock
 
+# Disable transaction check by default unless the test wants otherwise
+# because all tests run in a transaction.
+config :pleroma, Pleroma.Workers.BackgroundWorker, ignore_transaction_check: true
+
 if File.exists?("./config/test.secret.exs") do
   import_config "test.secret.exs"
 else

lib/pleroma/object/fetcher.ex

@@ -37,7 +37,7 @@ defmodule Pleroma.Object.Fetcher do
     Logger.debug("Reinjecting object #{new_data["id"]}")
 
     with data <- maybe_reinject_internal_fields(object, new_data),
-         {:ok, data, _} <- ObjectValidator.validate(data, %{}),
+         {:ok, data, _} <- ObjectValidator.validate(data, []),
          changeset <- Object.change(object, %{data: data}),
          changeset <- touch_changeset(changeset),
          {:ok, object} <- Repo.insert_or_update(changeset),

lib/pleroma/user.ex

@@ -1072,7 +1072,19 @@ defmodule Pleroma.User do
 
   def update_and_set_cache(changeset) do
     with {:ok, user} <- Repo.update(changeset, stale_error_field: :id) do
-      set_cache(user)
+      BackgroundWorker.execute_or_enqueue_if_in_transaction(fn
+        false ->
+          set_cache(user)
+
+        # If the function has been enqueued, there is a chance something changed
+        # before the worker got to executing it, so refetch the user from the database
+        true ->
+          user.id
+          |> get_by_id()
+          |> set_cache()
+      end)
+
+      {:ok, user}
     end
   end
 
@@ -1339,7 +1351,7 @@ defmodule Pleroma.User do
 
       user
       |> follow_information_changeset(%{follower_count: follower_count})
-      |> update_and_set_cache
+      |> update_and_set_cache()
     else
       {:ok, maybe_fetch_follow_information(user)}
     end
@@ -1726,48 +1738,83 @@ defmodule Pleroma.User do
 
   defp delete_or_deactivate(%User{local: false} = user), do: delete_and_invalidate_cache(user)
 
+  defp delete_or_deactivate(%User{is_confirmed: false} = user),
+    do: delete_and_invalidate_cache(user)
+
+  defp delete_or_deactivate(%User{is_approved: false} = user),
+    do: delete_and_invalidate_cache(user)
+
   defp delete_or_deactivate(%User{local: true} = user) do
-    status = account_status(user)
-
-    case status do
-      :confirmation_pending ->
-        delete_and_invalidate_cache(user)
-
-      :approval_pending ->
-        delete_and_invalidate_cache(user)
-
-      _ ->
-        user
-        |> purge_user_changeset()
-        |> update_and_set_cache()
-    end
+    user
+    |> purge_user_changeset()
+    |> update_and_set_cache()
   end
 
   def perform(:force_password_reset, user), do: force_password_reset(user)
 
   @spec perform(atom(), User.t()) :: {:ok, User.t()}
   def perform(:delete, %User{} = user) do
-    # Remove all relationships
-    user
-    |> get_followers()
-    |> Enum.each(fn follower ->
-      ActivityPub.unfollow(follower, user)
-      unfollow(follower, user)
-    end)
+    # Deactivate the user before starting the deletion
+    # to make sure they are not able to make new posts/follows during it
+    {:ok, user} = set_activation_status(user, false)
 
-    user
-    |> get_friends()
-    |> Enum.each(fn followed ->
-      ActivityPub.unfollow(user, followed)
-      unfollow(user, followed)
-    end)
+    Repo.transaction(
+      fn ->
+        # Remove all relationships
+        # No need to handle errors from ActivityPub.unfollow because
+        # they will automatically rollback the transaction.
+        user
+        |> get_followers()
+        |> Enum.each(fn follower ->
+          ActivityPub.unfollow(follower, user)
+          unfollow(follower, user)
+        end)
 
-    delete_user_activities(user)
-    delete_notifications_from_user_activities(user)
+        user
+        |> get_friends()
+        |> Enum.each(fn followed ->
+          ActivityPub.unfollow(user, followed, nil, true, true)
+          unfollow(user, followed)
+        end)
 
-    delete_outgoing_pending_follow_requests(user)
+        rollback_on_activity_deletion_errors =
+          Config.get([:database, :rollback_on_activity_deletion_errors], true)
 
-    delete_or_deactivate(user)
+        case {delete_user_activities(user), rollback_on_activity_deletion_errors} do
+          {res, rollback} when res == :ok or rollback == false ->
+            case res do
+              {:error, _} ->
+                Logger.warn(fn ->
+                  "Deleting #{user.ap_id}: Failed deleting some of the activities, proceeding anyway."
+                end)
+
+              _ ->
+                :noop
+            end
+
+            delete_notifications_from_user_activities(user)
+
+            delete_outgoing_pending_follow_requests(user)
+
+            case delete_or_deactivate(user) do
+              {:ok, user} -> user
+              {:error, e} -> Repo.rollback(e)
+            end
+
+          {{:error, e}, true} ->
+            Logger.error(fn ->
+              """
+              Deleting #{user.ap_id}: Failed deleting some of the activities, rolling back.
+              Set `config :pleroma, :database, rollback_on_activity_deletion_errors: true`
+              and restart the deletion if you want to continue anyway. Please report this on Pleroma bugtracker.
+              """
+            end)
+
+            Repo.rollback({:deleting_activities, e})
+        end
+      end,
+      timeout: :infinity
+    )
   end
 
   def perform(:set_activation_async, user, status), do: set_activation(user, status)
@@ -1807,20 +1854,52 @@ defmodule Pleroma.User do
     |> Repo.delete_all()
   end
 
+  @type activity_id :: String.t()
+  @spec delete_user_activities(User.t()) ::
+          :ok | {:error, [{:error, activity_id(), any()}]}
   def delete_user_activities(%User{ap_id: ap_id} = user) do
-    ap_id
-    |> Activity.Queries.by_actor()
-    |> Repo.chunk_stream(50, :batches)
-    |> Stream.each(fn activities ->
-      Enum.each(activities, fn activity -> delete_activity(activity, user) end)
-    end)
-    |> Stream.run()
+    errors =
+      ap_id
+      |> Activity.Queries.by_actor()
+      |> Repo.chunk_stream(50)
+      |> Stream.flat_map(fn activity ->
+        case delete_activity(activity, user) do
+          {:ok, _activity, _meta} ->
+            []
+
+          {:error, error} ->
+            Logger.error(fn ->
+              "Deleting #{ap_id}: could not delete or undo #{activity.data["id"]}.\n Reason: #{
+                inspect(error)
+              }"
+            end)
+
+            [{:error, activity.id, error}]
+
+          :noop ->
+            Logger.debug(fn ->
+              "Deleting #{ap_id}: nothing to do for #{activity.data["id"]} of type #{
+                activity.data["type"]
+              }"
+            end)
+
+            []
+        end
+      end)
+      |> Enum.to_list()
+
+    case errors do
+      [] -> :ok
+      errors -> {:error, errors}
+    end
   end
 
+  @spec delete_activity(Pleroma.Activity.t(), User.t()) ::
+          {:ok, Activity.t(), keyword()} | {:error, any()} | :noop
   defp delete_activity(%{data: %{"type" => "Create", "object" => object}} = activity, user) do
     with {_, %Object{}} <- {:find_object, Object.get_by_ap_id(object)},
          {:ok, delete_data, _} <- Builder.delete(user, object) do
-      Pipeline.common_pipeline(delete_data, local: user.local)
+      Pipeline.common_pipeline(delete_data, local: user.local, allow_deactivated_actor: true)
     else
       {:find_object, nil} ->
         # We have the create activity, but not the object, it was probably pruned.
@@ -1831,18 +1910,20 @@ defmodule Pleroma.User do
         end
 
       e ->
-        Logger.error("Could not delete #{object} created by #{activity.data["ap_id"]}")
-        Logger.error("Error: #{inspect(e)}")
+        e
     end
   end
 
   defp delete_activity(%{data: %{"type" => type}} = activity, user)
        when type in ["Like", "Announce"] do
-    {:ok, undo, _} = Builder.undo(user, activity)
-    Pipeline.common_pipeline(undo, local: user.local)
+    with {:ok, undo, _} <- Builder.undo(user, activity) do
+      Pipeline.common_pipeline(undo, local: user.local, allow_deactivated_actor: true)
+    else
+      e -> e
+    end
   end
 
-  defp delete_activity(_activity, _user), do: "Doing nothing"
+  defp delete_activity(_activity, _user), do: :noop
 
   defp delete_outgoing_pending_follow_requests(user) do
     user

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -318,20 +318,22 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     end
   end
 
-  @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
+  @spec unfollow(User.t(), User.t(), String.t() | nil, boolean(), boolean()) ::
           {:ok, Activity.t()} | nil | {:error, any()}
-  def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
+  def unfollow(follower, followed, activity_id \\ nil, local \\ true, bypass_actor_check \\ false) do
     with {:ok, result} <-
-           Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
+           Repo.transaction(fn ->
+             do_unfollow(follower, followed, activity_id, local, bypass_actor_check)
+           end) do
       result
     end
   end
 
-  defp do_unfollow(follower, followed, activity_id, local) do
+  defp do_unfollow(follower, followed, activity_id, local, bypass_actor_check) do
     with %Activity{} = follow_activity <- fetch_latest_follow(follower, followed),
          {:ok, follow_activity} <- update_follow_state(follow_activity, "cancelled"),
          unfollow_data <- make_unfollow_data(follower, followed, follow_activity, activity_id),
-         {:ok, activity} <- insert(unfollow_data, local),
+         {:ok, activity} <- insert(unfollow_data, local, false, bypass_actor_check),
          _ <- notify_and_stream(activity),
          :ok <- maybe_federate(activity) do
       {:ok, activity}

lib/pleroma/web/activity_pub/object_validator.ex

@@ -41,7 +41,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
       when type in ~w[Accept Reject] do
     with {:ok, object} <-
            object
-           |> AcceptRejectValidator.cast_and_validate()
+           |> AcceptRejectValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -51,7 +51,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Event"} = object, meta) do
     with {:ok, object} <-
            object
-           |> EventValidator.cast_and_validate()
+           |> EventValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -61,7 +61,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Follow"} = object, meta) do
     with {:ok, object} <-
            object
-           |> FollowValidator.cast_and_validate()
+           |> FollowValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -71,7 +71,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Block"} = block_activity, meta) do
     with {:ok, block_activity} <-
            block_activity
-           |> BlockValidator.cast_and_validate()
+           |> BlockValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       block_activity = stringify_keys(block_activity)
       outgoing_blocks = Pleroma.Config.get([:activitypub, :outgoing_blocks])
@@ -90,7 +90,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Update"} = update_activity, meta) do
     with {:ok, update_activity} <-
            update_activity
-           |> UpdateValidator.cast_and_validate()
+           |> UpdateValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       update_activity = stringify_keys(update_activity)
       {:ok, update_activity, meta}
@@ -100,7 +100,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Undo"} = object, meta) do
     with {:ok, object} <-
            object
-           |> UndoValidator.cast_and_validate()
+           |> UndoValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       undone_object = Activity.get_by_ap_id(object["object"])
@@ -114,7 +114,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   end
 
   def validate(%{"type" => "Delete"} = object, meta) do
-    with cng <- DeleteValidator.cast_and_validate(object),
+    with cng <- DeleteValidator.cast_and_validate(object, meta),
          do_not_federate <- DeleteValidator.do_not_federate?(cng),
          {:ok, object} <- Ecto.Changeset.apply_action(cng, :insert) do
       object = stringify_keys(object)
@@ -126,7 +126,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Like"} = object, meta) do
     with {:ok, object} <-
            object
-           |> LikeValidator.cast_and_validate()
+           |> LikeValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -146,7 +146,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Question"} = object, meta) do
     with {:ok, object} <-
            object
-           |> QuestionValidator.cast_and_validate()
+           |> QuestionValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -156,7 +156,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => type} = object, meta) when type in ~w[Audio Video] do
     with {:ok, object} <-
            object
-           |> AudioVideoValidator.cast_and_validate()
+           |> AudioVideoValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -166,7 +166,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Article"} = object, meta) do
     with {:ok, object} <-
            object
-           |> ArticleNoteValidator.cast_and_validate()
+           |> ArticleNoteValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -176,7 +176,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Answer"} = object, meta) do
     with {:ok, object} <-
            object
-           |> AnswerValidator.cast_and_validate()
+           |> AnswerValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -186,7 +186,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "EmojiReact"} = object, meta) do
     with {:ok, object} <-
            object
-           |> EmojiReactValidator.cast_and_validate()
+           |> EmojiReactValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}
@@ -227,7 +227,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidator do
   def validate(%{"type" => "Announce"} = object, meta) do
     with {:ok, object} <-
            object
-           |> AnnounceValidator.cast_and_validate()
+           |> AnnounceValidator.cast_and_validate(meta)
            |> Ecto.Changeset.apply_action(:insert) do
       object = stringify_keys(object)
       {:ok, object, meta}

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

@@ -27,19 +27,19 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.AcceptRejectValidator do
     |> cast(data, __schema__(:fields))
   end
 
-  def validate_data(cng) do
+  def validate_data(cng, meta \\ []) do
     cng
     |> validate_required([:id, :type, :actor, :to, :cc, :object])
     |> validate_inclusion(:type, ["Accept", "Reject"])
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_object_presence(allowed_types: ["Follow"])
     |> validate_accept_reject_rights()
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data
-    |> validate_data
+    |> validate_data(meta)
   end
 
   def validate_accept_reject_rights(cng) do

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

@@ -29,10 +29,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.AnnounceValidator do
     field(:published, ObjectValidators.DateTime)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -50,11 +50,11 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.AnnounceValidator do
     cng
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta \\ []) do
     data_cng
     |> validate_inclusion(:type, ["Announce"])
     |> validate_required([:id, :type, :object, :actor, :to, :cc])
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_object_presence()
     |> validate_existing_announce()
     |> validate_announcable()

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

@@ -34,10 +34,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.AnswerValidator do
     |> apply_action(:insert)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -50,13 +50,13 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.AnswerValidator do
     |> cast(data, __schema__(:fields))
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta \\ []) do
     data_cng
     |> validate_inclusion(:type, ["Answer"])
     |> validate_required([:id, :inReplyTo, :name, :attributedTo, :actor])
     |> CommonValidations.validate_any_presence([:cc, :to])
     |> CommonValidations.validate_fields_match([:actor, :attributedTo])
-    |> CommonValidations.validate_actor_presence()
+    |> CommonValidations.validate_actor_presence(meta)
     |> CommonValidations.validate_host_match()
   end
 end

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

@@ -58,10 +58,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.ArticleNoteValidator do
     |> apply_action(:insert)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -94,13 +94,13 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.ArticleNoteValidator do
     |> cast_embed(:attachment)
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta \\ []) do
     data_cng
     |> validate_inclusion(:type, ["Article", "Note"])
     |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id])
     |> CommonValidations.validate_any_presence([:cc, :to])
     |> CommonValidations.validate_fields_match([:actor, :attributedTo])
-    |> CommonValidations.validate_actor_presence()
+    |> CommonValidations.validate_actor_presence(meta)
     |> CommonValidations.validate_host_match()
   end
 end

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

@@ -59,10 +59,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.AudioVideoValidator do
     |> apply_action(:insert)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -122,13 +122,13 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.AudioVideoValidator do
     |> cast_embed(:attachment)
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta \\ []) do
     data_cng
     |> validate_inclusion(:type, ["Audio", "Video"])
     |> validate_required([:id, :actor, :attributedTo, :type, :context, :attachment])
     |> CommonValidations.validate_any_presence([:cc, :to])
     |> CommonValidations.validate_fields_match([:actor, :attributedTo])
-    |> CommonValidations.validate_actor_presence()
+    |> CommonValidations.validate_actor_presence(meta)
     |> CommonValidations.validate_host_match()
   end
 end

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

@@ -26,17 +26,17 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.BlockValidator do
     |> cast(data, __schema__(:fields))
   end
 
-  def validate_data(cng) do
+  def validate_data(cng, meta \\ []) do
     cng
     |> validate_required([:id, :type, :actor, :to, :cc, :object])
     |> validate_inclusion(:type, ["Block"])
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_actor_presence(field_name: :object)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data
-    |> validate_data
+    |> validate_data(meta)
   end
 end

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

@@ -36,7 +36,11 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CommonValidations do
     |> validate_change(field_name, fn field_name, actor ->
       case User.get_cached_by_ap_id(actor) do
         %User{is_active: false} ->
-          [{field_name, "user is deactivated"}]
+          unless options[:allow_deactivated_actor] do
+            [{field_name, "user is deactivated"}]
+          else
+            []
+          end
 
         %User{} ->
           []

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

@@ -83,7 +83,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CreateGenericValidator do
     cng
     |> validate_required([:actor, :type, :object])
     |> validate_inclusion(:type, ["Create"])
-    |> CommonValidations.validate_actor_presence()
+    |> CommonValidations.validate_actor_presence(meta)
     |> CommonValidations.validate_any_presence([:to, :cc])
     |> validate_actors_match(meta)
     |> validate_context_match(meta)

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

@@ -53,11 +53,11 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.DeleteValidator do
     Tombstone
     Video
   }
-  def validate_data(cng) do
+  def validate_data(cng, meta \\ []) do
     cng
     |> validate_required([:id, :type, :actor, :to, :cc, :object])
     |> validate_inclusion(:type, ["Delete"])
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_modification_rights()
     |> validate_object_or_user_presence(allowed_types: @deletable_types)
     |> add_deleted_activity_id()
@@ -67,9 +67,9 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.DeleteValidator do
     !same_domain?(cng)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data
-    |> validate_data
+    |> validate_data(meta)
   end
 end

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

@@ -24,10 +24,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.EmojiReactValidator do
     field(:cc, ObjectValidators.Recipients, default: [])
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -70,11 +70,11 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.EmojiReactValidator do
     end
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta \\ []) do
     data_cng
     |> validate_inclusion(:type, ["EmojiReact"])
     |> validate_required([:id, :type, :object, :actor, :context, :to, :cc, :content])
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_object_presence()
     |> validate_emoji()
   end

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

@@ -59,10 +59,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.EventValidator do
     |> apply_action(:insert)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -85,13 +85,13 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.EventValidator do
     |> cast_embed(:attachment)
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta \\ []) do
     data_cng
     |> validate_inclusion(:type, ["Event"])
     |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id])
     |> CommonValidations.validate_any_presence([:cc, :to])
     |> CommonValidations.validate_fields_match([:actor, :attributedTo])
-    |> CommonValidations.validate_actor_presence()
+    |> CommonValidations.validate_actor_presence(meta)
     |> CommonValidations.validate_host_match()
   end
 end

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

@@ -27,18 +27,18 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.FollowValidator do
     |> cast(data, __schema__(:fields))
   end
 
-  def validate_data(cng) do
+  def validate_data(cng, meta \\ []) do
     cng
     |> validate_required([:id, :type, :actor, :to, :cc, :object])
     |> validate_inclusion(:type, ["Follow"])
     |> validate_inclusion(:state, ~w{pending reject accept})
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_actor_presence(field_name: :object)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data
-    |> validate_data
+    |> validate_data(meta)
   end
 end

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

@@ -24,10 +24,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.LikeValidator do
     field(:cc, ObjectValidators.Recipients, default: [])
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -76,11 +76,11 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.LikeValidator do
     end
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta) do
     data_cng
     |> validate_inclusion(:type, ["Like"])
     |> validate_required([:id, :type, :object, :actor, :context, :to, :cc])
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_object_presence()
     |> validate_existing_like()
   end

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

@@ -62,10 +62,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.QuestionValidator do
     |> apply_action(:insert)
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -99,13 +99,13 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.QuestionValidator do
     |> cast_embed(:oneOf)
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta \\ []) do
     data_cng
     |> validate_inclusion(:type, ["Question"])
     |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id])
     |> CommonValidations.validate_any_presence([:cc, :to])
     |> CommonValidations.validate_fields_match([:actor, :attributedTo])
-    |> CommonValidations.validate_actor_presence()
+    |> CommonValidations.validate_actor_presence(meta)
     |> CommonValidations.validate_any_presence([:oneOf, :anyOf])
     |> CommonValidations.validate_host_match()
   end

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

@@ -22,10 +22,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.UndoValidator do
     field(:cc, ObjectValidators.Recipients, default: [])
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data()
-    |> validate_data()
+    |> validate_data(meta)
   end
 
   def cast_data(data) do
@@ -38,11 +38,11 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.UndoValidator do
     |> cast(data, __schema__(:fields))
   end
 
-  def validate_data(data_cng) do
+  def validate_data(data_cng, meta \\ []) do
     data_cng
     |> validate_inclusion(:type, ["Undo"])
     |> validate_required([:id, :type, :object, :actor, :to, :cc])
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_object_presence()
     |> validate_undo_rights()
   end

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

@@ -28,18 +28,18 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.UpdateValidator do
     |> cast(data, __schema__(:fields))
   end
 
-  def validate_data(cng) do
+  def validate_data(cng, meta \\ []) do
     cng
     |> validate_required([:id, :type, :actor, :to, :cc, :object])
     |> validate_inclusion(:type, ["Update"])
-    |> validate_actor_presence()
+    |> validate_actor_presence(meta)
     |> validate_updating_rights()
   end
 
-  def cast_and_validate(data) do
+  def cast_and_validate(data, meta) do
     data
     |> cast_data
-    |> validate_data
+    |> validate_data(meta)
   end
 
   # For now we only support updating users, and here the rule is easy:

lib/pleroma/web/activity_pub/pipeline.ex

@@ -13,6 +13,7 @@ defmodule Pleroma.Web.ActivityPub.Pipeline do
   alias Pleroma.Web.ActivityPub.SideEffects
   alias Pleroma.Web.ActivityPub.Visibility
   alias Pleroma.Web.Federator
+  alias Pleroma.Workers.BackgroundWorker
 
   @side_effects Config.get([:pipeline, :side_effects], SideEffects)
   @federator Config.get([:pipeline, :federator], Federator)
@@ -21,12 +22,17 @@ defmodule Pleroma.Web.ActivityPub.Pipeline do
   @activity_pub Config.get([:pipeline, :activity_pub], ActivityPub)
   @config Config.get([:pipeline, :config], Config)
 
-  @spec common_pipeline(map(), keyword()) ::
+  @type common_pipeline_meta_option ::
+          {:local, boolean()} | {:allow_deactivated_actor, boolean()} | {atom(), term()}
+  @spec common_pipeline(map(), [common_pipeline_meta_option()]) ::
           {:ok, Activity.t() | Object.t(), keyword()} | {:error, any()}
   def common_pipeline(object, meta) do
     case Repo.transaction(fn -> do_common_pipeline(object, meta) end) do
       {:ok, {:ok, activity, meta}} ->
-        @side_effects.handle_after_transaction(meta)
+        BackgroundWorker.execute_or_enqueue_if_in_transaction(fn ->
+          @side_effects.handle_after_transaction(meta)
+        end)
+
         {:ok, activity, meta}
 
       {:ok, value} ->

lib/pleroma/web/streamer.ex

@@ -18,6 +18,7 @@ defmodule Pleroma.Web.Streamer do
   alias Pleroma.Web.OAuth.Token
   alias Pleroma.Web.Plugs.OAuthScopesPlug
   alias Pleroma.Web.StreamerView
+  alias Pleroma.Workers.BackgroundWorker
 
   @mix_env Mix.env()
   @registry Pleroma.Web.StreamerRegistry
@@ -135,9 +136,11 @@ defmodule Pleroma.Web.Streamer do
 
   def stream(topics, items) do
     if should_env_send?() do
-      for topic <- List.wrap(topics), item <- List.wrap(items) do
-        spawn(fn -> do_stream(topic, item) end)
-      end
+      BackgroundWorker.execute_or_enqueue_if_in_transaction(fn ->
+        for topic <- List.wrap(topics), item <- List.wrap(items) do
+          spawn(fn -> do_stream(topic, item) end)
+        end
+      end)
     end
   end
 

lib/pleroma/workers/background_worker.ex

@@ -38,4 +38,45 @@ defmodule Pleroma.Workers.BackgroundWorker do
 
     Pleroma.FollowingRelationship.move_following(origin, target)
   end
+
+  def perform(%Job{args: %{"op" => "transaction_side_effects", "function" => encoded_function}}) do
+    function =
+      encoded_function
+      |> Base.decode64!()
+      |> :erlang.binary_to_term()
+
+    maybe_execute_function_with_worker_info(function, true)
+    :ok
+  end
+
+  @doc "Executes a function right away if not running in transaction. Otherwise enqueues it to be executed by BackgroundWorker after transaction commit. Intended for side effects that can not be rolled back. If the function has an arity of 1, the first argument will be a boolean indicating whether it is run by BackgroundWorker or not."
+  @spec execute_or_enqueue_if_in_transaction((() -> any()) | (boolean() -> any())) ::
+          {:ok, {:enqueued, Oban.Job.t()}}
+          | {:error, {:enqueue, Oban.job_changeset()}}
+          | {:error, {:enqueue, term()}}
+          | {:ok, {:executed, term()}}
+  def execute_or_enqueue_if_in_transaction(function) do
+    if Pleroma.Repo.in_transaction?() and
+         !Pleroma.Config.get([__MODULE__, :ignore_transaction_check], false) do
+      encoded_function =
+        function
+        |> :erlang.term_to_binary()
+        |> Base.encode64()
+
+      case enqueue("transaction_side_effects", %{"function" => encoded_function}) do
+        {:ok, job} -> {:ok, {:enqueued, job}}
+        {:error, e} -> {:error, {:enqueue, e}}
+      end
+    else
+      {:ok, {:executed, maybe_execute_function_with_worker_info(function, false)}}
+    end
+  end
+
+  defp maybe_execute_function_with_worker_info(function, executed_by_worker) do
+    if :erlang.fun_info(function)[:arity] == 1 do
+      function.(executed_by_worker)
+    else
+      function.()
+    end
+  end
 end

test/pleroma/web/activity_pub/object_validators/article_note_validator_test.exs

@@ -29,7 +29,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.ArticleNoteValidatorTest do
     end
 
     test "a basic note validates", %{note: note} do
-      %{valid?: true} = ArticleNoteValidator.cast_and_validate(note)
+      %{valid?: true} = ArticleNoteValidator.cast_and_validate(note, [])
     end
   end
 end

test/pleroma/web/activity_pub/object_validators/like_validation_test.exs

@@ -37,7 +37,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.LikeValidationTest do
     end
 
     test "is valid for a valid object", %{valid_like: valid_like} do
-      assert LikeValidator.cast_and_validate(valid_like).valid?
+      assert LikeValidator.cast_and_validate(valid_like, []).valid?
     end
 
     test "sets the 'to' field to the object actor if no recipients are given", %{
@@ -69,21 +69,21 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.LikeValidationTest do
     test "it errors when the actor is missing or not known", %{valid_like: valid_like} do
       without_actor = Map.delete(valid_like, "actor")
 
-      refute LikeValidator.cast_and_validate(without_actor).valid?
+      refute LikeValidator.cast_and_validate(without_actor, []).valid?
 
       with_invalid_actor = Map.put(valid_like, "actor", "invalidactor")
 
-      refute LikeValidator.cast_and_validate(with_invalid_actor).valid?
+      refute LikeValidator.cast_and_validate(with_invalid_actor, []).valid?
     end
 
     test "it errors when the object is missing or not known", %{valid_like: valid_like} do
       without_object = Map.delete(valid_like, "object")
 
-      refute LikeValidator.cast_and_validate(without_object).valid?
+      refute LikeValidator.cast_and_validate(without_object, []).valid?
 
       with_invalid_object = Map.put(valid_like, "object", "invalidobject")
 
-      refute LikeValidator.cast_and_validate(with_invalid_object).valid?
+      refute LikeValidator.cast_and_validate(with_invalid_object, []).valid?
     end
 
     test "it errors when the actor has already like the object", %{
@@ -93,7 +93,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.LikeValidationTest do
     } do
       _like = CommonAPI.favorite(user, post_activity.id)
 
-      refute LikeValidator.cast_and_validate(valid_like).valid?
+      refute LikeValidator.cast_and_validate(valid_like, []).valid?
     end
 
     test "it works when actor or object are wrapped in maps", %{valid_like: valid_like} do
@@ -102,7 +102,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.LikeValidationTest do
         |> Map.put("actor", %{"id" => valid_like["actor"]})
         |> Map.put("object", %{"id" => valid_like["object"]})
 
-      validated = LikeValidator.cast_and_validate(wrapped_like)
+      validated = LikeValidator.cast_and_validate(wrapped_like, [])
 
       assert validated.valid?