Pipeline Ingestion: Note

[akkoma] / lib / pleroma / web / activity_pub / side_effects.ex

diff --git a/lib/pleroma/web/activity_pub/side_effects.ex b/lib/pleroma/web/activity_pub/side_effects.ex
index 8556fca1db8cc5121efcb6f97f5942842970767c..3234b9e431e67b2b95a877bf868b8a4fbf1d2422 100644 (file)
--- a/lib/pleroma/web/activity_pub/side_effects.ex
+++ b/lib/pleroma/web/activity_pub/side_effects.ex
@@ -1,5 +1,5 @@
 # Pleroma: A lightweight social networking server
-# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.SideEffects do
@@ -27,11 +27,19 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
 
   require Logger
 
+  @cachex Pleroma.Config.get([:cachex, :provider], Cachex)
+  @ap_streamer Pleroma.Config.get([:side_effects, :ap_streamer], ActivityPub)
+  @logger Pleroma.Config.get([:side_effects, :logger], Logger)
+
+  @behaviour Pleroma.Web.ActivityPub.SideEffects.Handling
+
+  @impl true
   def handle(object, meta \\ [])
 
   # Task this handles
   # - Follows
   # - Sends a notification
+  @impl true
   def handle(
         %{
           data: %{
@@ -59,6 +67,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   # - Rejects all existing follow activities for this person
   # - Updates the follow state
   # - Dismisses notification
+  @impl true
   def handle(
         %{
           data: %{
@@ -85,6 +94,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   # - Follows if possible
   # - Sends a notification
   # - Generates accept or reject if appropriate
+  @impl true
   def handle(
         %{
           data: %{
@@ -126,6 +136,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
 
   # Tasks this handles:
   # - Unfollow and block
+  @impl true
   def handle(
         %{data: %{"type" => "Block", "object" => blocked_user, "actor" => blocking_user}} =
           object,
@@ -144,6 +155,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   #
   # For a local user, we also get a changeset with the full information, so we
   # can update non-federating, non-activitypub settings as well.
+  @impl true
   def handle(%{data: %{"type" => "Update", "object" => updated_object}} = object, meta) do
     if changeset = Keyword.get(meta, :user_update_changeset) do
       changeset
@@ -162,6 +174,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   # Tasks this handles:
   # - Add like to object
   # - Set up notification
+  @impl true
   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)
@@ -179,6 +192,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   # - Increase replies count
   # - Set up ActivityExpiration
   # - Set up notifications
+  @impl true
   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
@@ -189,6 +203,19 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
         Object.increase_replies_count(in_reply_to)
       end
 
+      reply_depth = (meta[:depth] || 0) + 1
+
+      # FIXME: Force inReplyTo to replies
+      if Pleroma.Web.Federator.allowed_thread_distance?(reply_depth) and
+           object.data["replies"] != nil do
+        for reply_id <- object.data["replies"] do
+          Pleroma.Workers.RemoteFetcherWorker.enqueue("fetch_remote", %{
+            "id" => reply_id,
+            "depth" => reply_depth
+          })
+        end
+      end
+
       ConcurrentLimiter.limit(Pleroma.Web.RichMedia.Helpers, fn ->
         Task.start(fn -> Pleroma.Web.RichMedia.Helpers.fetch_data_for_activity(activity) end)
       end)
@@ -207,6 +234,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   # - Add announce to object
   # - Set up notification
   # - Stream out the announce
+  @impl true
   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"])
@@ -224,6 +252,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
     {:ok, object, meta}
   end
 
+  @impl true
   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
@@ -234,6 +263,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   # Tasks this handles:
   # - Add reaction to object
   # - Set up notification
+  @impl true
   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)
@@ -250,9 +280,10 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   # - Reduce the user note count
   # - Reduce the reply count
   # - Stream out the activity
+  @impl true
   def handle(%{data: %{"type" => "Delete", "object" => deleted_object}} = object, meta) do
     deleted_object =
-      Object.normalize(deleted_object, false) ||
+      Object.normalize(deleted_object, fetch: false) ||
         User.get_cached_by_ap_id(deleted_object)
 
     result =
@@ -271,12 +302,12 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
 
             MessageReference.delete_for_object(deleted_object)
 
-            ActivityPub.stream_out(object)
-            ActivityPub.stream_out_participations(deleted_object, user)
+            @ap_streamer.stream_out(object)
+            @ap_streamer.stream_out_participations(deleted_object, user)
             :ok
           else
             {:actor, _} ->
-              Logger.error("The object doesn't have an actor: #{inspect(deleted_object)}")
+              @logger.error("The object doesn't have an actor: #{inspect(deleted_object)}")
               :no_object_actor
           end
 
@@ -295,6 +326,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   end
 
   # Nothing to do
+  @impl true
   def handle(object, meta) do
     {:ok, object, meta}
   end
@@ -312,7 +344,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects 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)
 
-            Cachex.put(
+            @cachex.put(
               :chat_message_id_idempotency_key_cache,
               cm_ref.id,
               meta[:idempotency_key]
@@ -347,7 +379,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
   end
 
   def handle_object_creation(%{"type" => objtype} = object, meta)
-      when objtype in ~w[Audio Video Question Event Article] do
+      when objtype in ~w[Audio Video Question Event Article Note] do
     with {:ok, object, meta} <- Pipeline.common_pipeline(object, meta) do
       {:ok, object, meta}
     end
@@ -439,6 +471,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
     |> Keyword.put(:notifications, notifications ++ existing)
   end
 
+  @impl true
   def handle_after_transaction(meta) do
     meta
     |> send_notifications()