git.squeep.com Git - akkoma/blob - lib/pleroma/web/activity_pub/side_effects.ex

   1 defmodule Pleroma.Web.ActivityPub.SideEffects do
   2   @moduledoc """
   3   This module looks at an inserted object and executes the side effects that it
   4   implies. For example, a `Like` activity will increase the like count on the
   5   liked object, a `Follow` activity will add the user to the follower
   6   collection, and so on.
   7   """
   8   alias Pleroma.Activity
   9   alias Pleroma.Activity.Ir.Topics
  10   alias Pleroma.Chat
  11   alias Pleroma.Chat.MessageReference
  12   alias Pleroma.Notification
  13   alias Pleroma.Object
  14   alias Pleroma.Repo
  15   alias Pleroma.User
  16   alias Pleroma.Web.ActivityPub.ActivityPub
  17   alias Pleroma.Web.ActivityPub.Pipeline
  18   alias Pleroma.Web.ActivityPub.Utils
  19   alias Pleroma.Web.Push
  20   alias Pleroma.Web.Streamer
  21
  22   def handle(object, meta \\ [])
  23
  24   # Tasks this handles:
  25   # - Unfollow and block
  26   def handle(
  27         %{data: %{"type" => "Block", "object" => blocked_user, "actor" => blocking_user}} =
  28           object,
  29         meta
  30       ) do
  31     with %User{} = blocker <- User.get_cached_by_ap_id(blocking_user),
  32          %User{} = blocked <- User.get_cached_by_ap_id(blocked_user) do
  33       User.block(blocker, blocked)
  34     end
  35
  36     {:ok, object, meta}
  37   end
  38
  39   # Tasks this handles:
  40   # - Update the user
  41   #
  42   # For a local user, we also get a changeset with the full information, so we
  43   # can update non-federating, non-activitypub settings as well.
  44   def handle(%{data: %{"type" => "Update", "object" => updated_object}} = object, meta) do
  45     if changeset = Keyword.get(meta, :user_update_changeset) do
  46       changeset
  47       |> User.update_and_set_cache()
  48     else
  49       {:ok, new_user_data} = ActivityPub.user_data_from_user_object(updated_object)
  50
  51       User.get_by_ap_id(updated_object["id"])
  52       |> User.remote_user_changeset(new_user_data)
  53       |> User.update_and_set_cache()
  54     end
  55
  56     {:ok, object, meta}
  57   end
  58
  59   # Tasks this handles:
  60   # - Add like to object
  61   # - Set up notification
  62   def handle(%{data: %{"type" => "Like"}} = object, meta) do
  63     liked_object = Object.get_by_ap_id(object.data["object"])
  64     Utils.add_like_to_object(object, liked_object)
  65
  66     Notification.create_notifications(object)
  67
  68     {:ok, object, meta}
  69   end
  70
  71   # Tasks this handles
  72   # - Actually create object
  73   # - Rollback if we couldn't create it
  74   # - Set up notifications
  75   def handle(%{data: %{"type" => "Create"}} = activity, meta) do
  76     with {:ok, _object, meta} <- handle_object_creation(meta[:object_data], meta) do
  77       {:ok, notifications} = Notification.create_notifications(activity, do_send: false)
  78
  79       meta =
  80         meta
  81         |> add_notifications(notifications)
  82
  83       {:ok, activity, meta}
  84     else
  85       e -> Repo.rollback(e)
  86     end
  87   end
  88
  89   # Tasks this handles:
  90   # - Add announce to object
  91   # - Set up notification
  92   # - Stream out the announce
  93   def handle(%{data: %{"type" => "Announce"}} = object, meta) do
  94     announced_object = Object.get_by_ap_id(object.data["object"])
  95     user = User.get_cached_by_ap_id(object.data["actor"])
  96
  97     Utils.add_announce_to_object(object, announced_object)
  98
  99     if !User.is_internal_user?(user) do
 100       Notification.create_notifications(object)
 101
 102       object
 103       |> Topics.get_activity_topics()
 104       |> Streamer.stream(object)
 105     end
 106
 107     {:ok, object, meta}
 108   end
 109
 110   def handle(%{data: %{"type" => "Undo", "object" => undone_object}} = object, meta) do
 111     with undone_object <- Activity.get_by_ap_id(undone_object),
 112          :ok <- handle_undoing(undone_object) do
 113       {:ok, object, meta}
 114     end
 115   end
 116
 117   # Tasks this handles:
 118   # - Add reaction to object
 119   # - Set up notification
 120   def handle(%{data: %{"type" => "EmojiReact"}} = object, meta) do
 121     reacted_object = Object.get_by_ap_id(object.data["object"])
 122     Utils.add_emoji_reaction_to_object(object, reacted_object)
 123
 124     Notification.create_notifications(object)
 125
 126     {:ok, object, meta}
 127   end
 128
 129   # Tasks this handles:
 130   # - Delete and unpins the create activity
 131   # - Replace object with Tombstone
 132   # - Set up notification
 133   # - Reduce the user note count
 134   # - Reduce the reply count
 135   # - Stream out the activity
 136   def handle(%{data: %{"type" => "Delete", "object" => deleted_object}} = object, meta) do
 137     deleted_object =
 138       Object.normalize(deleted_object, false) || User.get_cached_by_ap_id(deleted_object)
 139
 140     result =
 141       case deleted_object do
 142         %Object{} ->
 143           with {:ok, deleted_object, activity} <- Object.delete(deleted_object),
 144                %User{} = user <- User.get_cached_by_ap_id(deleted_object.data["actor"]) do
 145             User.remove_pinnned_activity(user, activity)
 146
 147             {:ok, user} = ActivityPub.decrease_note_count_if_public(user, deleted_object)
 148
 149             if in_reply_to = deleted_object.data["inReplyTo"] do
 150               Object.decrease_replies_count(in_reply_to)
 151             end
 152
 153             MessageReference.delete_for_object(deleted_object)
 154
 155             ActivityPub.stream_out(object)
 156             ActivityPub.stream_out_participations(deleted_object, user)
 157             :ok
 158           end
 159
 160         %User{} ->
 161           with {:ok, _} <- User.delete(deleted_object) do
 162             :ok
 163           end
 164       end
 165
 166     if result == :ok do
 167       Notification.create_notifications(object)
 168       {:ok, object, meta}
 169     else
 170       {:error, result}
 171     end
 172   end
 173
 174   # Nothing to do
 175   def handle(object, meta) do
 176     {:ok, object, meta}
 177   end
 178
 179   def handle_object_creation(%{"type" => "ChatMessage"} = object, meta) do
 180     with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
 181       actor = User.get_cached_by_ap_id(object.data["actor"])
 182       recipient = User.get_cached_by_ap_id(hd(object.data["to"]))
 183
 184       streamables =
 185         [[actor, recipient], [recipient, actor]]
 186         |> Enum.map(fn [user, other_user] ->
 187           if user.local do
 188             {:ok, chat} = Chat.bump_or_create(user.id, other_user.ap_id)
 189             {:ok, cm_ref} = MessageReference.create(chat, object, user.ap_id != actor.ap_id)
 190
 191             {
 192               ["user", "user:pleroma_chat"],
 193               {user, %{cm_ref | chat: chat, object: object}}
 194             }
 195           end
 196         end)
 197         |> Enum.filter(& &1)
 198
 199       meta =
 200         meta
 201         |> add_streamables(streamables)
 202
 203       {:ok, object, meta}
 204     end
 205   end
 206
 207   # Nothing to do
 208   def handle_object_creation(object) do
 209     {:ok, object}
 210   end
 211
 212   defp undo_like(nil, object), do: delete_object(object)
 213
 214   defp undo_like(%Object{} = liked_object, object) do
 215     with {:ok, _} <- Utils.remove_like_from_object(object, liked_object) do
 216       delete_object(object)
 217     end
 218   end
 219
 220   def handle_undoing(%{data: %{"type" => "Like"}} = object) do
 221     object.data["object"]
 222     |> Object.get_by_ap_id()
 223     |> undo_like(object)
 224   end
 225
 226   def handle_undoing(%{data: %{"type" => "EmojiReact"}} = object) do
 227     with %Object{} = reacted_object <- Object.get_by_ap_id(object.data["object"]),
 228          {:ok, _} <- Utils.remove_emoji_reaction_from_object(object, reacted_object),
 229          {:ok, _} <- Repo.delete(object) do
 230       :ok
 231     end
 232   end
 233
 234   def handle_undoing(%{data: %{"type" => "Announce"}} = object) do
 235     with %Object{} = liked_object <- Object.get_by_ap_id(object.data["object"]),
 236          {:ok, _} <- Utils.remove_announce_from_object(object, liked_object),
 237          {:ok, _} <- Repo.delete(object) do
 238       :ok
 239     end
 240   end
 241
 242   def handle_undoing(
 243         %{data: %{"type" => "Block", "actor" => blocker, "object" => blocked}} = object
 244       ) do
 245     with %User{} = blocker <- User.get_cached_by_ap_id(blocker),
 246          %User{} = blocked <- User.get_cached_by_ap_id(blocked),
 247          {:ok, _} <- User.unblock(blocker, blocked),
 248          {:ok, _} <- Repo.delete(object) do
 249       :ok
 250     end
 251   end
 252
 253   def handle_undoing(object), do: {:error, ["don't know how to handle", object]}
 254
 255   @spec delete_object(Object.t()) :: :ok | {:error, Ecto.Changeset.t()}
 256   defp delete_object(object) do
 257     with {:ok, _} <- Repo.delete(object), do: :ok
 258   end
 259
 260   defp send_notifications(meta) do
 261     Keyword.get(meta, :notifications, [])
 262     |> Enum.each(fn notification ->
 263       Streamer.stream(["user", "user:notification"], notification)
 264       Push.send(notification)
 265     end)
 266
 267     meta
 268   end
 269
 270   defp send_streamables(meta) do
 271     Keyword.get(meta, :streamables, [])
 272     |> Enum.each(fn {topics, items} ->
 273       Streamer.stream(topics, items)
 274     end)
 275
 276     meta
 277   end
 278
 279   defp add_streamables(meta, streamables) do
 280     existing = Keyword.get(meta, :streamables, [])
 281
 282     meta
 283     |> Keyword.put(:streamables, streamables ++ existing)
 284   end
 285
 286   defp add_notifications(meta, notifications) do
 287     existing = Keyword.get(meta, :notifications, [])
 288
 289     meta
 290     |> Keyword.put(:notifications, notifications ++ existing)
 291   end
 292
 293   def handle_after_transaction(meta) do
 294     meta
 295     |> send_notifications()
 296     |> send_streamables()
 297   end
 298 end