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.FollowingRelationship
  13   alias Pleroma.Notification
  14   alias Pleroma.Object
  15   alias Pleroma.Repo
  16   alias Pleroma.User
  17   alias Pleroma.Web.ActivityPub.ActivityPub
  18   alias Pleroma.Web.ActivityPub.Pipeline
  19   alias Pleroma.Web.ActivityPub.Utils
  20   alias Pleroma.Web.Push
  21   alias Pleroma.Web.Streamer
  22
  23   def handle(object, meta \\ [])
  24
  25   # Tasks this handle
  26   # - Follows if possible
  27   # - Sends a notification
  28   # - Generates accept or reject if appropriate
  29   def handle(
  30         %{
  31           data: %{
  32             "id" => follow_id,
  33             "type" => "Follow",
  34             "object" => followed_user,
  35             "actor" => following_user
  36           }
  37         } = object,
  38         meta
  39       ) do
  40     with %User{} = follower <- User.get_cached_by_ap_id(following_user),
  41          %User{} = followed <- User.get_cached_by_ap_id(followed_user),
  42          {_, {:ok, _}, _, _} <-
  43            {:following, User.follow(follower, followed, :follow_pending), follower, followed} do
  44       if followed.local && !followed.locked do
  45         Utils.update_follow_state_for_all(object, "accept")
  46         FollowingRelationship.update(follower, followed, :follow_accept)
  47         User.update_follower_count(followed)
  48         User.update_following_count(follower)
  49
  50         %{
  51           to: [following_user],
  52           actor: followed,
  53           object: follow_id,
  54           local: true
  55         }
  56         |> ActivityPub.accept()
  57       end
  58     else
  59       {:following, {:error, _}, follower, followed} ->
  60         Utils.update_follow_state_for_all(object, "reject")
  61         FollowingRelationship.update(follower, followed, :follow_reject)
  62
  63         if followed.local do
  64           %{
  65             to: [follower.ap_id],
  66             actor: followed,
  67             object: follow_id,
  68             local: true
  69           }
  70           |> ActivityPub.reject()
  71         end
  72
  73       _ ->
  74         nil
  75     end
  76
  77     {:ok, notifications} = Notification.create_notifications(object, do_send: false)
  78
  79     meta =
  80       meta
  81       |> add_notifications(notifications)
  82
  83     updated_object = Activity.get_by_ap_id(follow_id)
  84
  85     {:ok, updated_object, meta}
  86   end
  87
  88   # Tasks this handles:
  89   # - Unfollow and block
  90   def handle(
  91         %{data: %{"type" => "Block", "object" => blocked_user, "actor" => blocking_user}} =
  92           object,
  93         meta
  94       ) do
  95     with %User{} = blocker <- User.get_cached_by_ap_id(blocking_user),
  96          %User{} = blocked <- User.get_cached_by_ap_id(blocked_user) do
  97       User.block(blocker, blocked)
  98     end
  99
 100     {:ok, object, meta}
 101   end
 102
 103   # Tasks this handles:
 104   # - Update the user
 105   #
 106   # For a local user, we also get a changeset with the full information, so we
 107   # can update non-federating, non-activitypub settings as well.
 108   def handle(%{data: %{"type" => "Update", "object" => updated_object}} = object, meta) do
 109     if changeset = Keyword.get(meta, :user_update_changeset) do
 110       changeset
 111       |> User.update_and_set_cache()
 112     else
 113       {:ok, new_user_data} = ActivityPub.user_data_from_user_object(updated_object)
 114
 115       User.get_by_ap_id(updated_object["id"])
 116       |> User.remote_user_changeset(new_user_data)
 117       |> User.update_and_set_cache()
 118     end
 119
 120     {:ok, object, meta}
 121   end
 122
 123   # Tasks this handles:
 124   # - Add like to object
 125   # - Set up notification
 126   def handle(%{data: %{"type" => "Like"}} = object, meta) do
 127     liked_object = Object.get_by_ap_id(object.data["object"])
 128     Utils.add_like_to_object(object, liked_object)
 129
 130     Notification.create_notifications(object)
 131
 132     {:ok, object, meta}
 133   end
 134
 135   # Tasks this handles
 136   # - Actually create object
 137   # - Rollback if we couldn't create it
 138   # - Set up notifications
 139   def handle(%{data: %{"type" => "Create"}} = activity, meta) do
 140     with {:ok, _object, meta} <- handle_object_creation(meta[:object_data], meta) do
 141       {:ok, notifications} = Notification.create_notifications(activity, do_send: false)
 142
 143       meta =
 144         meta
 145         |> add_notifications(notifications)
 146
 147       {:ok, activity, meta}
 148     else
 149       e -> Repo.rollback(e)
 150     end
 151   end
 152
 153   # Tasks this handles:
 154   # - Add announce to object
 155   # - Set up notification
 156   # - Stream out the announce
 157   def handle(%{data: %{"type" => "Announce"}} = object, meta) do
 158     announced_object = Object.get_by_ap_id(object.data["object"])
 159     user = User.get_cached_by_ap_id(object.data["actor"])
 160
 161     Utils.add_announce_to_object(object, announced_object)
 162
 163     if !User.is_internal_user?(user) do
 164       Notification.create_notifications(object)
 165
 166       object
 167       |> Topics.get_activity_topics()
 168       |> Streamer.stream(object)
 169     end
 170
 171     {:ok, object, meta}
 172   end
 173
 174   def handle(%{data: %{"type" => "Undo", "object" => undone_object}} = object, meta) do
 175     with undone_object <- Activity.get_by_ap_id(undone_object),
 176          :ok <- handle_undoing(undone_object) do
 177       meta =
 178         meta
 179         |> Keyword.put(:embedded_object, undone_object.data)
 180
 181       {:ok, object, meta}
 182     end
 183   end
 184
 185   # Tasks this handles:
 186   # - Add reaction to object
 187   # - Set up notification
 188   def handle(%{data: %{"type" => "EmojiReact"}} = object, meta) do
 189     reacted_object = Object.get_by_ap_id(object.data["object"])
 190     Utils.add_emoji_reaction_to_object(object, reacted_object)
 191
 192     Notification.create_notifications(object)
 193
 194     {:ok, object, meta}
 195   end
 196
 197   # Tasks this handles:
 198   # - Delete and unpins the create activity
 199   # - Replace object with Tombstone
 200   # - Set up notification
 201   # - Reduce the user note count
 202   # - Reduce the reply count
 203   # - Stream out the activity
 204   def handle(%{data: %{"type" => "Delete", "object" => deleted_object}} = object, meta) do
 205     deleted_object =
 206       Object.normalize(deleted_object, false) || User.get_cached_by_ap_id(deleted_object)
 207
 208     result =
 209       case deleted_object do
 210         %Object{} ->
 211           with {:ok, deleted_object, activity} <- Object.delete(deleted_object),
 212                %User{} = user <- User.get_cached_by_ap_id(deleted_object.data["actor"]) do
 213             User.remove_pinnned_activity(user, activity)
 214
 215             {:ok, user} = ActivityPub.decrease_note_count_if_public(user, deleted_object)
 216
 217             if in_reply_to = deleted_object.data["inReplyTo"] do
 218               Object.decrease_replies_count(in_reply_to)
 219             end
 220
 221             MessageReference.delete_for_object(deleted_object)
 222
 223             ActivityPub.stream_out(object)
 224             ActivityPub.stream_out_participations(deleted_object, user)
 225             :ok
 226           end
 227
 228         %User{} ->
 229           with {:ok, _} <- User.delete(deleted_object) do
 230             :ok
 231           end
 232       end
 233
 234     if result == :ok do
 235       Notification.create_notifications(object)
 236       {:ok, object, meta}
 237     else
 238       {:error, result}
 239     end
 240   end
 241
 242   # Nothing to do
 243   def handle(object, meta) do
 244     {:ok, object, meta}
 245   end
 246
 247   def handle_object_creation(%{"type" => "ChatMessage"} = object, meta) do
 248     with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
 249       actor = User.get_cached_by_ap_id(object.data["actor"])
 250       recipient = User.get_cached_by_ap_id(hd(object.data["to"]))
 251
 252       streamables =
 253         [[actor, recipient], [recipient, actor]]
 254         |> Enum.map(fn [user, other_user] ->
 255           if user.local do
 256             {:ok, chat} = Chat.bump_or_create(user.id, other_user.ap_id)
 257             {:ok, cm_ref} = MessageReference.create(chat, object, user.ap_id != actor.ap_id)
 258
 259             {
 260               ["user", "user:pleroma_chat"],
 261               {user, %{cm_ref | chat: chat, object: object}}
 262             }
 263           end
 264         end)
 265         |> Enum.filter(& &1)
 266
 267       meta =
 268         meta
 269         |> add_streamables(streamables)
 270
 271       {:ok, object, meta}
 272     end
 273   end
 274
 275   # Nothing to do
 276   def handle_object_creation(object) do
 277     {:ok, object}
 278   end
 279
 280   defp undo_like(nil, object), do: delete_object(object)
 281
 282   defp undo_like(%Object{} = liked_object, object) do
 283     with {:ok, _} <- Utils.remove_like_from_object(object, liked_object) do
 284       delete_object(object)
 285     end
 286   end
 287
 288   def handle_undoing(%{data: %{"type" => "Like"}} = object) do
 289     object.data["object"]
 290     |> Object.get_by_ap_id()
 291     |> undo_like(object)
 292   end
 293
 294   def handle_undoing(%{data: %{"type" => "EmojiReact"}} = object) do
 295     with %Object{} = reacted_object <- Object.get_by_ap_id(object.data["object"]),
 296          {:ok, _} <- Utils.remove_emoji_reaction_from_object(object, reacted_object),
 297          {:ok, _} <- Repo.delete(object) do
 298       :ok
 299     end
 300   end
 301
 302   def handle_undoing(%{data: %{"type" => "Announce"}} = object) do
 303     with %Object{} = liked_object <- Object.get_by_ap_id(object.data["object"]),
 304          {:ok, _} <- Utils.remove_announce_from_object(object, liked_object),
 305          {:ok, _} <- Repo.delete(object) do
 306       :ok
 307     end
 308   end
 309
 310   def handle_undoing(
 311         %{data: %{"type" => "Block", "actor" => blocker, "object" => blocked}} = object
 312       ) do
 313     with %User{} = blocker <- User.get_cached_by_ap_id(blocker),
 314          %User{} = blocked <- User.get_cached_by_ap_id(blocked),
 315          {:ok, _} <- User.unblock(blocker, blocked),
 316          {:ok, _} <- Repo.delete(object) do
 317       :ok
 318     end
 319   end
 320
 321   def handle_undoing(object), do: {:error, ["don't know how to handle", object]}
 322
 323   @spec delete_object(Object.t()) :: :ok | {:error, Ecto.Changeset.t()}
 324   defp delete_object(object) do
 325     with {:ok, _} <- Repo.delete(object), do: :ok
 326   end
 327
 328   defp send_notifications(meta) do
 329     Keyword.get(meta, :notifications, [])
 330     |> Enum.each(fn notification ->
 331       Streamer.stream(["user", "user:notification"], notification)
 332       Push.send(notification)
 333     end)
 334
 335     meta
 336   end
 337
 338   defp send_streamables(meta) do
 339     Keyword.get(meta, :streamables, [])
 340     |> Enum.each(fn {topics, items} ->
 341       Streamer.stream(topics, items)
 342     end)
 343
 344     meta
 345   end
 346
 347   defp add_streamables(meta, streamables) do
 348     existing = Keyword.get(meta, :streamables, [])
 349
 350     meta
 351     |> Keyword.put(:streamables, streamables ++ existing)
 352   end
 353
 354   defp add_notifications(meta, notifications) do
 355     existing = Keyword.get(meta, :notifications, [])
 356
 357     meta
 358     |> Keyword.put(:notifications, notifications ++ existing)
 359   end
 360
 361   def handle_after_transaction(meta) do
 362     meta
 363     |> send_notifications()
 364     |> send_streamables()
 365   end
 366 end