[#1149] Added more oban workers. Refactoring.

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -17,6 +17,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Web.ActivityPub.MRF
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.WebFinger
+  alias Pleroma.Workers.BackgroundWorker
 
   import Ecto.Query
   import Pleroma.Web.ActivityPub.Utils
@@ -25,6 +26,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   require Logger
   require Pleroma.Constants
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   # For Announce activities, we filter the recipients based on following status for any actors
   # that match actual users.  See issue #164 for more information about why this is necessary.
   defp get_recipients(%{"type" => "Announce"} = data) do
@@ -145,7 +148,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
           activity
         end
 
-      PleromaJobQueue.enqueue(:background, Pleroma.Web.RichMedia.Helpers, [:fetch, activity])
+      %{"op" => "fetch_data_for_activity", "activity_id" => activity.id}
+      |> BackgroundWorker.new(worker_args(:background))
+      |> Repo.insert()
 
       Notification.create_notifications(activity)
 

lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex

@@ -7,7 +7,9 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
   @behaviour Pleroma.Web.ActivityPub.MRF
 
   alias Pleroma.HTTP
+  alias Pleroma.Repo
   alias Pleroma.Web.MediaProxy
+  alias Pleroma.Workers.BackgroundWorker
 
   require Logger
 
@@ -16,6 +18,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
     recv_timeout: 10_000
   ]
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   def perform(:prefetch, url) do
     Logger.info("Prefetching #{inspect(url)}")
 
@@ -30,7 +34,9 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
         url
         |> Enum.each(fn
           %{"href" => href} ->
-            PleromaJobQueue.enqueue(:background, __MODULE__, [:prefetch, href])
+            %{"op" => "media_proxy_prefetch", "url" => href}
+            |> BackgroundWorker.new(worker_args(:background))
+            |> Repo.insert()
 
           x ->
             Logger.debug("Unhandled attachment URL object #{inspect(x)}")
@@ -46,7 +52,9 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
         %{"type" => "Create", "object" => %{"attachment" => attachments} = _object} = message
       )
       when is_list(attachments) and length(attachments) > 0 do
-    PleromaJobQueue.enqueue(:background, __MODULE__, [:preload, message])
+    %{"op" => "media_proxy_preload", "message" => message}
+    |> BackgroundWorker.new(worker_args(:background))
+    |> Repo.insert()
 
     {:ok, message}
   end

lib/pleroma/web/activity_pub/transmogrifier.ex

@@ -15,12 +15,15 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.ActivityPub.Visibility
   alias Pleroma.Web.Federator
+  alias Pleroma.Workers.Transmogrifier, as: TransmogrifierWorker
 
   import Ecto.Query
 
   require Logger
   require Pleroma.Constants
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   @doc """
   Modifies an incoming AP object (mastodon format) to our internal format.
   """
@@ -1073,7 +1076,9 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
          already_ap <- User.ap_enabled?(user),
          {:ok, user} <- user |> User.upgrade_changeset(data) |> User.update_and_set_cache() do
       unless already_ap do
-        PleromaJobQueue.enqueue(:transmogrifier, __MODULE__, [:user_upgrade, user])
+        %{"op" => "user_upgrade", "user_id" => user.id}
+        |> TransmogrifierWorker.new(worker_args(:transmogrifier))
+        |> Repo.insert()
       end
 
       {:ok, user}

lib/pleroma/web/federator/federator.ex

@@ -3,12 +3,23 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Federator do
+  alias Pleroma.Activity
+  alias Pleroma.Object.Containment
+  alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.ActivityPub
+  alias Pleroma.Web.ActivityPub.Transmogrifier
+  alias Pleroma.Web.ActivityPub.Utils
+  alias Pleroma.Web.Federator.Publisher
+  alias Pleroma.Web.OStatus
+  alias Pleroma.Web.Websub
   alias Pleroma.Workers.Publisher, as: PublisherWorker
   alias Pleroma.Workers.Receiver, as: ReceiverWorker
   alias Pleroma.Workers.Subscriber, as: SubscriberWorker
 
   require Logger
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   def init do
     # 1 minute
     refresh_subscriptions(schedule_in: 60)
@@ -41,7 +52,7 @@ defmodule Pleroma.Web.Federator do
   end
 
   def publish(%{id: "pleroma:fakeid"} = activity) do
-    PublisherWorker.perform_publish(activity)
+    perform(:publish, activity)
   end
 
   def publish(activity) do
@@ -68,11 +79,88 @@ defmodule Pleroma.Web.Federator do
     |> Pleroma.Repo.insert()
   end
 
-  defp worker_args(queue) do
-    if max_attempts = Pleroma.Config.get([:workers, :retries, queue]) do
-      [max_attempts: max_attempts]
+  # Job Worker Callbacks
+
+  @spec perform(atom(), module(), any()) :: {:ok, any()} | {:error, any()}
+  def perform(:publish_one, module, params) do
+    apply(module, :publish_one, [params])
+  end
+
+  def perform(:publish, activity) do
+    Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)
+
+    with %User{} = actor <- User.get_cached_by_ap_id(activity.data["actor"]),
+         {:ok, actor} <- User.ensure_keys_present(actor) do
+      Publisher.publish(actor, activity)
+    end
+  end
+
+  def perform(:incoming_doc, doc) do
+    Logger.info("Got document, trying to parse")
+    OStatus.handle_incoming(doc)
+  end
+
+  def perform(:incoming_ap_doc, params) do
+    Logger.info("Handling incoming AP activity")
+
+    params = Utils.normalize_params(params)
+
+    # NOTE: we use the actor ID to do the containment, this is fine because an
+    # actor shouldn't be acting on objects outside their own AP server.
+    with {:ok, _user} <- ap_enabled_actor(params["actor"]),
+         nil <- Activity.normalize(params["id"]),
+         :ok <- Containment.contain_origin_from_id(params["actor"], params),
+         {:ok, activity} <- Transmogrifier.handle_incoming(params) do
+      {:ok, activity}
     else
-      []
+      %Activity{} ->
+        Logger.info("Already had #{params["id"]}")
+        :error
+
+      _e ->
+        # Just drop those for now
+        Logger.info("Unhandled activity")
+        Logger.info(Jason.encode!(params, pretty: true))
+        :error
+    end
+  end
+
+  def perform(:request_subscription, websub) do
+    Logger.debug("Refreshing #{websub.topic}")
+
+    with {:ok, websub} <- Websub.request_subscription(websub) do
+      Logger.debug("Successfully refreshed #{websub.topic}")
+    else
+      _e -> Logger.debug("Couldn't refresh #{websub.topic}")
+    end
+  end
+
+  def perform(:verify_websub, websub) do
+    Logger.debug(fn ->
+      "Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})"
+    end)
+
+    Websub.verify(websub)
+  end
+
+  def perform(:refresh_subscriptions) do
+    Logger.debug("Federator running refresh subscriptions")
+    Websub.refresh_subscriptions()
+
+    spawn(fn ->
+      # 6 hours
+      Process.sleep(1000 * 60 * 60 * 6)
+      refresh_subscriptions()
+    end)
+  end
+
+  def ap_enabled_actor(id) do
+    user = User.get_cached_by_ap_id(id)
+
+    if User.ap_enabled?(user) do
+      {:ok, user}
+    else
+      ActivityPub.make_user_from_ap_id(id)
     end
   end
 end

lib/pleroma/web/oauth/token/clean_worker.ex

@@ -14,9 +14,12 @@ defmodule Pleroma.Web.OAuth.Token.CleanWorker do
               [:oauth2, :clean_expired_tokens_interval],
               86_400_000
             )
-  @queue :background
 
+  alias Pleroma.Repo
   alias Pleroma.Web.OAuth.Token
+  alias Pleroma.Workers.BackgroundWorker
+
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
 
   def start_link, do: GenServer.start_link(__MODULE__, nil)
 
@@ -31,8 +34,11 @@ defmodule Pleroma.Web.OAuth.Token.CleanWorker do
 
   @doc false
   def handle_info(:perform, state) do
+    %{"op" => "clean_expired_tokens"}
+    |> BackgroundWorker.new(worker_args(:background))
+    |> Repo.insert()
+
     Process.send_after(self(), :perform, @interval)
-    PleromaJobQueue.enqueue(@queue, __MODULE__, [:clean])
     {:noreply, state}
   end
 

lib/pleroma/web/push/push.ex

@@ -3,10 +3,13 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Push do
-  alias Pleroma.Web.Push.Impl
+  alias Pleroma.Repo
+  alias Pleroma.Workers.WebPusher
 
   require Logger
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   def init do
     unless enabled() do
       Logger.warn("""
@@ -31,6 +34,9 @@ defmodule Pleroma.Web.Push do
     end
   end
 
-  def send(notification),
-    do: PleromaJobQueue.enqueue(:web_push, Impl, [notification])
+  def send(notification) do
+    %{"op" => "web_push", "notification_id" => notification.id}
+    |> WebPusher.new(worker_args(:web_push))
+    |> Repo.insert()
+  end
 end

lib/pleroma/web/twitter_api/controllers/util_controller.ex

@@ -265,12 +265,7 @@ defmodule Pleroma.Web.TwitterAPI.UtilController do
              String.split(line, ",") |> List.first()
            end)
            |> List.delete("Account address") do
-      PleromaJobQueue.enqueue(:background, User, [
-        :follow_import,
-        follower,
-        followed_identifiers
-      ])
-
+      User.follow_import(follower, followed_identifiers)
       json(conn, "job started")
     end
   end
@@ -281,12 +276,7 @@ defmodule Pleroma.Web.TwitterAPI.UtilController do
 
   def blocks_import(%{assigns: %{user: blocker}} = conn, %{"list" => list}) do
     with blocked_identifiers <- String.split(list) do
-      PleromaJobQueue.enqueue(:background, User, [
-        :blocks_import,
-        blocker,
-        blocked_identifiers
-      ])
-
+      User.blocks_import(blocker, blocked_identifiers)
       json(conn, "job started")
     end
   end