use proper workers for fetching pins instead of an ad-hoc task (#788)

Reviewed-on: https://akkoma.dev/AkkomaGang/akkoma/pulls/788
Co-authored-by: Floatingghost <hannah@coffee-and-dreams.uk>
Co-committed-by: Floatingghost <hannah@coffee-and-dreams.uk>

lib/mix/pleroma.ex

@@ -16,7 +16,7 @@ defmodule Mix.Pleroma do
     :fast_html,
     :oban
   ]
-  @cachex_children ["object", "user", "scrubber", "web_resp"]
+  @cachex_children ["object", "user", "scrubber", "web_resp", "http_backoff"]
   @doc "Common functions to be reused in mix tasks"
   def start_pleroma do
     Pleroma.Config.Holder.save_default()

lib/mix/tasks/pleroma/diagnostics.ex

@@ -17,6 +17,13 @@ defmodule Mix.Tasks.Pleroma.Diagnostics do
     |> IO.inspect()
   end
 
+  def run(["fetch_object", url]) do
+    start_pleroma()
+
+    Pleroma.Object.Fetcher.fetch_object_from_id(url)
+    |> IO.inspect()
+  end
+
   def run(["home_timeline", nickname]) do
     start_pleroma()
     user = Repo.get_by!(User, nickname: nickname)

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1824,18 +1824,19 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     end
   end
 
-  def pinned_fetch_task(nil), do: nil
+  def enqueue_pin_fetches(%{pinned_objects: pins}) do
+    # enqueue a task to fetch all pinned objects
+    Enum.each(pins, fn {ap_id, _} ->
+      if is_nil(Object.get_cached_by_ap_id(ap_id)) do
+        Pleroma.Workers.RemoteFetcherWorker.enqueue("fetch_remote", %{
+          "id" => ap_id,
+          "depth" => 1
+        })
+      end
+    end)
+  end
 
-  def pinned_fetch_task(%{pinned_objects: pins}) do
+  def enqueue_pin_fetches(_), do: nil
-    if Enum.all?(pins, fn {ap_id, _} ->
-         Object.get_cached_by_ap_id(ap_id) ||
-           match?({:ok, _object}, Fetcher.fetch_object_from_id(ap_id))
-       end) do
-      :ok
-    else
-      :error
-    end
-  end
 
   def make_user_from_ap_id(ap_id, additional \\ []) do
     user = User.get_cached_by_ap_id(ap_id)
@@ -1844,7 +1845,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
       Transmogrifier.upgrade_user_from_ap_id(ap_id)
     else
       with {:ok, data} <- fetch_and_prepare_user_from_ap_id(ap_id, additional) do
-        {:ok, _pid} = Task.start(fn -> pinned_fetch_task(data) end)
+        enqueue_pin_fetches(data)
 
         user =
           if data.ap_id != ap_id do

lib/pleroma/web/activity_pub/transmogrifier.ex

@@ -1034,7 +1034,7 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
     with %User{local: false} = user <- User.get_cached_by_ap_id(ap_id),
          {:ok, data} <- ActivityPub.fetch_and_prepare_user_from_ap_id(ap_id),
          {:ok, user} <- update_user(user, data) do
-      {:ok, _pid} = Task.start(fn -> ActivityPub.pinned_fetch_task(user) end)
+      ActivityPub.enqueue_pin_fetches(user)
       TransmogrifierWorker.enqueue("user_upgrade", %{"user_id" => user.id})
       {:ok, user}
     else

test/pleroma/web/activity_pub/activity_pub_test.exs

@@ -325,9 +325,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
             body: featured_data,
             headers: [{"content-type", "application/activity+json"}]
           }
-      end)
 
-      Tesla.Mock.mock_global(fn
         %{
           method: :get,
           url: ^object_url
@@ -340,7 +338,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
       end)
 
       {:ok, user} = ActivityPub.make_user_from_ap_id(ap_id)
-      Process.sleep(50)
+      # wait for oban
+      Pleroma.Tests.ObanHelpers.perform_all()
 
       assert user.featured_address == featured_url
       assert Map.has_key?(user.pinned_objects, object_url)