[#1505] Background fetching of incoming activities' `replies` collections.

config/config.exs

@@ -501,6 +501,7 @@
     transmogrifier: 20,
     scheduled_activities: 10,
     background: 5,
+    remote_fetcher: 2,
     attachments_cleanup: 5
   ]
 

lib/pleroma/web/activity_pub/transmogrifier.ex

@@ -424,7 +424,13 @@ def handle_incoming(
           ])
       }
 
-      ActivityPub.create(params)
+      with {:ok, created_activity} <- ActivityPub.create(params) do
+        for reply_id <- replies(object) do
+          Pleroma.Workers.RemoteFetcherWorker.enqueue("fetch_remote", %{"id" => reply_id})
+        end
+
+        {:ok, created_activity}
+      end
     else
       %Activity{} = activity -> {:ok, activity}
       _e -> :error
@@ -946,6 +952,13 @@ defp set_replies(obj, replies_uris) do
     Map.merge(obj, %{"replies" => replies_collection})
   end
 
+  def replies(%{"replies" => replies = %{}}) do
+    replies = with %{} <- replies["first"], do: replies["first"], else: (_ -> replies)
+    replies["items"] || []
+  end
+
+  def replies(_), do: []
+
   # Prepares the object of an outgoing create activity.
   def prepare_object(object) do
     object

lib/pleroma/workers/remote_fetcher_worker.ex

@@ -0,0 +1,20 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.RemoteFetcherWorker do
+  alias Pleroma.Object.Fetcher
+
+  use Pleroma.Workers.WorkerHelper, queue: "remote_fetcher"
+
+  @impl Oban.Worker
+  def perform(
+        %{
+          "op" => "fetch_remote",
+          "id" => id
+        },
+        _job
+      ) do
+    Fetcher.fetch_object_from_id!(id)
+  end
+end

test/support/oban_helpers.ex

@@ -9,6 +9,10 @@ defmodule Pleroma.Tests.ObanHelpers do
 
   alias Pleroma.Repo
 
+  def wipe_all do
+    Repo.delete_all(Oban.Job)
+  end
+
   def perform_all do
     Oban.Job
     |> Repo.all()

test/web/activity_pub/transmogrifier_test.exs

@@ -3,7 +3,9 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.TransmogrifierTest do
+  use Oban.Testing, repo: Pleroma.Repo
   use Pleroma.DataCase
+
   alias Pleroma.Activity
   alias Pleroma.Object
   alias Pleroma.Object.Fetcher
@@ -1329,6 +1331,52 @@ test "it accepts Move activities" do
     end
   end
 
+  describe "handle_incoming:`replies` handling" do
+    setup do
+      data =
+        File.read!("test/fixtures/mastodon-post-activity.json")
+        |> Poison.decode!()
+
+      items = ["https://shitposter.club/notice/2827873", "https://shitposter.club/notice/7387606"]
+      collection = %{"items" => items}
+      %{data: data, items: items, collection: collection}
+    end
+
+    test "it schedules background fetching of wrapped `replies` collection items", %{
+      data: data,
+      items: items,
+      collection: collection
+    } do
+      replies = %{"first" => collection}
+
+      object = Map.put(data["object"], "replies", replies)
+      data = Map.put(data, "object", object)
+      {:ok, _activity} = Transmogrifier.handle_incoming(data)
+
+      for id <- items do
+        job_args = %{"op" => "fetch_remote", "id" => id}
+        assert_enqueued(worker: Pleroma.Workers.RemoteFetcherWorker, args: job_args)
+      end
+    end
+
+    test "it schedules background fetching of unwrapped `replies` collection items", %{
+      data: data,
+      items: items,
+      collection: collection
+    } do
+      replies = collection
+
+      object = Map.put(data["object"], "replies", replies)
+      data = Map.put(data, "object", object)
+      {:ok, _activity} = Transmogrifier.handle_incoming(data)
+
+      for id <- items do
+        job_args = %{"op" => "fetch_remote", "id" => id}
+        assert_enqueued(worker: Pleroma.Workers.RemoteFetcherWorker, args: job_args)
+      end
+    end
+  end
+
   describe "prepare outgoing" do
     test "it inlines private announced objects" do
       user = insert(:user)