add a job queue

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -711,20 +711,18 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
     public = is_public?(activity)
 
-    remote_inboxes =
-      (Pleroma.Web.Salmon.remote_users(activity) ++ followers)
-      |> Enum.filter(fn user -> User.ap_enabled?(user) end)
-      |> Enum.map(fn %{info: %{source_data: data}} ->
-        (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
-      end)
-      |> Enum.uniq()
-      |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
-
     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
     json = Jason.encode!(data)
 
-    Enum.each(remote_inboxes, fn inbox ->
-      Federator.enqueue(:publish_single_ap, %{
+    (Pleroma.Web.Salmon.remote_users(activity) ++ followers)
+    |> Enum.filter(fn user -> User.ap_enabled?(user) end)
+    |> Enum.map(fn %{info: %{source_data: data}} ->
+      (is_map(data["endpoints"]) && Map.get(data["endpoints"], "sharedInbox")) || data["inbox"]
+    end)
+    |> Enum.uniq()
+    |> Enum.filter(fn inbox -> should_federate?(inbox, public) end)
+    |> Enum.each(fn inbox ->
+      Federator.publish_single_ap(%{
         inbox: inbox,
         json: json,
         actor: actor,

lib/pleroma/web/activity_pub/activity_pub_controller.ex

@@ -150,13 +150,13 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
     with %User{} = user <- User.get_cached_by_nickname(nickname),
          true <- Utils.recipient_in_message(user.ap_id, params),
          params <- Utils.maybe_splice_recipient(user.ap_id, params) do
-      Federator.enqueue(:incoming_ap_doc, params)
+      Federator.incoming_ap_doc(params)
       json(conn, "ok")
     end
   end
 
   def inbox(%{assigns: %{valid_signature: true}} = conn, params) do
-    Federator.enqueue(:incoming_ap_doc, params)
+    Federator.incoming_ap_doc(params)
     json(conn, "ok")
   end
 

lib/pleroma/web/activity_pub/utils.ex

@@ -156,7 +156,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
         _ -> 5
       end
 
-    Pleroma.Web.Federator.enqueue(:publish, activity, priority)
+    Pleroma.Web.Federator.publish(activity, priority)
     :ok
   end
 

lib/pleroma/web/federator/federator.ex

@@ -3,9 +3,9 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Federator do
-  use GenServer
   alias Pleroma.User
   alias Pleroma.Activity
+  alias Pleroma.Jobs
   alias Pleroma.Web.{WebFinger, Websub}
   alias Pleroma.Web.Federator.RetryQueue
   alias Pleroma.Web.ActivityPub.ActivityPub
@@ -18,39 +18,60 @@ defmodule Pleroma.Web.Federator do
   @websub Application.get_env(:pleroma, :websub)
   @ostatus Application.get_env(:pleroma, :ostatus)
 
-  def init(args) do
-    {:ok, args}
+  def init() do
+    # 1 minute
+    Process.sleep(1000 * 60 * 1)
+    refresh_subscriptions()
   end
 
-  def start_link do
-    spawn(fn ->
-      # 1 minute
-      Process.sleep(1000 * 60 * 1)
-      enqueue(:refresh_subscriptions, nil)
-    end)
+  # Client API
 
-    GenServer.start_link(
-      __MODULE__,
-      %{
-        in: {:sets.new(), []},
-        out: {:sets.new(), []}
-      },
-      name: __MODULE__
-    )
+  def incoming_doc(doc) do
+    Jobs.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc])
   end
 
-  def handle(:refresh_subscriptions, _) do
+  def incoming_ap_doc(params) do
+    Jobs.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params])
+  end
+
+  def publish(activity, priority \\ 1) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:publish, activity], priority)
+  end
+
+  def publish_single_ap(params) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:publish_single_ap, params])
+  end
+
+  def publish_single_websub(websub) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:publish_single_websub, websub])
+  end
+
+  def verify_websub(websub) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:verify_websub, websub])
+  end
+
+  def request_subscription(sub) do
+    Jobs.enqueue(:federator_out, __MODULE__, [:request_subscription, sub])
+  end
+
+  def refresh_subscriptions() do
+    Jobs.enqueue(:federator_out, __MODULE__, [:refresh_subscriptions])
+  end
+
+  # Job Worker Callbacks
+
+  def perform(:refresh_subscriptions) do
     Logger.debug("Federator running refresh subscriptions")
     Websub.refresh_subscriptions()
 
     spawn(fn ->
       # 6 hours
       Process.sleep(1000 * 60 * 60 * 6)
-      enqueue(:refresh_subscriptions, nil)
+      refresh_subscriptions()
     end)
   end
 
-  def handle(:request_subscription, websub) do
+  def perform(:request_subscription, websub) do
     Logger.debug("Refreshing #{websub.topic}")
 
     with {:ok, websub} <- Websub.request_subscription(websub) do
@@ -60,7 +81,7 @@ defmodule Pleroma.Web.Federator do
     end
   end
 
-  def handle(:publish, activity) do
+  def perform(:publish, activity) do
     Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)
 
     with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do
@@ -86,7 +107,7 @@ defmodule Pleroma.Web.Federator do
     end
   end
 
-  def handle(:verify_websub, websub) do
+  def perform(:verify_websub, websub) do
     Logger.debug(fn ->
       "Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})"
     end)
@@ -94,12 +115,12 @@ defmodule Pleroma.Web.Federator do
     @websub.verify(websub)
   end
 
-  def handle(:incoming_doc, doc) do
+  def perform(:incoming_doc, doc) do
     Logger.info("Got document, trying to parse")
     @ostatus.handle_incoming(doc)
   end
 
-  def handle(:incoming_ap_doc, params) do
+  def perform(:incoming_ap_doc, params) do
     Logger.info("Handling incoming AP activity")
 
     params = Utils.normalize_params(params)
@@ -124,7 +145,7 @@ defmodule Pleroma.Web.Federator do
     end
   end
 
-  def handle(:publish_single_ap, params) do
+  def perform(:publish_single_ap, params) do
     case ActivityPub.publish_one(params) do
       {:ok, _} ->
         :ok
@@ -134,7 +155,7 @@ defmodule Pleroma.Web.Federator do
     end
   end
 
-  def handle(
+  def perform(
         :publish_single_websub,
         %{xml: _xml, topic: _topic, callback: _callback, secret: _secret} = params
       ) do
@@ -147,75 +168,11 @@ defmodule Pleroma.Web.Federator do
     end
   end
 
-  def handle(type, _) do
+  def perform(type, _) do
     Logger.debug(fn -> "Unknown task: #{type}" end)
     {:error, "Don't know what to do with this"}
   end
 
-  if Mix.env() == :test do
-    def enqueue(type, payload, _priority \\ 1) do
-      if Pleroma.Config.get([:instance, :federating]) do
-        handle(type, payload)
-      end
-    end
-  else
-    def enqueue(type, payload, priority \\ 1) do
-      if Pleroma.Config.get([:instance, :federating]) do
-        GenServer.cast(__MODULE__, {:enqueue, type, payload, priority})
-      end
-    end
-  end
-
-  def maybe_start_job(running_jobs, queue) do
-    if :sets.size(running_jobs) < Pleroma.Config.get([__MODULE__, :max_jobs]) && queue != [] do
-      {{type, payload}, queue} = queue_pop(queue)
-      {:ok, pid} = Task.start(fn -> handle(type, payload) end)
-      mref = Process.monitor(pid)
-      {:sets.add_element(mref, running_jobs), queue}
-    else
-      {running_jobs, queue}
-    end
-  end
-
-  def handle_cast({:enqueue, type, payload, _priority}, state)
-      when type in [:incoming_doc, :incoming_ap_doc] do
-    %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
-    i_queue = enqueue_sorted(i_queue, {type, payload}, 1)
-    {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue)
-    {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
-  end
-
-  def handle_cast({:enqueue, type, payload, _priority}, state) do
-    %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
-    o_queue = enqueue_sorted(o_queue, {type, payload}, 1)
-    {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue)
-    {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
-  end
-
-  def handle_cast(m, state) do
-    IO.inspect("Unknown: #{inspect(m)}, #{inspect(state)}")
-    {:noreply, state}
-  end
-
-  def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
-    %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
-    i_running_jobs = :sets.del_element(ref, i_running_jobs)
-    o_running_jobs = :sets.del_element(ref, o_running_jobs)
-    {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue)
-    {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue)
-
-    {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
-  end
-
-  def enqueue_sorted(queue, element, priority) do
-    [%{item: element, priority: priority} | queue]
-    |> Enum.sort_by(fn %{priority: priority} -> priority end)
-  end
-
-  def queue_pop([%{item: element} | queue]) do
-    {element, queue}
-  end
-
   def ap_enabled_actor(id) do
     user = User.get_by_ap_id(id)
 

lib/pleroma/web/ostatus/ostatus_controller.ex

@@ -82,7 +82,7 @@ defmodule Pleroma.Web.OStatus.OStatusController do
     {:ok, body, _conn} = read_body(conn)
     {:ok, doc} = decode_or_retry(body)
 
-    Federator.enqueue(:incoming_doc, doc)
+    Federator.incoming_doc(doc)
 
     conn
     |> send_resp(200, "")

lib/pleroma/web/websub/websub.ex

@@ -7,7 +7,7 @@ defmodule Pleroma.Web.Websub do
   alias Pleroma.Repo
   alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
   alias Pleroma.Web.OStatus.FeedRepresenter
-  alias Pleroma.Web.{XML, Endpoint, OStatus}
+  alias Pleroma.Web.{XML, Endpoint, OStatus, Federator}
   alias Pleroma.Web.Router.Helpers
   require Logger
 
@@ -77,7 +77,7 @@ defmodule Pleroma.Web.Websub do
         secret: sub.secret
       }
 
-      Pleroma.Web.Federator.enqueue(:publish_single_websub, data)
+      Federator.publish_single_websub(data)
     end)
   end
 
@@ -109,7 +109,7 @@ defmodule Pleroma.Web.Websub do
 
       websub = Repo.update!(change)
 
-      Pleroma.Web.Federator.enqueue(:verify_websub, websub)
+      Federator.verify_websub(websub)
 
       {:ok, websub}
     else
@@ -259,7 +259,7 @@ defmodule Pleroma.Web.Websub do
     subs = Repo.all(query)
 
     Enum.each(subs, fn sub ->
-      Pleroma.Web.Federator.enqueue(:request_subscription, sub)
+      Federator.request_subscription(sub)
     end)
   end
 

lib/pleroma/web/websub/websub_controller.ex

@@ -80,7 +80,7 @@ defmodule Pleroma.Web.Websub.WebsubController do
          %WebsubClientSubscription{} = websub <- Repo.get(WebsubClientSubscription, id),
          {:ok, body, _conn} = read_body(conn),
          ^signature <- Websub.sign(websub.secret, body) do
-      Federator.enqueue(:incoming_doc, body)
+      Federator.incoming_doc(body)
 
       conn
       |> send_resp(200, "OK")