Merge branch 'feature/1482-activity_pub_transactions' into 'develop'

ActivityPub actions & side-effects in transaction

Closes #1482

See merge request pleroma/pleroma!2089

.gitignore

@@ -47,3 +47,5 @@ docs/generated_config.md
 .idea
 pleroma.iml
 
+# asdf
+.tool-versions

lib/pleroma/object.ex

@@ -5,6 +5,9 @@
 defmodule Pleroma.Object do
   use Ecto.Schema
 
+  import Ecto.Query
+  import Ecto.Changeset
+
   alias Pleroma.Activity
   alias Pleroma.Object
   alias Pleroma.Object.Fetcher
@@ -12,9 +15,6 @@ defmodule Pleroma.Object do
   alias Pleroma.Repo
   alias Pleroma.User
 
-  import Ecto.Query
-  import Ecto.Changeset
-
   require Logger
 
   @type t() :: %__MODULE__{}
@@ -145,18 +145,18 @@ def authorize_mutation(%Object{data: %{"actor" => actor}}, %User{ap_id: ap_id}),
   # Legacy objects can be mutated by anybody
   def authorize_mutation(%Object{}, %User{}), do: true
 
+  @spec get_cached_by_ap_id(String.t()) :: Object.t() | nil
   def get_cached_by_ap_id(ap_id) do
     key = "object:#{ap_id}"
 
-    Cachex.fetch!(:object_cache, key, fn _ ->
-      object = get_by_ap_id(ap_id)
-
-      if object do
-        {:commit, object}
+    with {:ok, nil} <- Cachex.get(:object_cache, key),
+         object when not is_nil(object) <- get_by_ap_id(ap_id),
+         {:ok, true} <- Cachex.put(:object_cache, key, object) do
+      object
     else
-        {:ignore, object}
+      {:ok, object} -> object
+      nil -> nil
     end
-    end)
   end
 
   def context_mapping(context) do

lib/pleroma/pagination.ex

@@ -12,12 +12,15 @@ defmodule Pleroma.Pagination do
 
   alias Pleroma.Repo
 
+  @type type :: :keyset | :offset
+
   @default_limit 20
   @max_limit 40
   @page_keys ["max_id", "min_id", "limit", "since_id", "order"]
 
   def page_keys, do: @page_keys
 
+  @spec fetch_paginated(Ecto.Query.t(), map(), type(), atom() | nil) :: [Ecto.Schema.t()]
   def fetch_paginated(query, params, type \\ :keyset, table_binding \\ nil)
 
   def fetch_paginated(query, %{"total" => true} = params, :keyset, table_binding) do
@@ -58,6 +61,7 @@ def fetch_paginated(query, params, :offset, table_binding) do
     |> Repo.all()
   end
 
+  @spec paginate(Ecto.Query.t(), map(), type(), atom() | nil) :: [Ecto.Schema.t()]
   def paginate(query, options, method \\ :keyset, table_binding \\ nil)
 
   def paginate(query, options, :keyset, table_binding) do

lib/pleroma/upload.ex

@@ -37,6 +37,7 @@ defmodule Pleroma.Upload do
           Plug.Upload.t()
           | (data_uri_string :: String.t())
           | {:from_local, name :: String.t(), id :: String.t(), path :: String.t()}
+          | map()
 
   @type option ::
           {:type, :avatar | :banner | :background}

lib/pleroma/user.ex

@@ -749,9 +749,18 @@ def invalidate_cache(user) do
     Cachex.del(:user_cache, "nickname:#{user.nickname}")
   end
 
+  @spec get_cached_by_ap_id(String.t()) :: User.t() | nil
   def get_cached_by_ap_id(ap_id) do
     key = "ap_id:#{ap_id}"
-    Cachex.fetch!(:user_cache, key, fn _ -> get_by_ap_id(ap_id) end)
+
+    with {:ok, nil} <- Cachex.get(:user_cache, key),
+         user when not is_nil(user) <- get_by_ap_id(ap_id),
+         {:ok, true} <- Cachex.put(:user_cache, key, user) do
+      user
+    else
+      {:ok, user} -> user
+      nil -> nil
+    end
   end
 
   def get_cached_by_id(id) do

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -6,6 +6,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Activity
   alias Pleroma.Activity.Ir.Topics
   alias Pleroma.Config
+  alias Pleroma.Constants
   alias Pleroma.Conversation
   alias Pleroma.Conversation.Participation
   alias Pleroma.Notification
@@ -124,6 +125,7 @@ def increase_poll_votes_if_vote(%{
 
   def increase_poll_votes_if_vote(_create_data), do: :noop
 
+  @spec insert(map(), boolean(), boolean(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
   def insert(map, local \\ true, fake \\ false, bypass_actor_check \\ false) when is_map(map) do
     with nil <- Activity.normalize(map),
          map <- lazy_put_activity_defaults(map, fake),
@@ -231,12 +233,19 @@ def stream_out(_activity) do
     :noop
   end
 
-  def create(%{to: to, actor: actor, context: context, object: object} = params, fake \\ false) do
+  @spec create(map(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
+  def create(params, fake \\ false) do
+    with {:ok, result} <- Repo.transaction(fn -> do_create(params, fake) end) do
+      result
+    end
+  end
+
+  defp do_create(%{to: to, actor: actor, context: context, object: object} = params, fake) do
     additional = params[:additional] || %{}
     # only accept false as false value
     local = !(params[:local] == false)
     published = params[:published]
-    quick_insert? = Pleroma.Config.get([:env]) == :benchmark
+    quick_insert? = Config.get([:env]) == :benchmark
 
     with create_data <-
            make_create_data(
@@ -259,10 +268,11 @@ def create(%{to: to, actor: actor, context: context, object: object} = params, f
         {:ok, activity}
 
       {:error, message} ->
-        {:error, message}
+        Repo.rollback(message)
     end
   end
 
+  @spec listen(map()) :: {:ok, Activity.t()} | {:error, any()}
   def listen(%{to: to, actor: actor, context: context, object: object} = params) do
     additional = params[:additional] || %{}
     # only accept false as false value
@@ -277,20 +287,20 @@ def listen(%{to: to, actor: actor, context: context, object: object} = params) d
          {:ok, activity} <- insert(listen_data, local),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
-    else
-      {:error, message} ->
-        {:error, message}
     end
   end
 
+  @spec accept(map()) :: {:ok, Activity.t()} | {:error, any()}
   def accept(params) do
     accept_or_reject("Accept", params)
   end
 
+  @spec reject(map()) :: {:ok, Activity.t()} | {:error, any()}
   def reject(params) do
     accept_or_reject("Reject", params)
   end
 
+  @spec accept_or_reject(String.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
   def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
     local = Map.get(params, :local, true)
     activity_id = Map.get(params, :activity_id, nil)
@@ -304,6 +314,7 @@ def accept_or_reject(type, %{to: to, actor: actor, object: object} = params) do
     end
   end
 
+  @spec update(map()) :: {:ok, Activity.t()} | {:error, any()}
   def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
     local = !(params[:local] == false)
     activity_id = params[:activity_id]
@@ -322,7 +333,16 @@ def update(%{to: to, cc: cc, actor: actor, object: object} = params) do
     end
   end
 
+  @spec react_with_emoji(User.t(), Object.t(), String.t(), keyword()) ::
+          {:ok, Activity.t(), Object.t()} | {:error, any()}
   def react_with_emoji(user, object, emoji, options \\ []) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_react_with_emoji(user, object, emoji, options) end) do
+      result
+    end
+  end
+
+  defp do_react_with_emoji(user, object, emoji, options) do
     with local <- Keyword.get(options, :local, true),
          activity_id <- Keyword.get(options, :activity_id, nil),
          true <- Pleroma.Emoji.is_unicode_emoji?(emoji),
@@ -332,11 +352,21 @@ def react_with_emoji(user, object, emoji, options \\ []) do
          :ok <- maybe_federate(activity) do
       {:ok, activity, object}
     else
-      e -> {:error, e}
+      false -> {:error, false}
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
+  @spec unreact_with_emoji(User.t(), String.t(), keyword()) ::
+          {:ok, Activity.t(), Object.t()} | {:error, any()}
   def unreact_with_emoji(user, reaction_id, options \\ []) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_unreact_with_emoji(user, reaction_id, options) end) do
+      result
+    end
+  end
+
+  defp do_unreact_with_emoji(user, reaction_id, options) do
     with local <- Keyword.get(options, :local, true),
          activity_id <- Keyword.get(options, :activity_id, nil),
          user_ap_id <- user.ap_id,
@@ -348,16 +378,24 @@ def unreact_with_emoji(user, reaction_id, options \\ []) do
          :ok <- maybe_federate(activity) do
       {:ok, activity, object}
     else
-      e -> {:error, e}
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
   # TODO: This is weird, maybe we shouldn't check here if we can make the activity.
-  def like(
+  @spec like(User.t(), Object.t(), String.t() | nil, boolean()) ::
+          {:ok, Activity.t(), Object.t()} | {:error, any()}
+  def like(user, object, activity_id \\ nil, local \\ true) do
+    with {:ok, result} <- Repo.transaction(fn -> do_like(user, object, activity_id, local) end) do
+      result
+    end
+  end
+
+  defp do_like(
          %User{ap_id: ap_id} = user,
          %Object{data: %{"id" => _}} = object,
-        activity_id \\ nil,
-        local \\ true
+         activity_id,
+         local
        ) do
     with nil <- get_existing_like(ap_id, object),
          like_data <- make_like_data(user, object, activity_id),
@@ -366,12 +404,24 @@ def like(
          :ok <- maybe_federate(activity) do
       {:ok, activity, object}
     else
-      %Activity{} = activity -> {:ok, activity, object}
-      error -> {:error, error}
+      %Activity{} = activity ->
+        {:ok, activity, object}
+
+      {:error, error} ->
+        Repo.rollback(error)
     end
   end
 
+  @spec unlike(User.t(), Object.t(), String.t() | nil, boolean()) ::
+          {:ok, Activity.t(), Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
   def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ true) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_unlike(actor, object, activity_id, local) end) do
+      result
+    end
+  end
+
+  defp do_unlike(actor, object, activity_id, local) do
     with %Activity{} = like_activity <- get_existing_like(actor.ap_id, object),
          unlike_data <- make_unlike_data(actor, like_activity, activity_id),
          {:ok, unlike_activity} <- insert(unlike_data, local),
@@ -380,10 +430,13 @@ def unlike(%User{} = actor, %Object{} = object, activity_id \\ nil, local \\ tru
          :ok <- maybe_federate(unlike_activity) do
       {:ok, unlike_activity, like_activity, object}
     else
-      _e -> {:ok, object}
+      nil -> {:ok, object}
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
+  @spec announce(User.t(), Object.t(), String.t() | nil, boolean(), boolean()) ::
+          {:ok, Activity.t(), Object.t()} | {:error, any()}
   def announce(
         %User{ap_id: _} = user,
         %Object{data: %{"id" => _}} = object,
@@ -391,6 +444,13 @@ def announce(
         local \\ true,
         public \\ true
       ) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_announce(user, object, activity_id, local, public) end) do
+      result
+    end
+  end
+
+  defp do_announce(user, object, activity_id, local, public) do
     with true <- is_announceable?(object, user, public),
          announce_data <- make_announce_data(user, object, activity_id, public),
          {:ok, activity} <- insert(announce_data, local),
@@ -398,16 +458,26 @@ def announce(
          :ok <- maybe_federate(activity) do
       {:ok, activity, object}
     else
-      error -> {:error, error}
+      false -> {:error, false}
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
+  @spec unannounce(User.t(), Object.t(), String.t() | nil, boolean()) ::
+          {:ok, Activity.t(), Object.t()} | {:ok, Object.t()} | {:error, any()}
   def unannounce(
         %User{} = actor,
         %Object{} = object,
         activity_id \\ nil,
         local \\ true
       ) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_unannounce(actor, object, activity_id, local) end) do
+      result
+    end
+  end
+
+  defp do_unannounce(actor, object, activity_id, local) do
     with %Activity{} = announce_activity <- get_existing_announce(actor.ap_id, object),
          unannounce_data <- make_unannounce_data(actor, announce_activity, activity_id),
          {:ok, unannounce_activity} <- insert(unannounce_data, local),
@@ -416,30 +486,61 @@ def unannounce(
          {:ok, object} <- remove_announce_from_object(announce_activity, object) do
       {:ok, unannounce_activity, object}
     else
-      _e -> {:ok, object}
+      nil -> {:ok, object}
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
+  @spec follow(User.t(), User.t(), String.t() | nil, boolean()) ::
+          {:ok, Activity.t()} | {:error, any()}
   def follow(follower, followed, activity_id \\ nil, local \\ true) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_follow(follower, followed, activity_id, local) end) do
+      result
+    end
+  end
+
+  defp do_follow(follower, followed, activity_id, local) do
     with data <- make_follow_data(follower, followed, activity_id),
          {:ok, activity} <- insert(data, local),
          :ok <- maybe_federate(activity),
          _ <- User.set_follow_state_cache(follower.ap_id, followed.ap_id, activity.data["state"]) do
       {:ok, activity}
+    else
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
+  @spec unfollow(User.t(), User.t(), String.t() | nil, boolean()) ::
+          {:ok, Activity.t()} | nil | {:error, any()}
   def unfollow(follower, followed, activity_id \\ nil, local \\ true) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_unfollow(follower, followed, activity_id, local) end) do
+      result
+    end
+  end
+
+  defp do_unfollow(follower, followed, activity_id, local) 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 <- maybe_federate(activity) do
       {:ok, activity}
+    else
+      nil -> nil
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
-  def delete(%User{ap_id: ap_id, follower_address: follower_address} = user) do
+  @spec delete(User.t() | Object.t(), keyword()) :: {:ok, User.t() | Object.t()} | {:error, any()}
+  def delete(entity, options \\ []) do
+    with {:ok, result} <- Repo.transaction(fn -> do_delete(entity, options) end) do
+      result
+    end
+  end
+
+  defp do_delete(%User{ap_id: ap_id, follower_address: follower_address} = user, _) do
     with data <- %{
            "to" => [follower_address],
            "type" => "Delete",
@@ -452,7 +553,7 @@ def delete(%User{ap_id: ap_id, follower_address: follower_address} = user) do
     end
   end
 
-  def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options \\ []) do
+  defp do_delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options) do
     local = Keyword.get(options, :local, true)
     activity_id = Keyword.get(options, :activity_id, nil)
     actor = Keyword.get(options, :actor, actor)
@@ -477,11 +578,22 @@ def delete(%Object{data: %{"id" => id, "actor" => actor}} = object, options \\ [
          {:ok, _actor} <- decrease_note_count_if_public(user, object),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
+    else
+      {:error, error} ->
+        Repo.rollback(error)
     end
   end
 
-  @spec block(User.t(), User.t(), String.t() | nil, boolean) :: {:ok, Activity.t() | nil}
+  @spec block(User.t(), User.t(), String.t() | nil, boolean()) ::
+          {:ok, Activity.t()} | {:error, any()}
   def block(blocker, blocked, activity_id \\ nil, local \\ true) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_block(blocker, blocked, activity_id, local) end) do
+      result
+    end
+  end
+
+  defp do_block(blocker, blocked, activity_id, local) do
     outgoing_blocks = Config.get([:activitypub, :outgoing_blocks])
     unfollow_blocked = Config.get([:activitypub, :unfollow_blocked])
 
@@ -496,20 +608,32 @@ def block(blocker, blocked, activity_id \\ nil, local \\ true) do
          :ok <- maybe_federate(activity) do
       {:ok, activity}
     else
-      _e -> {:ok, nil}
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
+  @spec unblock(User.t(), User.t(), String.t() | nil, boolean()) ::
+          {:ok, Activity.t()} | {:error, any()} | nil
   def unblock(blocker, blocked, activity_id \\ nil, local \\ true) do
+    with {:ok, result} <-
+           Repo.transaction(fn -> do_unblock(blocker, blocked, activity_id, local) end) do
+      result
+    end
+  end
+
+  defp do_unblock(blocker, blocked, activity_id, local) do
     with %Activity{} = block_activity <- fetch_latest_block(blocker, blocked),
          unblock_data <- make_unblock_data(blocker, blocked, block_activity, activity_id),
          {:ok, activity} <- insert(unblock_data, local),
          :ok <- maybe_federate(activity) do
       {:ok, activity}
+    else
+      nil -> nil
+      {:error, error} -> Repo.rollback(error)
     end
   end
 
-  @spec flag(map()) :: {:ok, Activity.t()} | any
+  @spec flag(map()) :: {:ok, Activity.t()} | {:error, any()}
   def flag(
         %{
           actor: actor,
@@ -546,6 +670,7 @@ def flag(
     end
   end
 
+  @spec move(User.t(), User.t(), boolean()) :: {:ok, Activity.t()} | {:error, any()}
   def move(%User{} = origin, %User{} = target, local \\ true) do
     params = %{
       "type" => "Move",
@@ -571,7 +696,7 @@ def move(%User{} = origin, %User{} = target, local \\ true) do
   end
 
   defp fetch_activities_for_context_query(context, opts) do
-    public = [Pleroma.Constants.as_public()]
+    public = [Constants.as_public()]
 
     recipients =
       if opts["user"],
@@ -616,10 +741,11 @@ def fetch_latest_activity_id_for_context(context, opts \\ %{}) do
     |> Repo.one()
   end
 
+  @spec fetch_public_activities(map(), Pagination.type()) :: [Activity.t()]
   def fetch_public_activities(opts \\ %{}, pagination \\ :keyset) do
     opts = Map.drop(opts, ["user"])
 
-    [Pleroma.Constants.as_public()]
+    [Constants.as_public()]
     |> fetch_activities_query(opts)
     |> restrict_unlisted()
     |> Pagination.fetch_paginated(opts, pagination)
@@ -791,9 +917,9 @@ defp user_activities_recipients(%{"godmode" => true}) do
 
   defp user_activities_recipients(%{"reading_user" => reading_user}) do
     if reading_user do
-      [Pleroma.Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
+      [Constants.as_public()] ++ [reading_user.ap_id | User.following(reading_user)]
     else
-      [Pleroma.Constants.as_public()]
+      [Constants.as_public()]
     end
   end
 
@@ -1000,7 +1126,7 @@ defp restrict_unlisted(query) do
         fragment(
           "not (coalesce(?->'cc', '{}'::jsonb) \\?| ?)",
           activity.data,
-          ^[Pleroma.Constants.as_public()]
+          ^[Constants.as_public()]
         )
     )
   end
@@ -1174,7 +1300,7 @@ def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
   @doc """
   Fetch favorites activities of user with order by sort adds to favorites
   """
-  @spec fetch_favourites(User.t(), map(), atom()) :: list(Activity.t())
+  @spec fetch_favourites(User.t(), map(), Pagination.type()) :: list(Activity.t())
   def fetch_favourites(user, params \\ %{}, pagination \\ :keyset) do
     user.ap_id
     |> Activity.Queries.by_actor()
@@ -1212,7 +1338,7 @@ def fetch_activities_bounded_query(query, recipients, recipients_with_public) do
       where:
         fragment("? && ?", activity.recipients, ^recipients) or
           (fragment("? && ?", activity.recipients, ^recipients_with_public) and
-             ^Pleroma.Constants.as_public() in activity.recipients)
+             ^Constants.as_public() in activity.recipients)
     )
   end
 
@@ -1228,6 +1354,7 @@ def fetch_activities_bounded(
     |> Enum.reverse()
   end
 
+  @spec upload(Upload.source(), keyword()) :: {:ok, Object.t()} | {:error, any()}
   def upload(file, opts \\ []) do
     with {:ok, data} <- Upload.store(file, opts) do
       obj_data =
@@ -1325,8 +1452,7 @@ defp normalize_counter(counter) when is_integer(counter), do: counter
   defp normalize_counter(_), do: 0
 
   defp maybe_update_follow_information(data) do
-    with {:enabled, true} <-
-           {:enabled, Pleroma.Config.get([:instance, :external_user_synchronization])},
+    with {:enabled, true} <- {:enabled, Config.get([:instance, :external_user_synchronization])},
          {:ok, info} <- fetch_follow_information_for_user(data) do
       info = Map.merge(data[:info] || %{}, info)
       Map.put(data, :info, info)

mix.lock

@@ -110,3 +110,4 @@
   "web_push_encryption": {:hex, :web_push_encryption, "0.2.3", "a0ceab85a805a30852f143d22d71c434046fbdbafbc7292e7887cec500826a80", [:mix], [{:httpoison, "~> 1.0", [hex: :httpoison, repo: "hexpm", optional: false]}, {:jose, "~> 1.8", [hex: :jose, repo: "hexpm", optional: false]}, {:poison, "~> 3.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm", "9315c8f37c108835cf3f8e9157d7a9b8f420a34f402d1b1620a31aed5b93ecdf"},
   "websocket_client": {:git, "https://github.com/jeremyong/websocket_client.git", "9a6f65d05ebf2725d62fb19262b21f1805a59fbf", []},
 }
+

test/web/activity_pub/activity_pub_test.exs

@@ -8,6 +8,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
 
   alias Pleroma.Activity
   alias Pleroma.Builders.ActivityBuilder
+  alias Pleroma.Config
   alias Pleroma.Notification
   alias Pleroma.Object
   alias Pleroma.User
@@ -15,6 +16,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.AdminAPI.AccountView
   alias Pleroma.Web.CommonAPI
+  alias Pleroma.Web.Federator
 
   import Pleroma.Factory
   import Tesla.Mock
@@ -224,7 +226,7 @@ test "it fetches the appropriate tag-restricted posts" do
 
   describe "insertion" do
     test "drops activities beyond a certain limit" do
-      limit = Pleroma.Config.get([:instance, :remote_limit])
+      limit = Config.get([:instance, :remote_limit])
 
       random_text =
         :crypto.strong_rand_bytes(limit + 1)
@@ -385,6 +387,27 @@ test "can be fetched into a timeline" do
   end
 
   describe "create activities" do
+    test "it reverts create" do
+      user = insert(:user)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} =
+                 ActivityPub.create(%{
+                   to: ["user1", "user2"],
+                   actor: user,
+                   context: "",
+                   object: %{
+                     "to" => ["user1", "user2"],
+                     "type" => "Note",
+                     "content" => "testing"
+                   }
+                 })
+      end
+
+      assert Repo.aggregate(Activity, :count, :id) == 0
+      assert Repo.aggregate(Object, :count, :id) == 0
+    end
+
     test "removes doubled 'to' recipients" do
       user = insert(:user)
 
@@ -852,8 +875,8 @@ test "returns reblogs for users for whom reblogs have not been muted" do
   end
 
   describe "react to an object" do
-    test_with_mock "sends an activity to federation", Pleroma.Web.Federator, [:passthrough], [] do
-      Pleroma.Config.put([:instance, :federating], true)
+    test_with_mock "sends an activity to federation", Federator, [:passthrough], [] do
+      Config.put([:instance, :federating], true)
       user = insert(:user)
       reactor = insert(:user)
       {:ok, activity} = CommonAPI.post(user, %{"status" => "YASSSS queen slay"})
@@ -861,7 +884,7 @@ test "returns reblogs for users for whom reblogs have not been muted" do
 
       {:ok, reaction_activity, _object} = ActivityPub.react_with_emoji(reactor, object, "🔥")
 
-      assert called(Pleroma.Web.Federator.publish(reaction_activity))
+      assert called(Federator.publish(reaction_activity))
     end
 
     test "adds an emoji reaction activity to the db" do
@@ -899,11 +922,26 @@ test "adds an emoji reaction activity to the db" do
                ["☕", [third_user.ap_id]]
              ]
     end
+
+    test "reverts emoji reaction on error" do
+      [user, reactor] = insert_list(2, :user)
+
+      {:ok, activity} = CommonAPI.post(user, %{"status" => "Status"})
+      object = Object.normalize(activity)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.react_with_emoji(reactor, object, "😀")
+      end
+
+      object = Object.get_by_ap_id(object.data["id"])
+      refute object.data["reaction_count"]
+      refute object.data["reactions"]
+    end
   end
 
   describe "unreacting to an object" do
-    test_with_mock "sends an activity to federation", Pleroma.Web.Federator, [:passthrough], [] do
-      Pleroma.Config.put([:instance, :federating], true)
+    test_with_mock "sends an activity to federation", Federator, [:passthrough], [] do
+      Config.put([:instance, :federating], true)
       user = insert(:user)
       reactor = insert(:user)
       {:ok, activity} = CommonAPI.post(user, %{"status" => "YASSSS queen slay"})
@@ -911,12 +949,12 @@ test "adds an emoji reaction activity to the db" do
 
       {:ok, reaction_activity, _object} = ActivityPub.react_with_emoji(reactor, object, "🔥")
 
-      assert called(Pleroma.Web.Federator.publish(reaction_activity))
+      assert called(Federator.publish(reaction_activity))
 
       {:ok, unreaction_activity, _object} =
         ActivityPub.unreact_with_emoji(reactor, reaction_activity.data["id"])
 
-      assert called(Pleroma.Web.Federator.publish(unreaction_activity))
+      assert called(Federator.publish(unreaction_activity))
     end
 
     test "adds an undo activity to the db" do
@@ -937,18 +975,36 @@ test "adds an undo activity to the db" do
       assert object.data["reaction_count"] == 0
       assert object.data["reactions"] == []
     end
+
+    test "reverts emoji unreact on error" do
+      [user, reactor] = insert_list(2, :user)
+      {:ok, activity} = CommonAPI.post(user, %{"status" => "Status"})
+      object = Object.normalize(activity)
+
+      {:ok, reaction_activity, _object} = ActivityPub.react_with_emoji(reactor, object, "😀")
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} =
+                 ActivityPub.unreact_with_emoji(reactor, reaction_activity.data["id"])
+      end
+
+      object = Object.get_by_ap_id(object.data["id"])
+
+      assert object.data["reaction_count"] == 1
+      assert object.data["reactions"] == [["😀", [reactor.ap_id]]]
+    end
   end
 
   describe "like an object" do
-    test_with_mock "sends an activity to federation", Pleroma.Web.Federator, [:passthrough], [] do
-      Pleroma.Config.put([:instance, :federating], true)
+    test_with_mock "sends an activity to federation", Federator, [:passthrough], [] do
+      Config.put([:instance, :federating], true)
       note_activity = insert(:note_activity)
       assert object_activity = Object.normalize(note_activity)
 
       user = insert(:user)
 
       {:ok, like_activity, _object} = ActivityPub.like(user, object_activity)
-      assert called(Pleroma.Web.Federator.publish(like_activity))
+      assert called(Federator.publish(like_activity))
     end
 
     test "returns exist activity if object already liked" do
@@ -963,6 +1019,19 @@ test "returns exist activity if object already liked" do
       assert like_activity == like_activity_exist
     end
 
+    test "reverts like activity on error" do
+      note_activity = insert(:note_activity)
+      object = Object.normalize(note_activity)
+      user = insert(:user)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.like(user, object)
+      end
+
+      assert Repo.aggregate(Activity, :count, :id) == 1
+      assert Repo.get(Object, object.id) == object
+    end
+
     test "adds a like activity to the db" do
       note_activity = insert(:note_activity)
       assert object = Object.normalize(note_activity)
@@ -993,15 +1062,15 @@ test "adds a like activity to the db" do
   end
 
   describe "unliking" do
-    test_with_mock "sends an activity to federation", Pleroma.Web.Federator, [:passthrough], [] do
-      Pleroma.Config.put([:instance, :federating], true)
+    test_with_mock "sends an activity to federation", Federator, [:passthrough], [] do
+      Config.put([:instance, :federating], true)
 
       note_activity = insert(:note_activity)
       object = Object.normalize(note_activity)
       user = insert(:user)
 
       {:ok, object} = ActivityPub.unlike(user, object)
-      refute called(Pleroma.Web.Federator.publish())
+      refute called(Federator.publish())
 
       {:ok, _like_activity, object} = ActivityPub.like(user, object)
       assert object.data["like_count"] == 1
@@ -1009,7 +1078,24 @@ test "adds a like activity to the db" do
       {:ok, unlike_activity, _, object} = ActivityPub.unlike(user, object)
       assert object.data["like_count"] == 0
 
-      assert called(Pleroma.Web.Federator.publish(unlike_activity))
+      assert called(Federator.publish(unlike_activity))
+    end
+
+    test "reverts unliking on error" do
+      note_activity = insert(:note_activity)
+      object = Object.normalize(note_activity)
+      user = insert(:user)
+
+      {:ok, like_activity, object} = ActivityPub.like(user, object)
+      assert object.data["like_count"] == 1
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.unlike(user, object)
+      end
+
+      assert Object.get_by_ap_id(object.data["id"]) == object
+      assert object.data["like_count"] == 1
+      assert Activity.get_by_id(like_activity.id)
     end
 
     test "unliking a previously liked object" do
@@ -1051,6 +1137,21 @@ test "adds an announce activity to the db" do
       assert announce_activity.data["actor"] == user.ap_id
       assert announce_activity.data["context"] == object.data["context"]
     end
+
+    test "reverts annouce from object on error" do
+      note_activity = insert(:note_activity)
+      object = Object.normalize(note_activity)
+      user = insert(:user)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.announce(user, object)
+      end
+
+      reloaded_object = Object.get_by_ap_id(object.data["id"])
+      assert reloaded_object == object
+      refute reloaded_object.data["announcement_count"]
+      refute reloaded_object.data["announcements"]
+    end
   end
 
   describe "announcing a private object" do
@@ -1093,8 +1194,8 @@ test "unannouncing a previously announced object" do
       user = insert(:user)
 
       # Unannouncing an object that is not announced does nothing
-      # {:ok, object} = ActivityPub.unannounce(user, object)
-      # assert object.data["announcement_count"] == 0
+      {:ok, object} = ActivityPub.unannounce(user, object)
+      refute object.data["announcement_count"]
 
       {:ok, announce_activity, object} = ActivityPub.announce(user, object)
       assert object.data["announcement_count"] == 1
@@ -1114,6 +1215,22 @@ test "unannouncing a previously announced object" do
 
       assert Activity.get_by_id(announce_activity.id) == nil
     end
+
+    test "reverts unannouncing on error" do
+      note_activity = insert(:note_activity)
+      object = Object.normalize(note_activity)
+      user = insert(:user)
+
+      {:ok, _announce_activity, object} = ActivityPub.announce(user, object)
+      assert object.data["announcement_count"] == 1
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.unannounce(user, object)
+      end
+
+      object = Object.get_by_ap_id(object.data["id"])
+      assert object.data["announcement_count"] == 1
+    end
   end
 
   describe "uploading files" do
@@ -1148,6 +1265,35 @@ test "fetches the latest Follow activity" do
   end
 
   describe "following / unfollowing" do
+    test "it reverts follow activity" do
+      follower = insert(:user)
+      followed = insert(:user)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.follow(follower, followed)
+      end
+
+      assert Repo.aggregate(Activity, :count, :id) == 0
+      assert Repo.aggregate(Object, :count, :id) == 0
+    end
+
+    test "it reverts unfollow activity" do
+      follower = insert(:user)
+      followed = insert(:user)
+
+      {:ok, follow_activity} = ActivityPub.follow(follower, followed)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.unfollow(follower, followed)
+      end
+
+      activity = Activity.get_by_id(follow_activity.id)
+      assert activity.data["type"] == "Follow"
+      assert activity.data["actor"] == follower.ap_id
+
+      assert activity.data["object"] == followed.ap_id
+    end
+
     test "creates a follow activity" do
       follower = insert(:user)
       followed = insert(:user)
@@ -1194,6 +1340,17 @@ test "creates an undo activity for a pending follow request" do
   end
 
   describe "blocking / unblocking" do
+    test "reverts block activity on error" do
+      [blocker, blocked] = insert_list(2, :user)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.block(blocker, blocked)
+      end
+
+      assert Repo.aggregate(Activity, :count, :id) == 0
+      assert Repo.aggregate(Object, :count, :id) == 0
+    end
+
     test "creates a block activity" do
       blocker = insert(:user)
       blocked = insert(:user)
@@ -1205,6 +1362,21 @@ test "creates a block activity" do
       assert activity.data["object"] == blocked.ap_id
     end
 
+    test "reverts unblock activity on error" do
+      [blocker, blocked] = insert_list(2, :user)
+      {:ok, block_activity} = ActivityPub.block(blocker, blocked)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.unblock(blocker, blocked)
+      end
+
+      assert block_activity.data["type"] == "Block"
+      assert block_activity.data["actor"] == blocker.ap_id
+
+      assert Repo.aggregate(Activity, :count, :id) == 1
+      assert Repo.aggregate(Object, :count, :id) == 1
+    end
+
     test "creates an undo activity for the last block" do
       blocker = insert(:user)
       blocked = insert(:user)
@@ -1226,6 +1398,19 @@ test "creates an undo activity for the last block" do
   describe "deletion" do
     clear_config([:instance, :rewrite_policy])
 
+    test "it reverts deletion on error" do
+      note = insert(:note_activity)
+      object = Object.normalize(note)
+
+      with_mock(Utils, [:passthrough], maybe_federate: fn _ -> {:error, :reverted} end) do
+        assert {:error, :reverted} = ActivityPub.delete(object)
+      end
+
+      assert Repo.aggregate(Activity, :count, :id) == 1
+      assert Repo.get(Object, object.id) == object
+      assert Activity.get_by_id(note.id) == note
+    end
+
     test "it creates a delete activity and deletes the original object" do
       note = insert(:note_activity)
       object = Object.normalize(note)
@@ -1419,7 +1604,7 @@ test "it creates an update activity with the new user data" do
   end
 
   test "returned pinned statuses" do
-    Pleroma.Config.put([:instance, :max_pinned_statuses], 3)
+    Config.put([:instance, :max_pinned_statuses], 3)
     user = insert(:user)
 
     {:ok, activity_one} = CommonAPI.post(user, %{"status" => "HI!!!"})