1 files changed, 441 insertions, 0 deletions
diff --git a/lib/pleroma/web/activity_pub/side_effects.ex b/lib/pleroma/web/activity_pub/side_effects.ex
new file mode 100644
index 000000000..a5e2323bd
--- /dev/null
+++ b/lib/pleroma/web/activity_pub/side_effects.ex
@@ -0,0 +1,441 @@
+defmodule Pleroma.Web.ActivityPub.SideEffects do
+  @moduledoc """
+  This module looks at an inserted object and executes the side effects that it
+  implies. For example, a `Like` activity will increase the like count on the
+  liked object, a `Follow` activity will add the user to the follower
+  collection, and so on.
+  """
+  alias Pleroma.Activity
+  alias Pleroma.Activity.Ir.Topics
+  alias Pleroma.ActivityExpiration
+  alias Pleroma.Chat
+  alias Pleroma.Chat.MessageReference
+  alias Pleroma.FollowingRelationship
+  alias Pleroma.Notification
+  alias Pleroma.Object
+  alias Pleroma.Repo
+  alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.ActivityPub
+  alias Pleroma.Web.ActivityPub.Builder
+  alias Pleroma.Web.ActivityPub.Pipeline
+  alias Pleroma.Web.ActivityPub.Utils
+  alias Pleroma.Web.Push
+  alias Pleroma.Web.Streamer
+  alias Pleroma.Workers.BackgroundWorker
+
+  require Logger
+
+  def handle(object, meta \\ [])
+
+  # Task this handles
+  # - Follows
+  # - Sends a notification
+  def handle(
+        %{
+          data: %{
+            "actor" => actor,
+            "type" => "Accept",
+            "object" => follow_activity_id
+          }
+        } = object,
+        meta
+      ) do
+    with %Activity{actor: follower_id} = follow_activity <-
+           Activity.get_by_ap_id(follow_activity_id),
+         %User{} = followed <- User.get_cached_by_ap_id(actor),
+         %User{} = follower <- User.get_cached_by_ap_id(follower_id),
+         {:ok, follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "accept"),
+         {:ok, _relationship} <- FollowingRelationship.update(follower, followed, :follow_accept) do
+      Notification.update_notification_type(followed, follow_activity)
+      User.update_follower_count(followed)
+      User.update_following_count(follower)
+    end
+
+    {:ok, object, meta}
+  end
+
+  # Task this handles
+  # - Rejects all existing follow activities for this person
+  # - Updates the follow state
+  # - Dismisses notification
+  def handle(
+        %{
+          data: %{
+            "actor" => actor,
+            "type" => "Reject",
+            "object" => follow_activity_id
+          }
+        } = object,
+        meta
+      ) do
+    with %Activity{actor: follower_id} = follow_activity <-
+           Activity.get_by_ap_id(follow_activity_id),
+         %User{} = followed <- User.get_cached_by_ap_id(actor),
+         %User{} = follower <- User.get_cached_by_ap_id(follower_id),
+         {:ok, _follow_activity} <- Utils.update_follow_state_for_all(follow_activity, "reject") do
+      FollowingRelationship.update(follower, followed, :follow_reject)
+      Notification.dismiss(follow_activity)
+    end
+
+    {:ok, object, meta}
+  end
+
+  # Tasks this handle
+  # - Follows if possible
+  # - Sends a notification
+  # - Generates accept or reject if appropriate
+  def handle(
+        %{
+          data: %{
+            "id" => follow_id,
+            "type" => "Follow",
+            "object" => followed_user,
+            "actor" => following_user
+          }
+        } = object,
+        meta
+      ) do
+    with %User{} = follower <- User.get_cached_by_ap_id(following_user),
+         %User{} = followed <- User.get_cached_by_ap_id(followed_user),
+         {_, {:ok, _}, _, _} <-
+           {:following, User.follow(follower, followed, :follow_pending), follower, followed} do
+      if followed.local && !followed.locked do
+        {:ok, accept_data, _} = Builder.accept(followed, object)
+        {:ok, _activity, _} = Pipeline.common_pipeline(accept_data, local: true)
+      end
+    else
+      {:following, {:error, _}, _follower, followed} ->
+        {:ok, reject_data, _} = Builder.reject(followed, object)
+        {:ok, _activity, _} = Pipeline.common_pipeline(reject_data, local: true)
+
+      _ ->
+        nil
+    end
+
+    {:ok, notifications} = Notification.create_notifications(object, do_send: false)
+
+    meta =
+      meta
+      |> add_notifications(notifications)
+
+    updated_object = Activity.get_by_ap_id(follow_id)
+
+    {:ok, updated_object, meta}
+  end
+
+  # Tasks this handles:
+  # - Unfollow and block
+  def handle(
+        %{data: %{"type" => "Block", "object" => blocked_user, "actor" => blocking_user}} =
+          object,
+        meta
+      ) do
+    with %User{} = blocker <- User.get_cached_by_ap_id(blocking_user),
+         %User{} = blocked <- User.get_cached_by_ap_id(blocked_user) do
+      User.block(blocker, blocked)
+    end
+
+    {:ok, object, meta}
+  end
+
+  # Tasks this handles:
+  # - Update the user
+  #
+  # For a local user, we also get a changeset with the full information, so we
+  # can update non-federating, non-activitypub settings as well.
+  def handle(%{data: %{"type" => "Update", "object" => updated_object}} = object, meta) do
+    if changeset = Keyword.get(meta, :user_update_changeset) do
+      changeset
+      |> User.update_and_set_cache()
+    else
+      {:ok, new_user_data} = ActivityPub.user_data_from_user_object(updated_object)
+
+      User.get_by_ap_id(updated_object["id"])
+      |> User.remote_user_changeset(new_user_data)
+      |> User.update_and_set_cache()
+    end
+
+    {:ok, object, meta}
+  end
+
+  # Tasks this handles:
+  # - Add like to object
+  # - Set up notification
+  def handle(%{data: %{"type" => "Like"}} = object, meta) do
+    liked_object = Object.get_by_ap_id(object.data["object"])
+    Utils.add_like_to_object(object, liked_object)
+
+    Notification.create_notifications(object)
+
+    {:ok, object, meta}
+  end
+
+  # Tasks this handles
+  # - Actually create object
+  # - Rollback if we couldn't create it
+  # - Increase the user note count
+  # - Increase the reply count
+  # - Increase replies count
+  # - Set up ActivityExpiration
+  # - Set up notifications
+  def handle(%{data: %{"type" => "Create"}} = activity, meta) do
+    with {:ok, object, meta} <- handle_object_creation(meta[:object_data], meta),
+         %User{} = user <- User.get_cached_by_ap_id(activity.data["actor"]) do
+      {:ok, notifications} = Notification.create_notifications(activity, do_send: false)
+      {:ok, _user} = ActivityPub.increase_note_count_if_public(user, object)
+
+      if in_reply_to = object.data["inReplyTo"] do
+        Object.increase_replies_count(in_reply_to)
+      end
+
+      if expires_at = activity.data["expires_at"] do
+        ActivityExpiration.create(activity, expires_at)
+      end
+
+      BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
+
+      meta =
+        meta
+        |> add_notifications(notifications)
+
+      {:ok, activity, meta}
+    else
+      e -> Repo.rollback(e)
+    end
+  end
+
+  # Tasks this handles:
+  # - Add announce to object
+  # - Set up notification
+  # - Stream out the announce
+  def handle(%{data: %{"type" => "Announce"}} = object, meta) do
+    announced_object = Object.get_by_ap_id(object.data["object"])
+    user = User.get_cached_by_ap_id(object.data["actor"])
+
+    Utils.add_announce_to_object(object, announced_object)
+
+    if !User.is_internal_user?(user) do
+      Notification.create_notifications(object)
+
+      object
+      |> Topics.get_activity_topics()
+      |> Streamer.stream(object)
+    end
+
+    {:ok, object, meta}
+  end
+
+  def handle(%{data: %{"type" => "Undo", "object" => undone_object}} = object, meta) do
+    with undone_object <- Activity.get_by_ap_id(undone_object),
+         :ok <- handle_undoing(undone_object) do
+      {:ok, object, meta}
+    end
+  end
+
+  # Tasks this handles:
+  # - Add reaction to object
+  # - Set up notification
+  def handle(%{data: %{"type" => "EmojiReact"}} = object, meta) do
+    reacted_object = Object.get_by_ap_id(object.data["object"])
+    Utils.add_emoji_reaction_to_object(object, reacted_object)
+
+    Notification.create_notifications(object)
+
+    {:ok, object, meta}
+  end
+
+  # Tasks this handles:
+  # - Delete and unpins the create activity
+  # - Replace object with Tombstone
+  # - Set up notification
+  # - Reduce the user note count
+  # - Reduce the reply count
+  # - Stream out the activity
+  def handle(%{data: %{"type" => "Delete", "object" => deleted_object}} = object, meta) do
+    deleted_object =
+      Object.normalize(deleted_object, false) ||
+        User.get_cached_by_ap_id(deleted_object)
+
+    result =
+      case deleted_object do
+        %Object{} ->
+          with {:ok, deleted_object, activity} <- Object.delete(deleted_object),
+               {_, actor} when is_binary(actor) <- {:actor, deleted_object.data["actor"]},
+               %User{} = user <- User.get_cached_by_ap_id(actor) do
+            User.remove_pinnned_activity(user, activity)
+
+            {:ok, user} = ActivityPub.decrease_note_count_if_public(user, deleted_object)
+
+            if in_reply_to = deleted_object.data["inReplyTo"] do
+              Object.decrease_replies_count(in_reply_to)
+            end
+
+            MessageReference.delete_for_object(deleted_object)
+
+            ActivityPub.stream_out(object)
+            ActivityPub.stream_out_participations(deleted_object, user)
+            :ok
+          else
+            {:actor, _} ->
+              Logger.error("The object doesn't have an actor: #{inspect(deleted_object)}")
+              :no_object_actor
+          end
+
+        %User{} ->
+          with {:ok, _} <- User.delete(deleted_object) do
+            :ok
+          end
+      end
+
+    if result == :ok do
+      Notification.create_notifications(object)
+      {:ok, object, meta}
+    else
+      {:error, result}
+    end
+  end
+
+  # Nothing to do
+  def handle(object, meta) do
+    {:ok, object, meta}
+  end
+
+  def handle_object_creation(%{"type" => "ChatMessage"} = object, meta) do
+    with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
+      actor = User.get_cached_by_ap_id(object.data["actor"])
+      recipient = User.get_cached_by_ap_id(hd(object.data["to"]))
+
+      streamables =
+        [[actor, recipient], [recipient, actor]]
+        |> Enum.map(fn [user, other_user] ->
+          if user.local do
+            {:ok, chat} = Chat.bump_or_create(user.id, other_user.ap_id)
+            {:ok, cm_ref} = MessageReference.create(chat, object, user.ap_id != actor.ap_id)
+
+            {
+              ["user", "user:pleroma_chat"],
+              {user, %{cm_ref | chat: chat, object: object}}
+            }
+          end
+        end)
+        |> Enum.filter(& &1)
+
+      meta =
+        meta
+        |> add_streamables(streamables)
+
+      {:ok, object, meta}
+    end
+  end
+
+  def handle_object_creation(%{"type" => "Answer"} = object_map, meta) do
+    with {:ok, object, meta} <- Pipeline.common_pipeline(object_map, meta) do
+      Object.increase_vote_count(
+        object.data["inReplyTo"],
+        object.data["name"],
+        object.data["actor"]
+      )
+
+      {:ok, object, meta}
+    end
+  end
+
+  def handle_object_creation(%{"type" => objtype} = object, meta)
+      when objtype in ~w[Audio Question Event] do
+    with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
+      {:ok, object, meta}
+    end
+  end
+
+  # Nothing to do
+  def handle_object_creation(object, meta) do
+    {:ok, object, meta}
+  end
+
+  defp undo_like(nil, object), do: delete_object(object)
+
+  defp undo_like(%Object{} = liked_object, object) do
+    with {:ok, _} <- Utils.remove_like_from_object(object, liked_object) do
+      delete_object(object)
+    end
+  end
+
+  def handle_undoing(%{data: %{"type" => "Like"}} = object) do
+    object.data["object"]
+    |> Object.get_by_ap_id()
+    |> undo_like(object)
+  end
+
+  def handle_undoing(%{data: %{"type" => "EmojiReact"}} = object) do
+    with %Object{} = reacted_object <- Object.get_by_ap_id(object.data["object"]),
+         {:ok, _} <- Utils.remove_emoji_reaction_from_object(object, reacted_object),
+         {:ok, _} <- Repo.delete(object) do
+      :ok
+    end
+  end
+
+  def handle_undoing(%{data: %{"type" => "Announce"}} = object) do
+    with %Object{} = liked_object <- Object.get_by_ap_id(object.data["object"]),
+         {:ok, _} <- Utils.remove_announce_from_object(object, liked_object),
+         {:ok, _} <- Repo.delete(object) do
+      :ok
+    end
+  end
+
+  def handle_undoing(
+        %{data: %{"type" => "Block", "actor" => blocker, "object" => blocked}} = object
+      ) do
+    with %User{} = blocker <- User.get_cached_by_ap_id(blocker),
+         %User{} = blocked <- User.get_cached_by_ap_id(blocked),
+         {:ok, _} <- User.unblock(blocker, blocked),
+         {:ok, _} <- Repo.delete(object) do
+      :ok
+    end
+  end
+
+  def handle_undoing(object), do: {:error, ["don't know how to handle", object]}
+
+  @spec delete_object(Object.t()) :: :ok | {:error, Ecto.Changeset.t()}
+  defp delete_object(object) do
+    with {:ok, _} <- Repo.delete(object), do: :ok
+  end
+
+  defp send_notifications(meta) do
+    Keyword.get(meta, :notifications, [])
+    |> Enum.each(fn notification ->
+      Streamer.stream(["user", "user:notification"], notification)
+      Push.send(notification)
+    end)
+
+    meta
+  end
+
+  defp send_streamables(meta) do
+    Keyword.get(meta, :streamables, [])
+    |> Enum.each(fn {topics, items} ->
+      Streamer.stream(topics, items)
+    end)
+
+    meta
+  end
+
+  defp add_streamables(meta, streamables) do
+    existing = Keyword.get(meta, :streamables, [])
+
+    meta
+    |> Keyword.put(:streamables, streamables ++ existing)
+  end
+
+  defp add_notifications(meta, notifications) do
+    existing = Keyword.get(meta, :notifications, [])
+
+    meta
+    |> Keyword.put(:notifications, notifications ++ existing)
+  end
+
+  def handle_after_transaction(meta) do
+    meta
+    |> send_notifications()
+    |> send_streamables()
+  end
+end