From 23d279e03ee1f7a1285614754738711359bc4b81 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Thu, 1 Aug 2019 17:28:00 +0300
Subject: [#1149] Replaced RetryQueue with oban-based retries.

---
 lib/pleroma/application.ex                |   4 +-
 lib/pleroma/web/activity_pub/publisher.ex |  16 +-
 lib/pleroma/web/federator/federator.ex    |  14 --
 lib/pleroma/web/federator/publisher.ex    |  22 +--
 lib/pleroma/web/federator/retry_queue.ex  | 239 ------------------------------
 lib/pleroma/web/salmon/salmon.ex          |  11 +-
 lib/pleroma/workers/publisher.ex          |  14 ++
 7 files changed, 43 insertions(+), 277 deletions(-)
 delete mode 100644 lib/pleroma/web/federator/retry_queue.ex
 create mode 100644 lib/pleroma/workers/publisher.ex

(limited to 'lib')

diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 035331491..ce7d8c4b2 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -120,8 +120,8 @@ defmodule Pleroma.Application do
         hackney_pool_children() ++
         [
           %{
-            id: Pleroma.Web.Federator.RetryQueue,
-            start: {Pleroma.Web.Federator.RetryQueue, :start_link, []}
+            id: Oban,
+            start: {Oban, :start_link, [Application.get_env(:pleroma, Oban)]}
           },
           %{
             id: Pleroma.Web.OAuth.Token.CleanWorker,
diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
index 46edab0bd..29f3221d1 100644
--- a/lib/pleroma/web/activity_pub/publisher.ex
+++ b/lib/pleroma/web/activity_pub/publisher.ex
@@ -85,6 +85,15 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
     end
   end
 
+  def publish_one(%{actor_id: actor_id} = params) do
+    actor = User.get_by_id(actor_id)
+
+    params
+    |> Map.delete(:actor_id)
+    |> Map.put(:actor, actor)
+    |> publish_one()
+  end
+
   defp should_federate?(inbox, public) do
     if public do
       true
@@ -160,7 +169,8 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
   Publishes an activity with BCC to all relevant peers.
   """
 
-  def publish(actor, %{data: %{"bcc" => bcc}} = activity) when is_list(bcc) and bcc != [] do
+  def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
+      when is_list(bcc) and bcc != [] do
     public = is_public?(activity)
     {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 
@@ -187,7 +197,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
       Pleroma.Web.Federator.Publisher.enqueue_one(__MODULE__, %{
         inbox: inbox,
         json: json,
-        actor: actor,
+        actor_id: actor.id,
         id: activity.data["id"],
         unreachable_since: unreachable_since
       })
@@ -222,7 +232,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
         %{
           inbox: inbox,
           json: json,
-          actor: actor,
+          actor_id: actor.id,
           id: activity.data["id"],
           unreachable_since: unreachable_since
         }
diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
index f4f9e83e0..97ec9d549 100644
--- a/lib/pleroma/web/federator/federator.ex
+++ b/lib/pleroma/web/federator/federator.ex
@@ -10,7 +10,6 @@ defmodule Pleroma.Web.Federator do
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.Federator.Publisher
-  alias Pleroma.Web.Federator.RetryQueue
   alias Pleroma.Web.OStatus
   alias Pleroma.Web.Websub
 
@@ -130,19 +129,6 @@ defmodule Pleroma.Web.Federator do
     end
   end
 
-  def perform(
-        :publish_single_websub,
-        %{xml: _xml, topic: _topic, callback: _callback, secret: _secret} = params
-      ) do
-    case Websub.publish_one(params) do
-      {:ok, _} ->
-        :ok
-
-      {:error, _} ->
-        RetryQueue.enqueue(params, Websub)
-    end
-  end
-
   def perform(type, _) do
     Logger.debug(fn -> "Unknown task: #{type}" end)
     {:error, "Don't know what to do with this"}
diff --git a/lib/pleroma/web/federator/publisher.ex b/lib/pleroma/web/federator/publisher.ex
index 70f870244..e8c1bf17f 100644
--- a/lib/pleroma/web/federator/publisher.ex
+++ b/lib/pleroma/web/federator/publisher.ex
@@ -6,7 +6,6 @@ defmodule Pleroma.Web.Federator.Publisher do
   alias Pleroma.Activity
   alias Pleroma.Config
   alias Pleroma.User
-  alias Pleroma.Web.Federator.RetryQueue
 
   require Logger
 
@@ -30,23 +29,10 @@ defmodule Pleroma.Web.Federator.Publisher do
   Enqueue publishing a single activity.
   """
   @spec enqueue_one(module(), Map.t()) :: :ok
-  def enqueue_one(module, %{} = params),
-    do: PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_one, module, params])
-
-  @spec perform(atom(), module(), any()) :: {:ok, any()} | {:error, any()}
-  def perform(:publish_one, module, params) do
-    case apply(module, :publish_one, [params]) do
-      {:ok, _} ->
-        :ok
-
-      {:error, _e} ->
-        RetryQueue.enqueue(params, module)
-    end
-  end
-
-  def perform(type, _, _) do
-    Logger.debug("Unknown task: #{type}")
-    {:error, "Don't know what to do with this"}
+  def enqueue_one(module, %{} = params) do
+    %{module: to_string(module), params: params}
+    |> Pleroma.Workers.Publisher.new()
+    |> Pleroma.Repo.insert()
   end
 
   @doc """
diff --git a/lib/pleroma/web/federator/retry_queue.ex b/lib/pleroma/web/federator/retry_queue.ex
deleted file mode 100644
index 3db948c2e..000000000
--- a/lib/pleroma/web/federator/retry_queue.ex
+++ /dev/null
@@ -1,239 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Web.Federator.RetryQueue do
-  use GenServer
-
-  require Logger
-
-  def init(args) do
-    queue_table = :ets.new(:pleroma_retry_queue, [:bag, :protected])
-
-    {:ok, %{args | queue_table: queue_table, running_jobs: :sets.new()}}
-  end
-
-  def start_link do
-    enabled =
-      if Pleroma.Config.get(:env) == :test,
-        do: true,
-        else: Pleroma.Config.get([__MODULE__, :enabled], false)
-
-    if enabled do
-      Logger.info("Starting retry queue")
-
-      linkres =
-        GenServer.start_link(
-          __MODULE__,
-          %{delivered: 0, dropped: 0, queue_table: nil, running_jobs: nil},
-          name: __MODULE__
-        )
-
-      maybe_kickoff_timer()
-      linkres
-    else
-      Logger.info("Retry queue disabled")
-      :ignore
-    end
-  end
-
-  def enqueue(data, transport, retries \\ 0) do
-    GenServer.cast(__MODULE__, {:maybe_enqueue, data, transport, retries + 1})
-  end
-
-  def get_stats do
-    GenServer.call(__MODULE__, :get_stats)
-  end
-
-  def reset_stats do
-    GenServer.call(__MODULE__, :reset_stats)
-  end
-
-  def get_retry_params(retries) do
-    if retries > Pleroma.Config.get([__MODULE__, :max_retries]) do
-      {:drop, "Max retries reached"}
-    else
-      {:retry, growth_function(retries)}
-    end
-  end
-
-  def get_retry_timer_interval do
-    Pleroma.Config.get([:retry_queue, :interval], 1000)
-  end
-
-  defp ets_count_expires(table, current_time) do
-    :ets.select_count(
-      table,
-      [
-        {
-          {:"$1", :"$2"},
-          [{:"=<", :"$1", {:const, current_time}}],
-          [true]
-        }
-      ]
-    )
-  end
-
-  defp ets_pop_n_expired(table, current_time, desired) do
-    {popped, _continuation} =
-      :ets.select(
-        table,
-        [
-          {
-            {:"$1", :"$2"},
-            [{:"=<", :"$1", {:const, current_time}}],
-            [:"$_"]
-          }
-        ],
-        desired
-      )
-
-    popped
-    |> Enum.each(fn e ->
-      :ets.delete_object(table, e)
-    end)
-
-    popped
-  end
-
-  def maybe_start_job(running_jobs, queue_table) do
-    # we don't want to hit the ets or the DateTime more times than we have to
-    # could optimize slightly further by not using the count, and instead grabbing
-    # up to N objects early...
-    current_time = DateTime.to_unix(DateTime.utc_now())
-    n_running_jobs = :sets.size(running_jobs)
-
-    if n_running_jobs < Pleroma.Config.get([__MODULE__, :max_jobs]) do
-      n_ready_jobs = ets_count_expires(queue_table, current_time)
-
-      if n_ready_jobs > 0 do
-        # figure out how many we could start
-        available_job_slots = Pleroma.Config.get([__MODULE__, :max_jobs]) - n_running_jobs
-        start_n_jobs(running_jobs, queue_table, current_time, available_job_slots)
-      else
-        running_jobs
-      end
-    else
-      running_jobs
-    end
-  end
-
-  defp start_n_jobs(running_jobs, _queue_table, _current_time, 0) do
-    running_jobs
-  end
-
-  defp start_n_jobs(running_jobs, queue_table, current_time, available_job_slots)
-       when available_job_slots > 0 do
-    candidates = ets_pop_n_expired(queue_table, current_time, available_job_slots)
-
-    candidates
-    |> List.foldl(running_jobs, fn {_, e}, rj ->
-      {:ok, pid} = Task.start(fn -> worker(e) end)
-      mref = Process.monitor(pid)
-      :sets.add_element(mref, rj)
-    end)
-  end
-
-  def worker({:send, data, transport, retries}) do
-    case transport.publish_one(data) do
-      {:ok, _} ->
-        GenServer.cast(__MODULE__, :inc_delivered)
-        :delivered
-
-      {:error, _reason} ->
-        enqueue(data, transport, retries)
-        :retry
-    end
-  end
-
-  def handle_call(:get_stats, _from, %{delivered: delivery_count, dropped: drop_count} = state) do
-    {:reply, %{delivered: delivery_count, dropped: drop_count}, state}
-  end
-
-  def handle_call(:reset_stats, _from, %{delivered: delivery_count, dropped: drop_count} = state) do
-    {:reply, %{delivered: delivery_count, dropped: drop_count},
-     %{state | delivered: 0, dropped: 0}}
-  end
-
-  def handle_cast(:reset_stats, state) do
-    {:noreply, %{state | delivered: 0, dropped: 0}}
-  end
-
-  def handle_cast(
-        {:maybe_enqueue, data, transport, retries},
-        %{dropped: drop_count, queue_table: queue_table, running_jobs: running_jobs} = state
-      ) do
-    case get_retry_params(retries) do
-      {:retry, timeout} ->
-        :ets.insert(queue_table, {timeout, {:send, data, transport, retries}})
-        running_jobs = maybe_start_job(running_jobs, queue_table)
-        {:noreply, %{state | running_jobs: running_jobs}}
-
-      {:drop, message} ->
-        Logger.debug(message)
-        {:noreply, %{state | dropped: drop_count + 1}}
-    end
-  end
-
-  def handle_cast(:kickoff_timer, state) do
-    retry_interval = get_retry_timer_interval()
-    Process.send_after(__MODULE__, :retry_timer_run, retry_interval)
-    {:noreply, state}
-  end
-
-  def handle_cast(:inc_delivered, %{delivered: delivery_count} = state) do
-    {:noreply, %{state | delivered: delivery_count + 1}}
-  end
-
-  def handle_cast(:inc_dropped, %{dropped: drop_count} = state) do
-    {:noreply, %{state | dropped: drop_count + 1}}
-  end
-
-  def handle_info({:send, data, transport, retries}, %{delivered: delivery_count} = state) do
-    case transport.publish_one(data) do
-      {:ok, _} ->
-        {:noreply, %{state | delivered: delivery_count + 1}}
-
-      {:error, _reason} ->
-        enqueue(data, transport, retries)
-        {:noreply, state}
-    end
-  end
-
-  def handle_info(
-        :retry_timer_run,
-        %{queue_table: queue_table, running_jobs: running_jobs} = state
-      ) do
-    maybe_kickoff_timer()
-    running_jobs = maybe_start_job(running_jobs, queue_table)
-    {:noreply, %{state | running_jobs: running_jobs}}
-  end
-
-  def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
-    %{running_jobs: running_jobs, queue_table: queue_table} = state
-    running_jobs = :sets.del_element(ref, running_jobs)
-    running_jobs = maybe_start_job(running_jobs, queue_table)
-    {:noreply, %{state | running_jobs: running_jobs}}
-  end
-
-  def handle_info(unknown, state) do
-    Logger.debug("RetryQueue: don't know what to do with #{inspect(unknown)}, ignoring")
-    {:noreply, state}
-  end
-
-  if Pleroma.Config.get(:env) == :test do
-    defp growth_function(_retries) do
-      _shutit = Pleroma.Config.get([__MODULE__, :initial_timeout])
-      DateTime.to_unix(DateTime.utc_now()) - 1
-    end
-  else
-    defp growth_function(retries) do
-      round(Pleroma.Config.get([__MODULE__, :initial_timeout]) * :math.pow(retries, 3)) +
-        DateTime.to_unix(DateTime.utc_now())
-    end
-  end
-
-  defp maybe_kickoff_timer do
-    GenServer.cast(__MODULE__, :kickoff_timer)
-  end
-end
diff --git a/lib/pleroma/web/salmon/salmon.ex b/lib/pleroma/web/salmon/salmon.ex
index 9b01ebcc6..bbaa293fd 100644
--- a/lib/pleroma/web/salmon/salmon.ex
+++ b/lib/pleroma/web/salmon/salmon.ex
@@ -170,6 +170,15 @@ defmodule Pleroma.Web.Salmon do
     end
   end
 
+  def publish_one(%{recipient_id: recipient_id} = params) do
+    recipient = User.get_by_id(recipient_id)
+
+    params
+    |> Map.delete(:recipient_id)
+    |> Map.put(:recipient, recipient)
+    |> publish_one()
+  end
+
   def publish_one(_), do: :noop
 
   @supported_activities [
@@ -218,7 +227,7 @@ defmodule Pleroma.Web.Salmon do
         Logger.debug(fn -> "Sending Salmon to #{remote_user.ap_id}" end)
 
         Publisher.enqueue_one(__MODULE__, %{
-          recipient: remote_user,
+          recipient_id: remote_user.id,
           feed: feed,
           unreachable_since: reachable_urls_metadata[remote_user.info.salmon]
         })
diff --git a/lib/pleroma/workers/publisher.ex b/lib/pleroma/workers/publisher.ex
new file mode 100644
index 000000000..639794830
--- /dev/null
+++ b/lib/pleroma/workers/publisher.ex
@@ -0,0 +1,14 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Publisher do
+  use Oban.Worker, queue: "federator_outgoing", max_attempts: 5
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{module: module_name, params: params}}) do
+    module_name
+    |> String.to_atom()
+    |> apply(:publish_one, [params])
+  end
+end
-- 
cgit v1.2.3


From b7fad8d395c2bd1afe445a370e539571f5ec0c18 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Fri, 9 Aug 2019 20:08:01 +0300
Subject: [#1149] Oban jobs implementation for :federator_incoming and
 :federator_outgoing queues.

---
 lib/pleroma/web/activity_pub/utils.ex  |   9 +--
 lib/pleroma/web/federator/federator.ex | 132 ++++++++-------------------------
 lib/pleroma/web/federator/publisher.ex |  12 ++-
 lib/pleroma/workers/publisher.ex       |  25 ++++++-
 lib/pleroma/workers/receiver.ex        |  61 +++++++++++++++
 lib/pleroma/workers/subscriber.ex      |  44 +++++++++++
 6 files changed, 171 insertions(+), 112 deletions(-)
 create mode 100644 lib/pleroma/workers/receiver.ex
 create mode 100644 lib/pleroma/workers/subscriber.ex

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index 39074888b..f0917f9d4 100644
--- a/lib/pleroma/web/activity_pub/utils.ex
+++ b/lib/pleroma/web/activity_pub/utils.ex
@@ -168,14 +168,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   """
   def maybe_federate(%Activity{local: true} = activity) do
     if Pleroma.Config.get!([:instance, :federating]) do
-      priority =
-        case activity.data["type"] do
-          "Delete" -> 10
-          "Create" -> 1
-          _ -> 5
-        end
-
-      Pleroma.Web.Federator.publish(activity, priority)
+      Pleroma.Web.Federator.publish(activity)
     end
 
     :ok
diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
index 97ec9d549..bb9eadfee 100644
--- a/lib/pleroma/web/federator/federator.ex
+++ b/lib/pleroma/web/federator/federator.ex
@@ -3,22 +3,15 @@
 # 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
 
   def init do
     # 1 minute
-    Process.sleep(1000 * 60)
-    refresh_subscriptions()
+    refresh_subscriptions(schedule_in: 60)
   end
 
   @doc "Addresses [memory leaks on recursive replies fetching](https://git.pleroma.social/pleroma/pleroma/issues/161)"
@@ -36,111 +29,50 @@ defmodule Pleroma.Web.Federator do
   # Client API
 
   def incoming_doc(doc) do
-    PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc])
+    %{"op" => "incoming_doc", "body" => doc}
+    |> ReceiverWorker.new(worker_args(:federator_incoming))
+    |> Pleroma.Repo.insert()
   end
 
   def incoming_ap_doc(params) do
-    PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params])
+    %{"op" => "incoming_ap_doc", "params" => params}
+    |> ReceiverWorker.new(worker_args(:federator_incoming))
+    |> Pleroma.Repo.insert()
   end
 
-  def publish(activity, priority \\ 1) do
-    PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish, activity], priority)
+  def publish(%{id: "pleroma:fakeid"} = activity) do
+    PublisherWorker.perform_publish(activity)
   end
 
-  def verify_websub(websub) do
-    PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:verify_websub, websub])
-  end
-
-  def request_subscription(sub) do
-    PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:request_subscription, sub])
-  end
-
-  def refresh_subscriptions do
-    PleromaJobQueue.enqueue(:federator_outgoing, __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)
-      refresh_subscriptions()
-    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
+  def publish(activity) do
+    %{"op" => "publish", "activity_id" => activity.id}
+    |> PublisherWorker.new(worker_args(:federator_outgoing))
+    |> Pleroma.Repo.insert()
   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(:verify_websub, websub) do
-    Logger.debug(fn ->
-      "Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})"
-    end)
-
-    Websub.verify(websub)
-  end
-
-  def perform(:incoming_doc, doc) do
-    Logger.info("Got document, trying to parse")
-    OStatus.handle_incoming(doc)
+  def verify_websub(websub) do
+    %{"op" => "verify_websub", "websub_id" => websub.id}
+    |> SubscriberWorker.new(worker_args(:federator_outgoing))
+    |> Pleroma.Repo.insert()
   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
+  def request_subscription(websub) do
+    %{"op" => "request_subscription", "websub_id" => websub.id}
+    |> SubscriberWorker.new(worker_args(:federator_outgoing))
+    |> Pleroma.Repo.insert()
   end
 
-  def perform(type, _) do
-    Logger.debug(fn -> "Unknown task: #{type}" end)
-    {:error, "Don't know what to do with this"}
+  def refresh_subscriptions(worker_args \\ []) do
+    %{"op" => "refresh_subscriptions"}
+    |> SubscriberWorker.new(worker_args ++ [max_attempts: 1] ++ worker_args(:federator_outgoing))
+    |> Pleroma.Repo.insert()
   end
 
-  def ap_enabled_actor(id) do
-    user = User.get_cached_by_ap_id(id)
-
-    if User.ap_enabled?(user) do
-      {:ok, user}
+  defp worker_args(queue) do
+    if max_attempts = Pleroma.Config.get([:workers, :retries, queue]) do
+      [max_attempts: max_attempts]
     else
-      ActivityPub.make_user_from_ap_id(id)
+      []
     end
   end
 end
diff --git a/lib/pleroma/web/federator/publisher.ex b/lib/pleroma/web/federator/publisher.ex
index e8c1bf17f..05d2be615 100644
--- a/lib/pleroma/web/federator/publisher.ex
+++ b/lib/pleroma/web/federator/publisher.ex
@@ -6,6 +6,7 @@ defmodule Pleroma.Web.Federator.Publisher do
   alias Pleroma.Activity
   alias Pleroma.Config
   alias Pleroma.User
+  alias Pleroma.Workers.Publisher, as: PublisherWorker
 
   require Logger
 
@@ -30,8 +31,15 @@ defmodule Pleroma.Web.Federator.Publisher do
   """
   @spec enqueue_one(module(), Map.t()) :: :ok
   def enqueue_one(module, %{} = params) do
-    %{module: to_string(module), params: params}
-    |> Pleroma.Workers.Publisher.new()
+    worker_args =
+      if max_attempts = Pleroma.Config.get([:workers, :retries, :federator_outgoing]) do
+        [max_attempts: max_attempts]
+      else
+        []
+      end
+
+    %{"op" => "publish_one", "module" => to_string(module), "params" => params}
+    |> PublisherWorker.new(worker_args)
     |> Pleroma.Repo.insert()
   end
 
diff --git a/lib/pleroma/workers/publisher.ex b/lib/pleroma/workers/publisher.ex
index 639794830..67871977a 100644
--- a/lib/pleroma/workers/publisher.ex
+++ b/lib/pleroma/workers/publisher.ex
@@ -3,12 +3,33 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.Publisher do
-  use Oban.Worker, queue: "federator_outgoing", max_attempts: 5
+  alias Pleroma.Activity
+  alias Pleroma.User
+
+  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  use Oban.Worker,
+    queue: "federator_outgoing",
+    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
 
   @impl Oban.Worker
-  def perform(%Oban.Job{args: %{module: module_name, params: params}}) do
+  def perform(%{"op" => "publish", "activity_id" => activity_id}) do
+    with %Activity{} = activity <- Activity.get_by_id(activity_id) do
+      perform_publish(activity)
+    else
+      _ -> raise "Non-existing activity: #{activity_id}"
+    end
+  end
+
+  def perform(%{"op" => "publish_one", "module" => module_name, "params" => params}) do
     module_name
     |> String.to_atom()
     |> apply(:publish_one, [params])
   end
+
+  def perform_publish(%Activity{} = activity) do
+    with %User{} = actor <- User.get_cached_by_ap_id(activity.data["actor"]),
+         {:ok, actor} <- User.ensure_keys_present(actor) do
+      Pleroma.Web.Federator.Publisher.publish(actor, activity)
+    end
+  end
 end
diff --git a/lib/pleroma/workers/receiver.ex b/lib/pleroma/workers/receiver.ex
new file mode 100644
index 000000000..43558b4e6
--- /dev/null
+++ b/lib/pleroma/workers/receiver.ex
@@ -0,0 +1,61 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Receiver 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.OStatus
+
+  require Logger
+
+  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  use Oban.Worker,
+    queue: "federator_incoming",
+    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+
+  @impl Oban.Worker
+  def perform(%{"op" => "incoming_doc", "body" => doc}) do
+    Logger.info("Got incoming document, trying to parse")
+    OStatus.handle_incoming(doc)
+  end
+
+  def perform(%{"op" => "incoming_ap_doc", "params" => 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
+
+  defp 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
diff --git a/lib/pleroma/workers/subscriber.ex b/lib/pleroma/workers/subscriber.ex
new file mode 100644
index 000000000..a8c01bb10
--- /dev/null
+++ b/lib/pleroma/workers/subscriber.ex
@@ -0,0 +1,44 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Subscriber do
+  alias Pleroma.Repo
+  alias Pleroma.Web.Websub
+  alias Pleroma.Web.Websub.WebsubClientSubscription
+
+  require Logger
+
+  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  use Oban.Worker,
+    queue: "federator_outgoing",
+    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+
+  @impl Oban.Worker
+  def perform(%{"op" => "refresh_subscriptions"}) do
+    Websub.refresh_subscriptions()
+    # Schedule the next run in 6 hours
+    Pleroma.Web.Federator.refresh_subscriptions(schedule_in: 3600 * 6)
+  end
+
+  def perform(%{"op" => "request_subscription", "websub_id" => websub_id}) do
+    websub = Repo.get(WebsubClientSubscription, websub_id)
+    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(%{"op" => "verify_websub", "websub_id" => websub_id}) do
+    websub = Repo.get(WebsubClientSubscription, websub_id)
+
+    Logger.debug(fn ->
+      "Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})"
+    end)
+
+    Websub.verify(websub)
+  end
+end
-- 
cgit v1.2.3


From 33a5fc4a70b6f9b8c2d8c03a412d7eec8d5b3db1 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Sat, 10 Aug 2019 20:38:31 +0300
Subject: [#1149] Fixed failing tests. Ensured Instance.set_unreachable/2
 supports ISO 8601 datetime.

---
 lib/pleroma/digest_email_worker.ex | 4 +---
 lib/pleroma/instances/instance.ex  | 8 +++++++-
 2 files changed, 8 insertions(+), 4 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/digest_email_worker.ex b/lib/pleroma/digest_email_worker.ex
index 18e67d39b..3b0e2bca6 100644
--- a/lib/pleroma/digest_email_worker.ex
+++ b/lib/pleroma/digest_email_worker.ex
@@ -1,8 +1,6 @@
 defmodule Pleroma.DigestEmailWorker do
   import Ecto.Query
 
-  @queue_name :digest_emails
-
   def perform do
     config = Pleroma.Config.get([:email_notifications, :digest])
     negative_interval = -Map.fetch!(config, :interval)
@@ -17,7 +15,7 @@ defmodule Pleroma.DigestEmailWorker do
       select: u
     )
     |> Pleroma.Repo.all()
-    |> Enum.each(&PleromaJobQueue.enqueue(@queue_name, __MODULE__, [&1]))
+    |> Enum.each(&PleromaJobQueue.enqueue(:digest_emails, __MODULE__, [&1]))
   end
 
   @doc """
diff --git a/lib/pleroma/instances/instance.ex b/lib/pleroma/instances/instance.ex
index 4d7ed4ca1..544c4b687 100644
--- a/lib/pleroma/instances/instance.ex
+++ b/lib/pleroma/instances/instance.ex
@@ -90,7 +90,7 @@ defmodule Pleroma.Instances.Instance do
   def set_unreachable(url_or_host, unreachable_since \\ nil)
 
   def set_unreachable(url_or_host, unreachable_since) when is_binary(url_or_host) do
-    unreachable_since = unreachable_since || DateTime.utc_now()
+    unreachable_since = parse_datetime(unreachable_since) || NaiveDateTime.utc_now()
     host = host(url_or_host)
     existing_record = Repo.get_by(Instance, %{host: host})
 
@@ -114,4 +114,10 @@ defmodule Pleroma.Instances.Instance do
   end
 
   def set_unreachable(_, _), do: {:error, nil}
+
+  defp parse_datetime(datetime) when is_binary(datetime) do
+    NaiveDateTime.from_iso8601(datetime)
+  end
+
+  defp parse_datetime(datetime), do: datetime
 end
-- 
cgit v1.2.3


From 0e1c481a94392b69833fbe6afc184ebbd90e1330 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Tue, 13 Aug 2019 20:20:26 +0300
Subject: [#1149] Added more oban workers. Refactoring.

---
 lib/pleroma/digest_email_worker.ex                 | 11 ++-
 lib/pleroma/scheduled_activity_worker.ex           |  8 +-
 lib/pleroma/user.ex                                | 57 ++++++++-----
 lib/pleroma/web/activity_pub/activity_pub.ex       |  7 +-
 .../activity_pub/mrf/mediaproxy_warming_policy.ex  | 12 ++-
 lib/pleroma/web/activity_pub/transmogrifier.ex     |  7 +-
 lib/pleroma/web/federator/federator.ex             | 98 ++++++++++++++++++++--
 lib/pleroma/web/oauth/token/clean_worker.ex        | 10 ++-
 lib/pleroma/web/push/push.ex                       | 12 ++-
 .../web/twitter_api/controllers/util_controller.ex | 14 +---
 lib/pleroma/workers/background_worker.ex           | 66 +++++++++++++++
 lib/pleroma/workers/helper.ex                      | 13 +++
 lib/pleroma/workers/mailer.ex                      | 18 ++++
 lib/pleroma/workers/publisher.ex                   | 20 +----
 lib/pleroma/workers/receiver.ex                    | 46 +---------
 lib/pleroma/workers/scheduled_activity_worker.ex   | 15 ++++
 lib/pleroma/workers/subscriber.ex                  | 23 +----
 lib/pleroma/workers/transmogrifier.ex              | 18 ++++
 lib/pleroma/workers/web_pusher.ex                  | 19 +++++
 19 files changed, 347 insertions(+), 127 deletions(-)
 create mode 100644 lib/pleroma/workers/background_worker.ex
 create mode 100644 lib/pleroma/workers/helper.ex
 create mode 100644 lib/pleroma/workers/mailer.ex
 create mode 100644 lib/pleroma/workers/scheduled_activity_worker.ex
 create mode 100644 lib/pleroma/workers/transmogrifier.ex
 create mode 100644 lib/pleroma/workers/web_pusher.ex

(limited to 'lib')

diff --git a/lib/pleroma/digest_email_worker.ex b/lib/pleroma/digest_email_worker.ex
index 3b0e2bca6..6e44cc955 100644
--- a/lib/pleroma/digest_email_worker.ex
+++ b/lib/pleroma/digest_email_worker.ex
@@ -1,6 +1,11 @@
 defmodule Pleroma.DigestEmailWorker do
+  alias Pleroma.Repo
+  alias Pleroma.Workers.Mailer, as: MailerWorker
+
   import Ecto.Query
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   def perform do
     config = Pleroma.Config.get([:email_notifications, :digest])
     negative_interval = -Map.fetch!(config, :interval)
@@ -15,7 +20,11 @@ defmodule Pleroma.DigestEmailWorker do
       select: u
     )
     |> Pleroma.Repo.all()
-    |> Enum.each(&PleromaJobQueue.enqueue(:digest_emails, __MODULE__, [&1]))
+    |> Enum.each(fn user ->
+      %{"op" => "digest_email", "user_id" => user.id}
+      |> MailerWorker.new([queue: "digest_emails"] ++ worker_args(:digest_emails))
+      |> Repo.insert()
+    end)
   end
 
   @doc """
diff --git a/lib/pleroma/scheduled_activity_worker.ex b/lib/pleroma/scheduled_activity_worker.ex
index 65b38622f..cabea51ca 100644
--- a/lib/pleroma/scheduled_activity_worker.ex
+++ b/lib/pleroma/scheduled_activity_worker.ex
@@ -8,14 +8,18 @@ defmodule Pleroma.ScheduledActivityWorker do
   """
 
   alias Pleroma.Config
+  alias Pleroma.Repo
   alias Pleroma.ScheduledActivity
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
+
   use GenServer
   require Logger
 
   @schedule_interval :timer.minutes(1)
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   def start_link do
     GenServer.start_link(__MODULE__, nil)
   end
@@ -45,7 +49,9 @@ defmodule Pleroma.ScheduledActivityWorker do
   def handle_info(:perform, state) do
     ScheduledActivity.due_activities(@schedule_interval)
     |> Enum.each(fn scheduled_activity ->
-      PleromaJobQueue.enqueue(:scheduled_activities, __MODULE__, [:execute, scheduled_activity.id])
+      %{"op" => "execute", "activity_id" => scheduled_activity.id}
+      |> Pleroma.Workers.ScheduledActivityWorker.new(worker_args(:scheduled_activities))
+      |> Repo.insert()
     end)
 
     schedule_next()
diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 7d18f099e..bc2102ca7 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -26,6 +26,7 @@ defmodule Pleroma.User do
   alias Pleroma.Web.OStatus
   alias Pleroma.Web.RelMe
   alias Pleroma.Web.Websub
+  alias Pleroma.Workers.BackgroundWorker
 
   require Logger
 
@@ -39,6 +40,8 @@ defmodule Pleroma.User do
   @strict_local_nickname_regex ~r/^[a-zA-Z\d]+$/
   @extended_local_nickname_regex ~r/^[a-zA-Z\d_-]+$/
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   schema "users" do
     field(:bio, :string)
     field(:email, :string)
@@ -579,8 +582,11 @@ defmodule Pleroma.User do
   end
 
   @doc "Fetch some posts when the user has just been federated with"
-  def fetch_initial_posts(user),
-    do: PleromaJobQueue.enqueue(:background, __MODULE__, [:fetch_initial_posts, user])
+  def fetch_initial_posts(user) do
+    %{"op" => "fetch_initial_posts", "user_id" => user.id}
+    |> BackgroundWorker.new(worker_args(:background))
+    |> Repo.insert()
+  end
 
   @spec get_followers_query(User.t(), pos_integer() | nil) :: Ecto.Query.t()
   def get_followers_query(%User{} = user, nil) do
@@ -1001,7 +1007,9 @@ defmodule Pleroma.User do
   end
 
   def deactivate_async(user, status \\ true) do
-    PleromaJobQueue.enqueue(:background, __MODULE__, [:deactivate_async, user, status])
+    %{"op" => "deactivate_user", "user_id" => user.id, "status" => status}
+    |> BackgroundWorker.new(worker_args(:background))
+    |> Repo.insert()
   end
 
   def deactivate(%User{} = user, status \\ true) do
@@ -1029,9 +1037,11 @@ defmodule Pleroma.User do
     |> update_and_set_cache()
   end
 
-  @spec delete(User.t()) :: :ok
-  def delete(%User{} = user),
-    do: PleromaJobQueue.enqueue(:background, __MODULE__, [:delete, user])
+  def delete(%User{} = user) do
+    %{"op" => "delete_user", "user_id" => user.id}
+    |> BackgroundWorker.new(worker_args(:background))
+    |> Repo.insert()
+  end
 
   @spec perform(atom(), User.t()) :: {:ok, User.t()}
   def perform(:delete, %User{} = user) do
@@ -1138,21 +1148,26 @@ defmodule Pleroma.User do
     Repo.all(query)
   end
 
-  def blocks_import(%User{} = blocker, blocked_identifiers) when is_list(blocked_identifiers),
-    do:
-      PleromaJobQueue.enqueue(:background, __MODULE__, [
-        :blocks_import,
-        blocker,
-        blocked_identifiers
-      ])
-
-  def follow_import(%User{} = follower, followed_identifiers) when is_list(followed_identifiers),
-    do:
-      PleromaJobQueue.enqueue(:background, __MODULE__, [
-        :follow_import,
-        follower,
-        followed_identifiers
-      ])
+  def blocks_import(%User{} = blocker, blocked_identifiers) when is_list(blocked_identifiers) do
+    %{
+      "op" => "blocks_import",
+      "blocker_id" => blocker.id,
+      "blocked_identifiers" => blocked_identifiers
+    }
+    |> BackgroundWorker.new(worker_args(:background))
+    |> Repo.insert()
+  end
+
+  def follow_import(%User{} = follower, followed_identifiers)
+      when is_list(followed_identifiers) do
+    %{
+      "op" => "follow_import",
+      "follower_id" => follower.id,
+      "followed_identifiers" => followed_identifiers
+    }
+    |> BackgroundWorker.new(worker_args(:background))
+    |> Repo.insert()
+  end
 
   def delete_user_activities(%User{ap_id: ap_id} = user) do
     ap_id
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 1a279a7df..8be8ac86f 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/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)
 
diff --git a/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex b/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex
index 01d21a299..1df3bb5b6 100644
--- a/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex
+++ b/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
diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
index 5403b71d8..0f117cd04 100644
--- a/lib/pleroma/web/activity_pub/transmogrifier.ex
+++ b/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}
diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
index bb9eadfee..d85fe824f 100644
--- a/lib/pleroma/web/federator/federator.ex
+++ b/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
diff --git a/lib/pleroma/web/oauth/token/clean_worker.ex b/lib/pleroma/web/oauth/token/clean_worker.ex
index dca852449..c0c9c3653 100644
--- a/lib/pleroma/web/oauth/token/clean_worker.ex
+++ b/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
 
diff --git a/lib/pleroma/web/push/push.ex b/lib/pleroma/web/push/push.ex
index 729dad02a..b4f0e5127 100644
--- a/lib/pleroma/web/push/push.ex
+++ b/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
diff --git a/lib/pleroma/web/twitter_api/controllers/util_controller.ex b/lib/pleroma/web/twitter_api/controllers/util_controller.ex
index 3405bd3b7..7ba4ad305 100644
--- a/lib/pleroma/web/twitter_api/controllers/util_controller.ex
+++ b/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
diff --git a/lib/pleroma/workers/background_worker.ex b/lib/pleroma/workers/background_worker.ex
new file mode 100644
index 000000000..3ab2b6bcc
--- /dev/null
+++ b/lib/pleroma/workers/background_worker.ex
@@ -0,0 +1,66 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.BackgroundWorker do
+  alias Pleroma.Activity
+  alias Pleroma.User
+  alias Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy
+  alias Pleroma.Web.OAuth.Token.CleanWorker
+
+  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  use Oban.Worker,
+    queue: "background",
+    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+
+  @impl Oban.Worker
+  def perform(%{"op" => "fetch_initial_posts", "user_id" => user_id}) do
+    user = User.get_by_id(user_id)
+    User.perform(:fetch_initial_posts, user)
+  end
+
+  def perform(%{"op" => "deactivate_user", "user_id" => user_id, "status" => status}) do
+    user = User.get_by_id(user_id)
+    User.perform(:deactivate_async, user, status)
+  end
+
+  def perform(%{"op" => "delete_user", "user_id" => user_id}) do
+    user = User.get_by_id(user_id)
+    User.perform(:delete, user)
+  end
+
+  def perform(%{
+        "op" => "blocks_import",
+        "blocker_id" => blocker_id,
+        "blocked_identifiers" => blocked_identifiers
+      }) do
+    blocker = User.get_by_id(blocker_id)
+    User.perform(:blocks_import, blocker, blocked_identifiers)
+  end
+
+  def perform(%{
+        "op" => "follow_import",
+        "follower_id" => follower_id,
+        "followed_identifiers" => followed_identifiers
+      }) do
+    follower = User.get_by_id(follower_id)
+    User.perform(:follow_import, follower, followed_identifiers)
+  end
+
+  def perform(%{"op" => "clean_expired_tokens"}) do
+    CleanWorker.perform(:clean)
+  end
+
+  def perform(%{"op" => "media_proxy_preload", "message" => message}) do
+    MediaProxyWarmingPolicy.perform(:preload, message)
+  end
+
+  def perform(%{"op" => "media_proxy_prefetch", "url" => url}) do
+    MediaProxyWarmingPolicy.perform(:prefetch, url)
+  end
+
+  def perform(%{"op" => "fetch_data_for_activity", "activity_id" => activity_id}) do
+    activity = Activity.get_by_id(activity_id)
+    Pleroma.Web.RichMedia.Helpers.perform(:fetch, activity)
+  end
+end
diff --git a/lib/pleroma/workers/helper.ex b/lib/pleroma/workers/helper.ex
new file mode 100644
index 000000000..3286ce0e8
--- /dev/null
+++ b/lib/pleroma/workers/helper.ex
@@ -0,0 +1,13 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Helper do
+  def worker_args(queue) do
+    if max_attempts = Pleroma.Config.get([:workers, :retries, queue]) do
+      [max_attempts: max_attempts]
+    else
+      []
+    end
+  end
+end
diff --git a/lib/pleroma/workers/mailer.ex b/lib/pleroma/workers/mailer.ex
new file mode 100644
index 000000000..da7fa6fd5
--- /dev/null
+++ b/lib/pleroma/workers/mailer.ex
@@ -0,0 +1,18 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Mailer do
+  alias Pleroma.User
+
+  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  use Oban.Worker,
+    queue: "mailer",
+    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+
+  @impl Oban.Worker
+  def perform(%{"op" => "digest_email", "user_id" => user_id}) do
+    user = User.get_by_id(user_id)
+    Pleroma.DigestEmailWorker.perform(user)
+  end
+end
diff --git a/lib/pleroma/workers/publisher.ex b/lib/pleroma/workers/publisher.ex
index 67871977a..c890ffb79 100644
--- a/lib/pleroma/workers/publisher.ex
+++ b/lib/pleroma/workers/publisher.ex
@@ -4,7 +4,7 @@
 
 defmodule Pleroma.Workers.Publisher do
   alias Pleroma.Activity
-  alias Pleroma.User
+  alias Pleroma.Web.Federator
 
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
@@ -13,23 +13,11 @@ defmodule Pleroma.Workers.Publisher do
 
   @impl Oban.Worker
   def perform(%{"op" => "publish", "activity_id" => activity_id}) do
-    with %Activity{} = activity <- Activity.get_by_id(activity_id) do
-      perform_publish(activity)
-    else
-      _ -> raise "Non-existing activity: #{activity_id}"
-    end
+    activity = Activity.get_by_id(activity_id)
+    Federator.perform(:publish, activity)
   end
 
   def perform(%{"op" => "publish_one", "module" => module_name, "params" => params}) do
-    module_name
-    |> String.to_atom()
-    |> apply(:publish_one, [params])
-  end
-
-  def perform_publish(%Activity{} = activity) do
-    with %User{} = actor <- User.get_cached_by_ap_id(activity.data["actor"]),
-         {:ok, actor} <- User.ensure_keys_present(actor) do
-      Pleroma.Web.Federator.Publisher.publish(actor, activity)
-    end
+    Federator.perform(:publish_one, String.to_atom(module_name), params)
   end
 end
diff --git a/lib/pleroma/workers/receiver.ex b/lib/pleroma/workers/receiver.ex
index 43558b4e6..d3de95716 100644
--- a/lib/pleroma/workers/receiver.ex
+++ b/lib/pleroma/workers/receiver.ex
@@ -3,15 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.Receiver 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.OStatus
-
-  require Logger
+  alias Pleroma.Web.Federator
 
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
@@ -20,42 +12,10 @@ defmodule Pleroma.Workers.Receiver do
 
   @impl Oban.Worker
   def perform(%{"op" => "incoming_doc", "body" => doc}) do
-    Logger.info("Got incoming document, trying to parse")
-    OStatus.handle_incoming(doc)
+    Federator.perform(:incoming_doc, doc)
   end
 
   def perform(%{"op" => "incoming_ap_doc", "params" => 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
-
-  defp 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
+    Federator.perform(:incoming_ap_doc, params)
   end
 end
diff --git a/lib/pleroma/workers/scheduled_activity_worker.ex b/lib/pleroma/workers/scheduled_activity_worker.ex
new file mode 100644
index 000000000..a49834fd8
--- /dev/null
+++ b/lib/pleroma/workers/scheduled_activity_worker.ex
@@ -0,0 +1,15 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.ScheduledActivityWorker do
+  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  use Oban.Worker,
+    queue: "scheduled_activities",
+    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+
+  @impl Oban.Worker
+  def perform(%{"op" => "execute", "activity_id" => activity_id}) do
+    Pleroma.ScheduledActivityWorker.perform(:execute, activity_id)
+  end
+end
diff --git a/lib/pleroma/workers/subscriber.ex b/lib/pleroma/workers/subscriber.ex
index a8c01bb10..6af3ad0a1 100644
--- a/lib/pleroma/workers/subscriber.ex
+++ b/lib/pleroma/workers/subscriber.ex
@@ -4,11 +4,9 @@
 
 defmodule Pleroma.Workers.Subscriber do
   alias Pleroma.Repo
-  alias Pleroma.Web.Websub
+  alias Pleroma.Web.Federator
   alias Pleroma.Web.Websub.WebsubClientSubscription
 
-  require Logger
-
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "federator_outgoing",
@@ -16,29 +14,16 @@ defmodule Pleroma.Workers.Subscriber do
 
   @impl Oban.Worker
   def perform(%{"op" => "refresh_subscriptions"}) do
-    Websub.refresh_subscriptions()
-    # Schedule the next run in 6 hours
-    Pleroma.Web.Federator.refresh_subscriptions(schedule_in: 3600 * 6)
+    Federator.perform(:refresh_subscriptions)
   end
 
   def perform(%{"op" => "request_subscription", "websub_id" => websub_id}) do
     websub = Repo.get(WebsubClientSubscription, websub_id)
-    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
+    Federator.perform(:request_subscription, websub)
   end
 
   def perform(%{"op" => "verify_websub", "websub_id" => websub_id}) do
     websub = Repo.get(WebsubClientSubscription, websub_id)
-
-    Logger.debug(fn ->
-      "Running WebSub verification for #{websub.id} (#{websub.topic}, #{websub.callback})"
-    end)
-
-    Websub.verify(websub)
+    Federator.perform(:verify_websub, websub)
   end
 end
diff --git a/lib/pleroma/workers/transmogrifier.ex b/lib/pleroma/workers/transmogrifier.ex
new file mode 100644
index 000000000..c6b4fab47
--- /dev/null
+++ b/lib/pleroma/workers/transmogrifier.ex
@@ -0,0 +1,18 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.Transmogrifier do
+  alias Pleroma.User
+
+  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  use Oban.Worker,
+    queue: "transmogrifier",
+    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+
+  @impl Oban.Worker
+  def perform(%{"op" => "user_upgrade", "user_id" => user_id}) do
+    user = User.get_by_id(user_id)
+    Pleroma.Web.ActivityPub.Transmogrifier.perform(:user_upgrade, user)
+  end
+end
diff --git a/lib/pleroma/workers/web_pusher.ex b/lib/pleroma/workers/web_pusher.ex
new file mode 100644
index 000000000..b99581eb0
--- /dev/null
+++ b/lib/pleroma/workers/web_pusher.ex
@@ -0,0 +1,19 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.WebPusher do
+  alias Pleroma.Notification
+  alias Pleroma.Repo
+
+  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  use Oban.Worker,
+    queue: "web_push",
+    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+
+  @impl Oban.Worker
+  def perform(%{"op" => "web_push", "notification_id" => notification_id}) do
+    notification = Repo.get(Notification, notification_id)
+    Pleroma.Web.Push.Impl.perform(notification)
+  end
+end
-- 
cgit v1.2.3


From a180c1360ecdbed76eccf3435bb2c831356746bc Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Wed, 14 Aug 2019 21:42:21 +0300
Subject: [#1149] Oban mailer job. Adjusted tests.

---
 lib/pleroma/application.ex    |  1 +
 lib/pleroma/emails/mailer.ex  | 13 ++++++++++++-
 lib/pleroma/workers/mailer.ex |  9 +++++++++
 3 files changed, 22 insertions(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 5550a4902..7cf60f44a 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -233,6 +233,7 @@ defmodule Pleroma.Application do
   defp after_supervisor_start do
     with digest_config <- Application.get_env(:pleroma, :email_notifications)[:digest],
          true <- digest_config[:active] do
+      # TODO: consider replacing with `quantum` scheduler
       PleromaJobQueue.schedule(
         digest_config[:schedule],
         :digest_emails,
diff --git a/lib/pleroma/emails/mailer.ex b/lib/pleroma/emails/mailer.ex
index 2e4657b7c..bb534f602 100644
--- a/lib/pleroma/emails/mailer.ex
+++ b/lib/pleroma/emails/mailer.ex
@@ -9,6 +9,8 @@ defmodule Pleroma.Emails.Mailer do
   The module contains functions to delivery email using Swoosh.Mailer.
   """
 
+  alias Pleroma.Repo
+  alias Pleroma.Workers.Mailer, as: MailerWorker
   alias Swoosh.DeliveryError
 
   @otp_app :pleroma
@@ -17,9 +19,18 @@ defmodule Pleroma.Emails.Mailer do
   @spec enabled?() :: boolean()
   def enabled?, do: Pleroma.Config.get([__MODULE__, :enabled])
 
+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
   @doc "add email to queue"
   def deliver_async(email, config \\ []) do
-    PleromaJobQueue.enqueue(:mailer, __MODULE__, [:deliver_async, email, config])
+    encoded_email =
+      email
+      |> :erlang.term_to_binary()
+      |> Base.encode64()
+
+    %{"op" => "email", "encoded_email" => encoded_email, "config" => config}
+    |> MailerWorker.new(worker_args(:mailer))
+    |> Repo.insert()
   end
 
   @doc "callback to perform send email from queue"
diff --git a/lib/pleroma/workers/mailer.ex b/lib/pleroma/workers/mailer.ex
index da7fa6fd5..8bf9952bc 100644
--- a/lib/pleroma/workers/mailer.ex
+++ b/lib/pleroma/workers/mailer.ex
@@ -11,6 +11,15 @@ defmodule Pleroma.Workers.Mailer do
     max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
 
   @impl Oban.Worker
+  def perform(%{"op" => "email", "encoded_email" => encoded_email, "config" => config}) do
+    email =
+      encoded_email
+      |> Base.decode64!()
+      |> :erlang.binary_to_term()
+
+    Pleroma.Emails.Mailer.deliver(email, config)
+  end
+
   def perform(%{"op" => "digest_email", "user_id" => user_id}) do
     user = User.get_by_id(user_id)
     Pleroma.DigestEmailWorker.perform(user)
-- 
cgit v1.2.3


From c29686309eaf2cdae039ce813755c0e23cdc4a03 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Fri, 23 Aug 2019 09:23:10 +0300
Subject: [#1149] Upgraded `oban` from 0.6.0 to 0.7.1.

---
 lib/pleroma/application.ex                       |  5 +--
 lib/pleroma/workers/background_worker.ex         | 42 ++++++++++++++----------
 lib/pleroma/workers/mailer.ex                    |  6 ++--
 lib/pleroma/workers/publisher.ex                 |  6 ++--
 lib/pleroma/workers/receiver.ex                  |  6 ++--
 lib/pleroma/workers/scheduled_activity_worker.ex |  4 +--
 lib/pleroma/workers/subscriber.ex                |  8 ++---
 lib/pleroma/workers/transmogrifier.ex            |  4 +--
 lib/pleroma/workers/web_pusher.ex                |  4 +--
 9 files changed, 44 insertions(+), 41 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 2e2922d28..384b03aa9 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -41,10 +41,7 @@ defmodule Pleroma.Application do
         hackney_pool_children() ++
         [
           Pleroma.Stats,
-          %{
-            id: Oban,
-            start: {Oban, :start_link, [Application.get_env(:pleroma, Oban)]}
-          },
+          {Oban, Application.get_env(:pleroma, Oban)},
           %{
             id: :web_push_init,
             start: {Task, :start_link, [&Pleroma.Web.Push.init/0]},
diff --git a/lib/pleroma/workers/background_worker.ex b/lib/pleroma/workers/background_worker.ex
index 3ab2b6bcc..3c021b9b4 100644
--- a/lib/pleroma/workers/background_worker.ex
+++ b/lib/pleroma/workers/background_worker.ex
@@ -11,55 +11,61 @@ defmodule Pleroma.Workers.BackgroundWorker do
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "background",
-    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+    max_attempts: 1
 
   @impl Oban.Worker
-  def perform(%{"op" => "fetch_initial_posts", "user_id" => user_id}) do
+  def perform(%{"op" => "fetch_initial_posts", "user_id" => user_id}, _job) do
     user = User.get_by_id(user_id)
     User.perform(:fetch_initial_posts, user)
   end
 
-  def perform(%{"op" => "deactivate_user", "user_id" => user_id, "status" => status}) do
+  def perform(%{"op" => "deactivate_user", "user_id" => user_id, "status" => status}, _job) do
     user = User.get_by_id(user_id)
     User.perform(:deactivate_async, user, status)
   end
 
-  def perform(%{"op" => "delete_user", "user_id" => user_id}) do
+  def perform(%{"op" => "delete_user", "user_id" => user_id}, _job) do
     user = User.get_by_id(user_id)
     User.perform(:delete, user)
   end
 
-  def perform(%{
-        "op" => "blocks_import",
-        "blocker_id" => blocker_id,
-        "blocked_identifiers" => blocked_identifiers
-      }) do
+  def perform(
+        %{
+          "op" => "blocks_import",
+          "blocker_id" => blocker_id,
+          "blocked_identifiers" => blocked_identifiers
+        },
+        _job
+      ) do
     blocker = User.get_by_id(blocker_id)
     User.perform(:blocks_import, blocker, blocked_identifiers)
   end
 
-  def perform(%{
-        "op" => "follow_import",
-        "follower_id" => follower_id,
-        "followed_identifiers" => followed_identifiers
-      }) do
+  def perform(
+        %{
+          "op" => "follow_import",
+          "follower_id" => follower_id,
+          "followed_identifiers" => followed_identifiers
+        },
+        _job
+      ) do
     follower = User.get_by_id(follower_id)
     User.perform(:follow_import, follower, followed_identifiers)
   end
 
-  def perform(%{"op" => "clean_expired_tokens"}) do
+  def perform(%{"op" => "clean_expired_tokens"}, _job) do
     CleanWorker.perform(:clean)
   end
 
-  def perform(%{"op" => "media_proxy_preload", "message" => message}) do
+  def perform(%{"op" => "media_proxy_preload", "message" => message}, _job) do
     MediaProxyWarmingPolicy.perform(:preload, message)
   end
 
-  def perform(%{"op" => "media_proxy_prefetch", "url" => url}) do
+  def perform(%{"op" => "media_proxy_prefetch", "url" => url}, _job) do
     MediaProxyWarmingPolicy.perform(:prefetch, url)
   end
 
-  def perform(%{"op" => "fetch_data_for_activity", "activity_id" => activity_id}) do
+  def perform(%{"op" => "fetch_data_for_activity", "activity_id" => activity_id}, _job) do
     activity = Activity.get_by_id(activity_id)
     Pleroma.Web.RichMedia.Helpers.perform(:fetch, activity)
   end
diff --git a/lib/pleroma/workers/mailer.ex b/lib/pleroma/workers/mailer.ex
index 8bf9952bc..1cce2ea03 100644
--- a/lib/pleroma/workers/mailer.ex
+++ b/lib/pleroma/workers/mailer.ex
@@ -8,10 +8,10 @@ defmodule Pleroma.Workers.Mailer do
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "mailer",
-    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+    max_attempts: 1
 
   @impl Oban.Worker
-  def perform(%{"op" => "email", "encoded_email" => encoded_email, "config" => config}) do
+  def perform(%{"op" => "email", "encoded_email" => encoded_email, "config" => config}, _job) do
     email =
       encoded_email
       |> Base.decode64!()
@@ -20,7 +20,7 @@ defmodule Pleroma.Workers.Mailer do
     Pleroma.Emails.Mailer.deliver(email, config)
   end
 
-  def perform(%{"op" => "digest_email", "user_id" => user_id}) do
+  def perform(%{"op" => "digest_email", "user_id" => user_id}, _job) do
     user = User.get_by_id(user_id)
     Pleroma.DigestEmailWorker.perform(user)
   end
diff --git a/lib/pleroma/workers/publisher.ex b/lib/pleroma/workers/publisher.ex
index c890ffb79..0a9084589 100644
--- a/lib/pleroma/workers/publisher.ex
+++ b/lib/pleroma/workers/publisher.ex
@@ -9,15 +9,15 @@ defmodule Pleroma.Workers.Publisher do
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "federator_outgoing",
-    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+    max_attempts: 1
 
   @impl Oban.Worker
-  def perform(%{"op" => "publish", "activity_id" => activity_id}) do
+  def perform(%{"op" => "publish", "activity_id" => activity_id}, _job) do
     activity = Activity.get_by_id(activity_id)
     Federator.perform(:publish, activity)
   end
 
-  def perform(%{"op" => "publish_one", "module" => module_name, "params" => params}) do
+  def perform(%{"op" => "publish_one", "module" => module_name, "params" => params}, _job) do
     Federator.perform(:publish_one, String.to_atom(module_name), params)
   end
 end
diff --git a/lib/pleroma/workers/receiver.ex b/lib/pleroma/workers/receiver.ex
index d3de95716..4ee270d74 100644
--- a/lib/pleroma/workers/receiver.ex
+++ b/lib/pleroma/workers/receiver.ex
@@ -8,14 +8,14 @@ defmodule Pleroma.Workers.Receiver do
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "federator_incoming",
-    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+    max_attempts: 1
 
   @impl Oban.Worker
-  def perform(%{"op" => "incoming_doc", "body" => doc}) do
+  def perform(%{"op" => "incoming_doc", "body" => doc}, _job) do
     Federator.perform(:incoming_doc, doc)
   end
 
-  def perform(%{"op" => "incoming_ap_doc", "params" => params}) do
+  def perform(%{"op" => "incoming_ap_doc", "params" => params}, _job) do
     Federator.perform(:incoming_ap_doc, params)
   end
 end
diff --git a/lib/pleroma/workers/scheduled_activity_worker.ex b/lib/pleroma/workers/scheduled_activity_worker.ex
index a49834fd8..d9724c78a 100644
--- a/lib/pleroma/workers/scheduled_activity_worker.ex
+++ b/lib/pleroma/workers/scheduled_activity_worker.ex
@@ -6,10 +6,10 @@ defmodule Pleroma.Workers.ScheduledActivityWorker do
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "scheduled_activities",
-    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+    max_attempts: 1
 
   @impl Oban.Worker
-  def perform(%{"op" => "execute", "activity_id" => activity_id}) do
+  def perform(%{"op" => "execute", "activity_id" => activity_id}, _job) do
     Pleroma.ScheduledActivityWorker.perform(:execute, activity_id)
   end
 end
diff --git a/lib/pleroma/workers/subscriber.ex b/lib/pleroma/workers/subscriber.ex
index 6af3ad0a1..783c44173 100644
--- a/lib/pleroma/workers/subscriber.ex
+++ b/lib/pleroma/workers/subscriber.ex
@@ -10,19 +10,19 @@ defmodule Pleroma.Workers.Subscriber do
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "federator_outgoing",
-    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+    max_attempts: 1
 
   @impl Oban.Worker
-  def perform(%{"op" => "refresh_subscriptions"}) do
+  def perform(%{"op" => "refresh_subscriptions"}, _job) do
     Federator.perform(:refresh_subscriptions)
   end
 
-  def perform(%{"op" => "request_subscription", "websub_id" => websub_id}) do
+  def perform(%{"op" => "request_subscription", "websub_id" => websub_id}, _job) do
     websub = Repo.get(WebsubClientSubscription, websub_id)
     Federator.perform(:request_subscription, websub)
   end
 
-  def perform(%{"op" => "verify_websub", "websub_id" => websub_id}) do
+  def perform(%{"op" => "verify_websub", "websub_id" => websub_id}, _job) do
     websub = Repo.get(WebsubClientSubscription, websub_id)
     Federator.perform(:verify_websub, websub)
   end
diff --git a/lib/pleroma/workers/transmogrifier.ex b/lib/pleroma/workers/transmogrifier.ex
index c6b4fab47..e13202c06 100644
--- a/lib/pleroma/workers/transmogrifier.ex
+++ b/lib/pleroma/workers/transmogrifier.ex
@@ -8,10 +8,10 @@ defmodule Pleroma.Workers.Transmogrifier do
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "transmogrifier",
-    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+    max_attempts: 1
 
   @impl Oban.Worker
-  def perform(%{"op" => "user_upgrade", "user_id" => user_id}) do
+  def perform(%{"op" => "user_upgrade", "user_id" => user_id}, _job) do
     user = User.get_by_id(user_id)
     Pleroma.Web.ActivityPub.Transmogrifier.perform(:user_upgrade, user)
   end
diff --git a/lib/pleroma/workers/web_pusher.ex b/lib/pleroma/workers/web_pusher.ex
index b99581eb0..7b78bb3ea 100644
--- a/lib/pleroma/workers/web_pusher.ex
+++ b/lib/pleroma/workers/web_pusher.ex
@@ -9,10 +9,10 @@ defmodule Pleroma.Workers.WebPusher do
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
     queue: "web_push",
-    max_attempts: Pleroma.Config.get([:workers, :retries, :compile_time_default])
+    max_attempts: 1
 
   @impl Oban.Worker
-  def perform(%{"op" => "web_push", "notification_id" => notification_id}) do
+  def perform(%{"op" => "web_push", "notification_id" => notification_id}, _job) do
     notification = Repo.get(Notification, notification_id)
     Pleroma.Web.Push.Impl.perform(notification)
   end
-- 
cgit v1.2.3


From c056736daaedb2a08557ee6c6a9bcb6bf44110ca Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Fri, 23 Aug 2019 16:11:39 +0300
Subject: [#1149] Publisher worker fix (atomized `params` keys).

---
 lib/pleroma/workers/publisher.ex | 1 +
 1 file changed, 1 insertion(+)

(limited to 'lib')

diff --git a/lib/pleroma/workers/publisher.ex b/lib/pleroma/workers/publisher.ex
index 0a9084589..00fae99c7 100644
--- a/lib/pleroma/workers/publisher.ex
+++ b/lib/pleroma/workers/publisher.ex
@@ -18,6 +18,7 @@ defmodule Pleroma.Workers.Publisher do
   end
 
   def perform(%{"op" => "publish_one", "module" => module_name, "params" => params}, _job) do
+    params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end)
     Federator.perform(:publish_one, String.to_atom(module_name), params)
   end
 end
-- 
cgit v1.2.3


From 581123f8bb703023cb652267a1fc34292f862852 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Fri, 23 Aug 2019 18:28:23 +0300
Subject: [#1149] Introduced `quantum` job scheduler. Documentation & config
 changes.

---
 lib/pleroma/application.ex             | 19 ++-----------------
 lib/pleroma/scheduler.ex               |  7 +++++++
 lib/pleroma/web/federator/federator.ex |  8 +-------
 3 files changed, 10 insertions(+), 24 deletions(-)
 create mode 100644 lib/pleroma/scheduler.ex

(limited to 'lib')

diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 384b03aa9..ce2d3ab59 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -31,6 +31,7 @@ defmodule Pleroma.Application do
     children =
       [
         Pleroma.Repo,
+        Pleroma.Scheduler,
         Pleroma.Config.TransferTask,
         Pleroma.Emoji,
         Pleroma.Captcha,
@@ -69,9 +70,7 @@ defmodule Pleroma.Application do
     # See http://elixir-lang.org/docs/stable/elixir/Supervisor.html
     # for other strategies and supported options
     opts = [strategy: :one_for_one, name: Pleroma.Supervisor]
-    result = Supervisor.start_link(children, opts)
-    :ok = after_supervisor_start()
-    result
+    Supervisor.start_link(children, opts)
   end
 
   defp setup_instrumenters do
@@ -162,18 +161,4 @@ defmodule Pleroma.Application do
       :hackney_pool.child_spec(pool, options)
     end
   end
-
-  defp after_supervisor_start do
-    with digest_config <- Application.get_env(:pleroma, :email_notifications)[:digest],
-         true <- digest_config[:active] do
-      # TODO: consider replacing with `quantum` scheduler
-      PleromaJobQueue.schedule(
-        digest_config[:schedule],
-        :digest_emails,
-        Pleroma.DigestEmailWorker
-      )
-    end
-
-    :ok
-  end
 end
diff --git a/lib/pleroma/scheduler.ex b/lib/pleroma/scheduler.ex
new file mode 100644
index 000000000..d84cd99ad
--- /dev/null
+++ b/lib/pleroma/scheduler.ex
@@ -0,0 +1,7 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Scheduler do
+  use Quantum.Scheduler, otp_app: :pleroma
+end
diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
index d85fe824f..cf7e50fee 100644
--- a/lib/pleroma/web/federator/federator.ex
+++ b/lib/pleroma/web/federator/federator.ex
@@ -21,7 +21,7 @@ defmodule Pleroma.Web.Federator do
   defdelegate worker_args(queue), to: Pleroma.Workers.Helper
 
   def init do
-    # 1 minute
+    # To do: consider removing this call in favor of scheduled execution (`quantum`-based)
     refresh_subscriptions(schedule_in: 60)
   end
 
@@ -146,12 +146,6 @@ defmodule Pleroma.Web.Federator do
   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
-- 
cgit v1.2.3


From cd78e63a2528ab813088d5e44a026f6bb05b344b Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Tue, 27 Aug 2019 14:34:37 +0300
Subject: [#1149] Bugfix: Pleroma.Workers.Subscriber / "verify_websub" works
 with WebsubServerSubscription.

---
 lib/pleroma/workers/subscriber.ex | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/workers/subscriber.ex b/lib/pleroma/workers/subscriber.ex
index 783c44173..e960b35bf 100644
--- a/lib/pleroma/workers/subscriber.ex
+++ b/lib/pleroma/workers/subscriber.ex
@@ -5,7 +5,7 @@
 defmodule Pleroma.Workers.Subscriber do
   alias Pleroma.Repo
   alias Pleroma.Web.Federator
-  alias Pleroma.Web.Websub.WebsubClientSubscription
+  alias Pleroma.Web.Websub
 
   # Note: `max_attempts` is intended to be overridden in `new/1` call
   use Oban.Worker,
@@ -18,12 +18,12 @@ defmodule Pleroma.Workers.Subscriber do
   end
 
   def perform(%{"op" => "request_subscription", "websub_id" => websub_id}, _job) do
-    websub = Repo.get(WebsubClientSubscription, websub_id)
+    websub = Repo.get(Websub.WebsubClientSubscription, websub_id)
     Federator.perform(:request_subscription, websub)
   end
 
   def perform(%{"op" => "verify_websub", "websub_id" => websub_id}, _job) do
-    websub = Repo.get(WebsubClientSubscription, websub_id)
+    websub = Repo.get(Websub.WebsubServerSubscription, websub_id)
     Federator.perform(:verify_websub, websub)
   end
 end
-- 
cgit v1.2.3


From a90ea8ba1562818b025f677ffeea35f7ca08ddf2 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Sat, 31 Aug 2019 19:08:56 +0300
Subject: [#1149] Addressed code review comments (code style, jobs pruning
 etc.).

---
 lib/pleroma/activity_expiration_worker.ex          |  6 ++---
 lib/pleroma/application.ex                         |  2 +-
 lib/pleroma/digest_email_worker.ex                 |  4 +--
 lib/pleroma/emails/mailer.ex                       |  4 +--
 lib/pleroma/scheduled_activity_worker.ex           |  2 +-
 lib/pleroma/user.ex                                |  2 +-
 lib/pleroma/web/activity_pub/activity_pub.ex       |  2 +-
 .../activity_pub/mrf/mediaproxy_warming_policy.ex  |  2 +-
 lib/pleroma/web/activity_pub/publisher.ex          |  2 +-
 lib/pleroma/web/activity_pub/transmogrifier.ex     |  4 +--
 lib/pleroma/web/federator/federator.ex             |  8 +++---
 lib/pleroma/web/federator/publisher.ex             |  9 ++-----
 lib/pleroma/web/oauth/token/clean_worker.ex        |  2 +-
 lib/pleroma/web/push/push.ex                       |  6 ++---
 lib/pleroma/web/salmon/salmon.ex                   |  2 +-
 lib/pleroma/workers/activity_expiration_worker.ex  | 21 ++++++++++++++++
 lib/pleroma/workers/background_worker.ex           | 19 +++++---------
 lib/pleroma/workers/helper.ex                      | 13 ----------
 lib/pleroma/workers/mailer.ex                      | 27 --------------------
 lib/pleroma/workers/mailer_worker.ex               | 26 +++++++++++++++++++
 lib/pleroma/workers/publisher.ex                   | 24 ------------------
 lib/pleroma/workers/publisher_worker.ex            | 28 +++++++++++++++++++++
 lib/pleroma/workers/receiver.ex                    | 21 ----------------
 lib/pleroma/workers/receiver_worker.ex             | 21 ++++++++++++++++
 lib/pleroma/workers/scheduled_activity_worker.ex   |  2 +-
 lib/pleroma/workers/subscriber.ex                  | 29 ----------------------
 lib/pleroma/workers/subscriber_worker.ex           | 29 ++++++++++++++++++++++
 lib/pleroma/workers/transmogrifier.ex              | 18 --------------
 lib/pleroma/workers/transmogrifier_worker.ex       | 18 ++++++++++++++
 lib/pleroma/workers/web_pusher.ex                  | 19 --------------
 lib/pleroma/workers/web_pusher_worker.ex           | 19 ++++++++++++++
 lib/pleroma/workers/worker_helper.ex               | 23 +++++++++++++++++
 32 files changed, 218 insertions(+), 196 deletions(-)
 create mode 100644 lib/pleroma/workers/activity_expiration_worker.ex
 delete mode 100644 lib/pleroma/workers/helper.ex
 delete mode 100644 lib/pleroma/workers/mailer.ex
 create mode 100644 lib/pleroma/workers/mailer_worker.ex
 delete mode 100644 lib/pleroma/workers/publisher.ex
 create mode 100644 lib/pleroma/workers/publisher_worker.ex
 delete mode 100644 lib/pleroma/workers/receiver.ex
 create mode 100644 lib/pleroma/workers/receiver_worker.ex
 delete mode 100644 lib/pleroma/workers/subscriber.ex
 create mode 100644 lib/pleroma/workers/subscriber_worker.ex
 delete mode 100644 lib/pleroma/workers/transmogrifier.ex
 create mode 100644 lib/pleroma/workers/transmogrifier_worker.ex
 delete mode 100644 lib/pleroma/workers/web_pusher.ex
 create mode 100644 lib/pleroma/workers/web_pusher_worker.ex
 create mode 100644 lib/pleroma/workers/worker_helper.ex

(limited to 'lib')

diff --git a/lib/pleroma/activity_expiration_worker.ex b/lib/pleroma/activity_expiration_worker.ex
index 5c0c53232..7aba7eece 100644
--- a/lib/pleroma/activity_expiration_worker.ex
+++ b/lib/pleroma/activity_expiration_worker.ex
@@ -9,13 +9,13 @@ defmodule Pleroma.ActivityExpirationWorker do
   alias Pleroma.Repo
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
-  alias Pleroma.Workers.BackgroundWorker
+  alias Pleroma.Workers.ActivityExpirationWorker
 
   require Logger
   use GenServer
   import Ecto.Query
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   @schedule_interval :timer.minutes(1)
 
@@ -57,7 +57,7 @@ defmodule Pleroma.ActivityExpirationWorker do
         "op" => "activity_expiration",
         "activity_expiration_id" => expiration.id
       }
-      |> BackgroundWorker.new(worker_args(:activity_expiration))
+      |> ActivityExpirationWorker.new(worker_args(:activity_expiration))
       |> Repo.insert()
     end)
 
diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 7d38ed5c4..f8f866dbd 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -43,7 +43,7 @@ defmodule Pleroma.Application do
         hackney_pool_children() ++
         [
           Pleroma.Stats,
-          {Oban, Application.get_env(:pleroma, Oban)},
+          {Oban, Pleroma.Config.get(Oban)},
           %{
             id: :web_push_init,
             start: {Task, :start_link, [&Pleroma.Web.Push.init/0]},
diff --git a/lib/pleroma/digest_email_worker.ex b/lib/pleroma/digest_email_worker.ex
index ffc48bfab..4ab2a4ef4 100644
--- a/lib/pleroma/digest_email_worker.ex
+++ b/lib/pleroma/digest_email_worker.ex
@@ -4,11 +4,11 @@
 
 defmodule Pleroma.DigestEmailWorker do
   alias Pleroma.Repo
-  alias Pleroma.Workers.Mailer, as: MailerWorker
+  alias Pleroma.Workers.MailerWorker
 
   import Ecto.Query
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   def perform do
     config = Pleroma.Config.get([:email_notifications, :digest])
diff --git a/lib/pleroma/emails/mailer.ex b/lib/pleroma/emails/mailer.ex
index bb534f602..9cbe7313c 100644
--- a/lib/pleroma/emails/mailer.ex
+++ b/lib/pleroma/emails/mailer.ex
@@ -10,7 +10,7 @@ defmodule Pleroma.Emails.Mailer do
   """
 
   alias Pleroma.Repo
-  alias Pleroma.Workers.Mailer, as: MailerWorker
+  alias Pleroma.Workers.MailerWorker
   alias Swoosh.DeliveryError
 
   @otp_app :pleroma
@@ -19,7 +19,7 @@ defmodule Pleroma.Emails.Mailer do
   @spec enabled?() :: boolean()
   def enabled?, do: Pleroma.Config.get([__MODULE__, :enabled])
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   @doc "add email to queue"
   def deliver_async(email, config \\ []) do
diff --git a/lib/pleroma/scheduled_activity_worker.ex b/lib/pleroma/scheduled_activity_worker.ex
index a01fb4fcb..8bf534f42 100644
--- a/lib/pleroma/scheduled_activity_worker.ex
+++ b/lib/pleroma/scheduled_activity_worker.ex
@@ -18,7 +18,7 @@ defmodule Pleroma.ScheduledActivityWorker do
 
   @schedule_interval :timer.minutes(1)
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   def start_link(_) do
     GenServer.start_link(__MODULE__, nil)
diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 18bba0fbb..abfa063fb 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -41,7 +41,7 @@ defmodule Pleroma.User do
   @strict_local_nickname_regex ~r/^[a-zA-Z\d]+$/
   @extended_local_nickname_regex ~r/^[a-zA-Z\d_-]+$/
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   schema "users" do
     field(:bio, :string)
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 50279cca5..74c5eb91c 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -26,7 +26,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   require Logger
   require Pleroma.Constants
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   # 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.
diff --git a/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex b/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex
index b188164ee..178321558 100644
--- a/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex
+++ b/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex
@@ -18,7 +18,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
     recv_timeout: 10_000
   ]
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   def perform(:prefetch, url) do
     Logger.info("Prefetching #{inspect(url)}")
diff --git a/lib/pleroma/web/activity_pub/publisher.ex b/lib/pleroma/web/activity_pub/publisher.ex
index 24d101dc8..a6322e25a 100644
--- a/lib/pleroma/web/activity_pub/publisher.ex
+++ b/lib/pleroma/web/activity_pub/publisher.ex
@@ -85,7 +85,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
   end
 
   def publish_one(%{actor_id: actor_id} = params) do
-    actor = User.get_by_id(actor_id)
+    actor = User.get_cached_by_id(actor_id)
 
     params
     |> Map.delete(:actor_id)
diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
index b068d28a7..9437f9a16 100644
--- a/lib/pleroma/web/activity_pub/transmogrifier.ex
+++ b/lib/pleroma/web/activity_pub/transmogrifier.ex
@@ -15,14 +15,14 @@ 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
+  alias Pleroma.Workers.TransmogrifierWorker
 
   import Ecto.Query
 
   require Logger
   require Pleroma.Constants
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   @doc """
   Modifies an incoming AP object (mastodon format) to our internal format.
diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
index cf7e50fee..8f43066e3 100644
--- a/lib/pleroma/web/federator/federator.ex
+++ b/lib/pleroma/web/federator/federator.ex
@@ -12,13 +12,13 @@ defmodule Pleroma.Web.Federator do
   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
+  alias Pleroma.Workers.PublisherWorker
+  alias Pleroma.Workers.ReceiverWorker
+  alias Pleroma.Workers.SubscriberWorker
 
   require Logger
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   def init do
     # To do: consider removing this call in favor of scheduled execution (`quantum`-based)
diff --git a/lib/pleroma/web/federator/publisher.ex b/lib/pleroma/web/federator/publisher.ex
index 05d2be615..42be109ab 100644
--- a/lib/pleroma/web/federator/publisher.ex
+++ b/lib/pleroma/web/federator/publisher.ex
@@ -6,7 +6,7 @@ defmodule Pleroma.Web.Federator.Publisher do
   alias Pleroma.Activity
   alias Pleroma.Config
   alias Pleroma.User
-  alias Pleroma.Workers.Publisher, as: PublisherWorker
+  alias Pleroma.Workers.PublisherWorker
 
   require Logger
 
@@ -31,12 +31,7 @@ defmodule Pleroma.Web.Federator.Publisher do
   """
   @spec enqueue_one(module(), Map.t()) :: :ok
   def enqueue_one(module, %{} = params) do
-    worker_args =
-      if max_attempts = Pleroma.Config.get([:workers, :retries, :federator_outgoing]) do
-        [max_attempts: max_attempts]
-      else
-        []
-      end
+    worker_args = Pleroma.Workers.WorkerHelper.worker_args(:federator_outgoing)
 
     %{"op" => "publish_one", "module" => to_string(module), "params" => params}
     |> PublisherWorker.new(worker_args)
diff --git a/lib/pleroma/web/oauth/token/clean_worker.ex b/lib/pleroma/web/oauth/token/clean_worker.ex
index 943e73289..b150a68a7 100644
--- a/lib/pleroma/web/oauth/token/clean_worker.ex
+++ b/lib/pleroma/web/oauth/token/clean_worker.ex
@@ -20,7 +20,7 @@ defmodule Pleroma.Web.OAuth.Token.CleanWorker do
   alias Pleroma.Web.OAuth.Token
   alias Pleroma.Workers.BackgroundWorker
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   def start_link(_), do: GenServer.start_link(__MODULE__, %{})
 
diff --git a/lib/pleroma/web/push/push.ex b/lib/pleroma/web/push/push.ex
index b4f0e5127..4973b529c 100644
--- a/lib/pleroma/web/push/push.ex
+++ b/lib/pleroma/web/push/push.ex
@@ -4,11 +4,11 @@
 
 defmodule Pleroma.Web.Push do
   alias Pleroma.Repo
-  alias Pleroma.Workers.WebPusher
+  alias Pleroma.Workers.WebPusherWorker
 
   require Logger
 
-  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
 
   def init do
     unless enabled() do
@@ -36,7 +36,7 @@ defmodule Pleroma.Web.Push do
 
   def send(notification) do
     %{"op" => "web_push", "notification_id" => notification.id}
-    |> WebPusher.new(worker_args(:web_push))
+    |> WebPusherWorker.new(worker_args(:web_push))
     |> Repo.insert()
   end
 end
diff --git a/lib/pleroma/web/salmon/salmon.ex b/lib/pleroma/web/salmon/salmon.ex
index bbaa293fd..8ba7380c0 100644
--- a/lib/pleroma/web/salmon/salmon.ex
+++ b/lib/pleroma/web/salmon/salmon.ex
@@ -171,7 +171,7 @@ defmodule Pleroma.Web.Salmon do
   end
 
   def publish_one(%{recipient_id: recipient_id} = params) do
-    recipient = User.get_by_id(recipient_id)
+    recipient = User.get_cached_by_id(recipient_id)
 
     params
     |> Map.delete(:recipient_id)
diff --git a/lib/pleroma/workers/activity_expiration_worker.ex b/lib/pleroma/workers/activity_expiration_worker.ex
new file mode 100644
index 000000000..0b491eabb
--- /dev/null
+++ b/lib/pleroma/workers/activity_expiration_worker.ex
@@ -0,0 +1,21 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.ActivityExpirationWorker do
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
+  use Oban.Worker,
+    queue: "activity_expiration",
+    max_attempts: 1
+
+  @impl Oban.Worker
+  def perform(
+        %{
+          "op" => "activity_expiration",
+          "activity_expiration_id" => activity_expiration_id
+        },
+        _job
+      ) do
+    Pleroma.ActivityExpirationWorker.perform(:execute, activity_expiration_id)
+  end
+end
diff --git a/lib/pleroma/workers/background_worker.ex b/lib/pleroma/workers/background_worker.ex
index fbce7d789..7b5575a5f 100644
--- a/lib/pleroma/workers/background_worker.ex
+++ b/lib/pleroma/workers/background_worker.ex
@@ -8,24 +8,24 @@ defmodule Pleroma.Workers.BackgroundWorker do
   alias Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy
   alias Pleroma.Web.OAuth.Token.CleanWorker
 
-  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
   use Oban.Worker,
     queue: "background",
     max_attempts: 1
 
   @impl Oban.Worker
   def perform(%{"op" => "fetch_initial_posts", "user_id" => user_id}, _job) do
-    user = User.get_by_id(user_id)
+    user = User.get_cached_by_id(user_id)
     User.perform(:fetch_initial_posts, user)
   end
 
   def perform(%{"op" => "deactivate_user", "user_id" => user_id, "status" => status}, _job) do
-    user = User.get_by_id(user_id)
+    user = User.get_cached_by_id(user_id)
     User.perform(:deactivate_async, user, status)
   end
 
   def perform(%{"op" => "delete_user", "user_id" => user_id}, _job) do
-    user = User.get_by_id(user_id)
+    user = User.get_cached_by_id(user_id)
     User.perform(:delete, user)
   end
 
@@ -37,7 +37,7 @@ defmodule Pleroma.Workers.BackgroundWorker do
         },
         _job
       ) do
-    blocker = User.get_by_id(blocker_id)
+    blocker = User.get_cached_by_id(blocker_id)
     User.perform(:blocks_import, blocker, blocked_identifiers)
   end
 
@@ -49,7 +49,7 @@ defmodule Pleroma.Workers.BackgroundWorker do
         },
         _job
       ) do
-    follower = User.get_by_id(follower_id)
+    follower = User.get_cached_by_id(follower_id)
     User.perform(:follow_import, follower, followed_identifiers)
   end
 
@@ -69,11 +69,4 @@ defmodule Pleroma.Workers.BackgroundWorker do
     activity = Activity.get_by_id(activity_id)
     Pleroma.Web.RichMedia.Helpers.perform(:fetch, activity)
   end
-
-  def perform(
-        %{"op" => "activity_expiration", "activity_expiration_id" => activity_expiration_id},
-        _job
-      ) do
-    Pleroma.ActivityExpirationWorker.perform(:execute, activity_expiration_id)
-  end
 end
diff --git a/lib/pleroma/workers/helper.ex b/lib/pleroma/workers/helper.ex
deleted file mode 100644
index 3286ce0e8..000000000
--- a/lib/pleroma/workers/helper.ex
+++ /dev/null
@@ -1,13 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.Helper do
-  def worker_args(queue) do
-    if max_attempts = Pleroma.Config.get([:workers, :retries, queue]) do
-      [max_attempts: max_attempts]
-    else
-      []
-    end
-  end
-end
diff --git a/lib/pleroma/workers/mailer.ex b/lib/pleroma/workers/mailer.ex
deleted file mode 100644
index 1cce2ea03..000000000
--- a/lib/pleroma/workers/mailer.ex
+++ /dev/null
@@ -1,27 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.Mailer do
-  alias Pleroma.User
-
-  # Note: `max_attempts` is intended to be overridden in `new/1` call
-  use Oban.Worker,
-    queue: "mailer",
-    max_attempts: 1
-
-  @impl Oban.Worker
-  def perform(%{"op" => "email", "encoded_email" => encoded_email, "config" => config}, _job) do
-    email =
-      encoded_email
-      |> Base.decode64!()
-      |> :erlang.binary_to_term()
-
-    Pleroma.Emails.Mailer.deliver(email, config)
-  end
-
-  def perform(%{"op" => "digest_email", "user_id" => user_id}, _job) do
-    user = User.get_by_id(user_id)
-    Pleroma.DigestEmailWorker.perform(user)
-  end
-end
diff --git a/lib/pleroma/workers/mailer_worker.ex b/lib/pleroma/workers/mailer_worker.ex
new file mode 100644
index 000000000..4f73d61bc
--- /dev/null
+++ b/lib/pleroma/workers/mailer_worker.ex
@@ -0,0 +1,26 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.MailerWorker do
+  alias Pleroma.User
+
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
+  use Oban.Worker,
+    queue: "mailer",
+    max_attempts: 1
+
+  @impl Oban.Worker
+  def perform(%{"op" => "email", "encoded_email" => encoded_email, "config" => config}, _job) do
+    encoded_email
+    |> Base.decode64!()
+    |> :erlang.binary_to_term()
+    |> Pleroma.Emails.Mailer.deliver(config)
+  end
+
+  def perform(%{"op" => "digest_email", "user_id" => user_id}, _job) do
+    user_id
+    |> User.get_cached_by_id()
+    |> Pleroma.DigestEmailWorker.perform()
+  end
+end
diff --git a/lib/pleroma/workers/publisher.ex b/lib/pleroma/workers/publisher.ex
deleted file mode 100644
index 00fae99c7..000000000
--- a/lib/pleroma/workers/publisher.ex
+++ /dev/null
@@ -1,24 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.Publisher do
-  alias Pleroma.Activity
-  alias Pleroma.Web.Federator
-
-  # Note: `max_attempts` is intended to be overridden in `new/1` call
-  use Oban.Worker,
-    queue: "federator_outgoing",
-    max_attempts: 1
-
-  @impl Oban.Worker
-  def perform(%{"op" => "publish", "activity_id" => activity_id}, _job) do
-    activity = Activity.get_by_id(activity_id)
-    Federator.perform(:publish, activity)
-  end
-
-  def perform(%{"op" => "publish_one", "module" => module_name, "params" => params}, _job) do
-    params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end)
-    Federator.perform(:publish_one, String.to_atom(module_name), params)
-  end
-end
diff --git a/lib/pleroma/workers/publisher_worker.ex b/lib/pleroma/workers/publisher_worker.ex
new file mode 100644
index 000000000..5671d2a29
--- /dev/null
+++ b/lib/pleroma/workers/publisher_worker.ex
@@ -0,0 +1,28 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.PublisherWorker do
+  alias Pleroma.Activity
+  alias Pleroma.Web.Federator
+
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
+  use Oban.Worker,
+    queue: "federator_outgoing",
+    max_attempts: 1
+
+  def backoff(attempt) when is_integer(attempt) do
+    Pleroma.Workers.WorkerHelper.sidekiq_backoff(attempt, 5)
+  end
+
+  @impl Oban.Worker
+  def perform(%{"op" => "publish", "activity_id" => activity_id}, _job) do
+    activity = Activity.get_by_id(activity_id)
+    Federator.perform(:publish, activity)
+  end
+
+  def perform(%{"op" => "publish_one", "module" => module_name, "params" => params}, _job) do
+    params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end)
+    Federator.perform(:publish_one, String.to_atom(module_name), params)
+  end
+end
diff --git a/lib/pleroma/workers/receiver.ex b/lib/pleroma/workers/receiver.ex
deleted file mode 100644
index 4ee270d74..000000000
--- a/lib/pleroma/workers/receiver.ex
+++ /dev/null
@@ -1,21 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.Receiver do
-  alias Pleroma.Web.Federator
-
-  # Note: `max_attempts` is intended to be overridden in `new/1` call
-  use Oban.Worker,
-    queue: "federator_incoming",
-    max_attempts: 1
-
-  @impl Oban.Worker
-  def perform(%{"op" => "incoming_doc", "body" => doc}, _job) do
-    Federator.perform(:incoming_doc, doc)
-  end
-
-  def perform(%{"op" => "incoming_ap_doc", "params" => params}, _job) do
-    Federator.perform(:incoming_ap_doc, params)
-  end
-end
diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex
new file mode 100644
index 000000000..cdce630f2
--- /dev/null
+++ b/lib/pleroma/workers/receiver_worker.ex
@@ -0,0 +1,21 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.ReceiverWorker do
+  alias Pleroma.Web.Federator
+
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
+  use Oban.Worker,
+    queue: "federator_incoming",
+    max_attempts: 1
+
+  @impl Oban.Worker
+  def perform(%{"op" => "incoming_doc", "body" => doc}, _job) do
+    Federator.perform(:incoming_doc, doc)
+  end
+
+  def perform(%{"op" => "incoming_ap_doc", "params" => params}, _job) do
+    Federator.perform(:incoming_ap_doc, params)
+  end
+end
diff --git a/lib/pleroma/workers/scheduled_activity_worker.ex b/lib/pleroma/workers/scheduled_activity_worker.ex
index d9724c78a..4094411ae 100644
--- a/lib/pleroma/workers/scheduled_activity_worker.ex
+++ b/lib/pleroma/workers/scheduled_activity_worker.ex
@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.ScheduledActivityWorker do
-  # Note: `max_attempts` is intended to be overridden in `new/1` call
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
   use Oban.Worker,
     queue: "scheduled_activities",
     max_attempts: 1
diff --git a/lib/pleroma/workers/subscriber.ex b/lib/pleroma/workers/subscriber.ex
deleted file mode 100644
index e960b35bf..000000000
--- a/lib/pleroma/workers/subscriber.ex
+++ /dev/null
@@ -1,29 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.Subscriber do
-  alias Pleroma.Repo
-  alias Pleroma.Web.Federator
-  alias Pleroma.Web.Websub
-
-  # Note: `max_attempts` is intended to be overridden in `new/1` call
-  use Oban.Worker,
-    queue: "federator_outgoing",
-    max_attempts: 1
-
-  @impl Oban.Worker
-  def perform(%{"op" => "refresh_subscriptions"}, _job) do
-    Federator.perform(:refresh_subscriptions)
-  end
-
-  def perform(%{"op" => "request_subscription", "websub_id" => websub_id}, _job) do
-    websub = Repo.get(Websub.WebsubClientSubscription, websub_id)
-    Federator.perform(:request_subscription, websub)
-  end
-
-  def perform(%{"op" => "verify_websub", "websub_id" => websub_id}, _job) do
-    websub = Repo.get(Websub.WebsubServerSubscription, websub_id)
-    Federator.perform(:verify_websub, websub)
-  end
-end
diff --git a/lib/pleroma/workers/subscriber_worker.ex b/lib/pleroma/workers/subscriber_worker.ex
new file mode 100644
index 000000000..22d1dc956
--- /dev/null
+++ b/lib/pleroma/workers/subscriber_worker.ex
@@ -0,0 +1,29 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.SubscriberWorker do
+  alias Pleroma.Repo
+  alias Pleroma.Web.Federator
+  alias Pleroma.Web.Websub
+
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
+  use Oban.Worker,
+    queue: "federator_outgoing",
+    max_attempts: 1
+
+  @impl Oban.Worker
+  def perform(%{"op" => "refresh_subscriptions"}, _job) do
+    Federator.perform(:refresh_subscriptions)
+  end
+
+  def perform(%{"op" => "request_subscription", "websub_id" => websub_id}, _job) do
+    websub = Repo.get(Websub.WebsubClientSubscription, websub_id)
+    Federator.perform(:request_subscription, websub)
+  end
+
+  def perform(%{"op" => "verify_websub", "websub_id" => websub_id}, _job) do
+    websub = Repo.get(Websub.WebsubServerSubscription, websub_id)
+    Federator.perform(:verify_websub, websub)
+  end
+end
diff --git a/lib/pleroma/workers/transmogrifier.ex b/lib/pleroma/workers/transmogrifier.ex
deleted file mode 100644
index e13202c06..000000000
--- a/lib/pleroma/workers/transmogrifier.ex
+++ /dev/null
@@ -1,18 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.Transmogrifier do
-  alias Pleroma.User
-
-  # Note: `max_attempts` is intended to be overridden in `new/1` call
-  use Oban.Worker,
-    queue: "transmogrifier",
-    max_attempts: 1
-
-  @impl Oban.Worker
-  def perform(%{"op" => "user_upgrade", "user_id" => user_id}, _job) do
-    user = User.get_by_id(user_id)
-    Pleroma.Web.ActivityPub.Transmogrifier.perform(:user_upgrade, user)
-  end
-end
diff --git a/lib/pleroma/workers/transmogrifier_worker.ex b/lib/pleroma/workers/transmogrifier_worker.ex
new file mode 100644
index 000000000..6f5c1a2f2
--- /dev/null
+++ b/lib/pleroma/workers/transmogrifier_worker.ex
@@ -0,0 +1,18 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.TransmogrifierWorker do
+  alias Pleroma.User
+
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
+  use Oban.Worker,
+    queue: "transmogrifier",
+    max_attempts: 1
+
+  @impl Oban.Worker
+  def perform(%{"op" => "user_upgrade", "user_id" => user_id}, _job) do
+    user = User.get_cached_by_id(user_id)
+    Pleroma.Web.ActivityPub.Transmogrifier.perform(:user_upgrade, user)
+  end
+end
diff --git a/lib/pleroma/workers/web_pusher.ex b/lib/pleroma/workers/web_pusher.ex
deleted file mode 100644
index 7b78bb3ea..000000000
--- a/lib/pleroma/workers/web_pusher.ex
+++ /dev/null
@@ -1,19 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Workers.WebPusher do
-  alias Pleroma.Notification
-  alias Pleroma.Repo
-
-  # Note: `max_attempts` is intended to be overridden in `new/1` call
-  use Oban.Worker,
-    queue: "web_push",
-    max_attempts: 1
-
-  @impl Oban.Worker
-  def perform(%{"op" => "web_push", "notification_id" => notification_id}, _job) do
-    notification = Repo.get(Notification, notification_id)
-    Pleroma.Web.Push.Impl.perform(notification)
-  end
-end
diff --git a/lib/pleroma/workers/web_pusher_worker.ex b/lib/pleroma/workers/web_pusher_worker.ex
new file mode 100644
index 000000000..2b1d3b99a
--- /dev/null
+++ b/lib/pleroma/workers/web_pusher_worker.ex
@@ -0,0 +1,19 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.WebPusherWorker do
+  alias Pleroma.Notification
+  alias Pleroma.Repo
+
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
+  use Oban.Worker,
+    queue: "web_push",
+    max_attempts: 1
+
+  @impl Oban.Worker
+  def perform(%{"op" => "web_push", "notification_id" => notification_id}, _job) do
+    notification = Repo.get(Notification, notification_id)
+    Pleroma.Web.Push.Impl.perform(notification)
+  end
+end
diff --git a/lib/pleroma/workers/worker_helper.ex b/lib/pleroma/workers/worker_helper.ex
new file mode 100644
index 000000000..f9ed2e64d
--- /dev/null
+++ b/lib/pleroma/workers/worker_helper.ex
@@ -0,0 +1,23 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.WorkerHelper do
+  alias Pleroma.Config
+
+  def worker_args(queue) do
+    case Config.get([:workers, :retries, queue]) do
+      nil -> []
+      max_attempts -> [max_attempts: max_attempts]
+    end
+  end
+
+  def sidekiq_backoff(attempt, pow \\ 4, base_backoff \\ 15) do
+    backoff =
+      :math.pow(attempt, pow) +
+        base_backoff +
+        :rand.uniform(2 * base_backoff) * attempt
+
+    trunc(backoff)
+  end
+end
-- 
cgit v1.2.3


From dd017c65a4b86501c435f5cb01804300e6b7c6dd Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Sat, 31 Aug 2019 21:58:42 +0300
Subject: [#1149] Refactored Oban workers API (introduced `enqueue/3`).

---
 lib/pleroma/activity_expiration_worker.ex          | 13 ++++------
 lib/pleroma/digest_email_worker.ex                 | 10 +++-----
 lib/pleroma/emails/mailer.ex                       |  7 +-----
 lib/pleroma/scheduled_activity_worker.ex           | 10 ++++----
 lib/pleroma/user.ex                                | 28 ++++++----------------
 lib/pleroma/web/activity_pub/activity_pub.ex       |  6 +----
 .../activity_pub/mrf/mediaproxy_warming_policy.ex  | 11 ++-------
 lib/pleroma/web/activity_pub/transmogrifier.ex     |  6 +----
 lib/pleroma/web/federator/federator.ex             | 26 +++++---------------
 lib/pleroma/web/federator/publisher.ex             |  9 ++++---
 lib/pleroma/web/oauth/token/clean_worker.ex        |  7 +-----
 lib/pleroma/web/push/push.ex                       |  7 +-----
 lib/pleroma/workers/activity_expiration_worker.ex  |  2 ++
 lib/pleroma/workers/background_worker.ex           |  2 ++
 lib/pleroma/workers/digest_emails_worker.ex        | 21 ++++++++++++++++
 lib/pleroma/workers/mailer_worker.ex               | 10 ++------
 lib/pleroma/workers/publisher_worker.ex            |  2 ++
 lib/pleroma/workers/receiver_worker.ex             |  2 ++
 lib/pleroma/workers/scheduled_activity_worker.ex   |  2 ++
 lib/pleroma/workers/subscriber_worker.ex           |  2 ++
 lib/pleroma/workers/transmogrifier_worker.ex       |  2 ++
 lib/pleroma/workers/web_pusher_worker.ex           |  2 ++
 lib/pleroma/workers/worker_helper.ex               | 18 ++++++++++++++
 23 files changed, 92 insertions(+), 113 deletions(-)
 create mode 100644 lib/pleroma/workers/digest_emails_worker.ex

(limited to 'lib')

diff --git a/lib/pleroma/activity_expiration_worker.ex b/lib/pleroma/activity_expiration_worker.ex
index 7aba7eece..c0820c202 100644
--- a/lib/pleroma/activity_expiration_worker.ex
+++ b/lib/pleroma/activity_expiration_worker.ex
@@ -9,14 +9,11 @@ defmodule Pleroma.ActivityExpirationWorker do
   alias Pleroma.Repo
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
-  alias Pleroma.Workers.ActivityExpirationWorker
 
   require Logger
   use GenServer
   import Ecto.Query
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   @schedule_interval :timer.minutes(1)
 
   def start_link(_) do
@@ -53,12 +50,10 @@ defmodule Pleroma.ActivityExpirationWorker do
   def handle_info(:perform, state) do
     ActivityExpiration.due_expirations(@schedule_interval)
     |> Enum.each(fn expiration ->
-      %{
-        "op" => "activity_expiration",
-        "activity_expiration_id" => expiration.id
-      }
-      |> ActivityExpirationWorker.new(worker_args(:activity_expiration))
-      |> Repo.insert()
+      Pleroma.Workers.ActivityExpirationWorker.enqueue(
+        "activity_expiration",
+        %{"activity_expiration_id" => expiration.id}
+      )
     end)
 
     schedule_next()
diff --git a/lib/pleroma/digest_email_worker.ex b/lib/pleroma/digest_email_worker.ex
index 4ab2a4ef4..5be7cf26b 100644
--- a/lib/pleroma/digest_email_worker.ex
+++ b/lib/pleroma/digest_email_worker.ex
@@ -4,12 +4,10 @@
 
 defmodule Pleroma.DigestEmailWorker do
   alias Pleroma.Repo
-  alias Pleroma.Workers.MailerWorker
+  alias Pleroma.Workers.DigestEmailsWorker
 
   import Ecto.Query
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   def perform do
     config = Pleroma.Config.get([:email_notifications, :digest])
     negative_interval = -Map.fetch!(config, :interval)
@@ -23,11 +21,9 @@ defmodule Pleroma.DigestEmailWorker do
       where: u.last_digest_emailed_at < datetime_add(^now, ^negative_interval, "day"),
       select: u
     )
-    |> Pleroma.Repo.all()
+    |> Repo.all()
     |> Enum.each(fn user ->
-      %{"op" => "digest_email", "user_id" => user.id}
-      |> MailerWorker.new([queue: "digest_emails"] ++ worker_args(:digest_emails))
-      |> Repo.insert()
+      DigestEmailsWorker.enqueue("digest_email", %{"user_id" => user.id})
     end)
   end
 
diff --git a/lib/pleroma/emails/mailer.ex b/lib/pleroma/emails/mailer.ex
index 9cbe7313c..eb96f2e8b 100644
--- a/lib/pleroma/emails/mailer.ex
+++ b/lib/pleroma/emails/mailer.ex
@@ -9,7 +9,6 @@ defmodule Pleroma.Emails.Mailer do
   The module contains functions to delivery email using Swoosh.Mailer.
   """
 
-  alias Pleroma.Repo
   alias Pleroma.Workers.MailerWorker
   alias Swoosh.DeliveryError
 
@@ -19,8 +18,6 @@ defmodule Pleroma.Emails.Mailer do
   @spec enabled?() :: boolean()
   def enabled?, do: Pleroma.Config.get([__MODULE__, :enabled])
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   @doc "add email to queue"
   def deliver_async(email, config \\ []) do
     encoded_email =
@@ -28,9 +25,7 @@ defmodule Pleroma.Emails.Mailer do
       |> :erlang.term_to_binary()
       |> Base.encode64()
 
-    %{"op" => "email", "encoded_email" => encoded_email, "config" => config}
-    |> MailerWorker.new(worker_args(:mailer))
-    |> Repo.insert()
+    MailerWorker.enqueue("email", %{"encoded_email" => encoded_email, "config" => config})
   end
 
   @doc "callback to perform send email from queue"
diff --git a/lib/pleroma/scheduled_activity_worker.ex b/lib/pleroma/scheduled_activity_worker.ex
index 8bf534f42..c41a542de 100644
--- a/lib/pleroma/scheduled_activity_worker.ex
+++ b/lib/pleroma/scheduled_activity_worker.ex
@@ -8,7 +8,6 @@ defmodule Pleroma.ScheduledActivityWorker do
   """
 
   alias Pleroma.Config
-  alias Pleroma.Repo
   alias Pleroma.ScheduledActivity
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
@@ -18,8 +17,6 @@ defmodule Pleroma.ScheduledActivityWorker do
 
   @schedule_interval :timer.minutes(1)
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   def start_link(_) do
     GenServer.start_link(__MODULE__, nil)
   end
@@ -49,9 +46,10 @@ defmodule Pleroma.ScheduledActivityWorker do
   def handle_info(:perform, state) do
     ScheduledActivity.due_activities(@schedule_interval)
     |> Enum.each(fn scheduled_activity ->
-      %{"op" => "execute", "activity_id" => scheduled_activity.id}
-      |> Pleroma.Workers.ScheduledActivityWorker.new(worker_args(:scheduled_activities))
-      |> Repo.insert()
+      Pleroma.Workers.ScheduledActivityWorker.enqueue(
+        "execute",
+        %{"activity_id" => scheduled_activity.id}
+      )
     end)
 
     schedule_next()
diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index abfa063fb..2fe7e1748 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -41,8 +41,6 @@ defmodule Pleroma.User do
   @strict_local_nickname_regex ~r/^[a-zA-Z\d]+$/
   @extended_local_nickname_regex ~r/^[a-zA-Z\d_-]+$/
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   schema "users" do
     field(:bio, :string)
     field(:email, :string)
@@ -623,9 +621,7 @@ defmodule Pleroma.User do
 
   @doc "Fetch some posts when the user has just been federated with"
   def fetch_initial_posts(user) do
-    %{"op" => "fetch_initial_posts", "user_id" => user.id}
-    |> BackgroundWorker.new(worker_args(:background))
-    |> Repo.insert()
+    BackgroundWorker.enqueue("fetch_initial_posts", %{"user_id" => user.id})
   end
 
   @spec get_followers_query(User.t(), pos_integer() | nil) :: Ecto.Query.t()
@@ -1056,9 +1052,7 @@ defmodule Pleroma.User do
   end
 
   def deactivate_async(user, status \\ true) do
-    %{"op" => "deactivate_user", "user_id" => user.id, "status" => status}
-    |> BackgroundWorker.new(worker_args(:background))
-    |> Repo.insert()
+    BackgroundWorker.enqueue("deactivate_user", %{"user_id" => user.id, "status" => status})
   end
 
   def deactivate(%User{} = user, status \\ true) do
@@ -1087,9 +1081,7 @@ defmodule Pleroma.User do
   end
 
   def delete(%User{} = user) do
-    %{"op" => "delete_user", "user_id" => user.id}
-    |> BackgroundWorker.new(worker_args(:background))
-    |> Repo.insert()
+    BackgroundWorker.enqueue("delete_user", %{"user_id" => user.id})
   end
 
   @spec perform(atom(), User.t()) :: {:ok, User.t()}
@@ -1198,24 +1190,18 @@ defmodule Pleroma.User do
   end
 
   def blocks_import(%User{} = blocker, blocked_identifiers) when is_list(blocked_identifiers) do
-    %{
-      "op" => "blocks_import",
+    BackgroundWorker.enqueue("blocks_import", %{
       "blocker_id" => blocker.id,
       "blocked_identifiers" => blocked_identifiers
-    }
-    |> BackgroundWorker.new(worker_args(:background))
-    |> Repo.insert()
+    })
   end
 
   def follow_import(%User{} = follower, followed_identifiers)
       when is_list(followed_identifiers) do
-    %{
-      "op" => "follow_import",
+    BackgroundWorker.enqueue("follow_import", %{
       "follower_id" => follower.id,
       "followed_identifiers" => followed_identifiers
-    }
-    |> BackgroundWorker.new(worker_args(:background))
-    |> Repo.insert()
+    })
   end
 
   def delete_user_activities(%User{ap_id: ap_id} = user) do
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 74c5eb91c..90b409606 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -26,8 +26,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   require Logger
   require Pleroma.Constants
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   # 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
@@ -148,9 +146,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
           activity
         end
 
-      %{"op" => "fetch_data_for_activity", "activity_id" => activity.id}
-      |> BackgroundWorker.new(worker_args(:background))
-      |> Repo.insert()
+      BackgroundWorker.enqueue("fetch_data_for_activity", %{"activity_id" => activity.id})
 
       Notification.create_notifications(activity)
 
diff --git a/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex b/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex
index 178321558..26b8539fe 100644
--- a/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex
+++ b/lib/pleroma/web/activity_pub/mrf/mediaproxy_warming_policy.ex
@@ -7,7 +7,6 @@ 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
 
@@ -18,8 +17,6 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
     recv_timeout: 10_000
   ]
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   def perform(:prefetch, url) do
     Logger.info("Prefetching #{inspect(url)}")
 
@@ -34,9 +31,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
         url
         |> Enum.each(fn
           %{"href" => href} ->
-            %{"op" => "media_proxy_prefetch", "url" => href}
-            |> BackgroundWorker.new(worker_args(:background))
-            |> Repo.insert()
+            BackgroundWorker.enqueue("media_proxy_prefetch", %{"url" => href})
 
           x ->
             Logger.debug("Unhandled attachment URL object #{inspect(x)}")
@@ -52,9 +47,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
         %{"type" => "Create", "object" => %{"attachment" => attachments} = _object} = message
       )
       when is_list(attachments) and length(attachments) > 0 do
-    %{"op" => "media_proxy_preload", "message" => message}
-    |> BackgroundWorker.new(worker_args(:background))
-    |> Repo.insert()
+    BackgroundWorker.enqueue("media_proxy_preload", %{"message" => message})
 
     {:ok, message}
   end
diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
index 9437f9a16..f27455e8b 100644
--- a/lib/pleroma/web/activity_pub/transmogrifier.ex
+++ b/lib/pleroma/web/activity_pub/transmogrifier.ex
@@ -22,8 +22,6 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
   require Logger
   require Pleroma.Constants
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   @doc """
   Modifies an incoming AP object (mastodon format) to our internal format.
   """
@@ -1054,9 +1052,7 @@ 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
-        %{"op" => "user_upgrade", "user_id" => user.id}
-        |> TransmogrifierWorker.new(worker_args(:transmogrifier))
-        |> Repo.insert()
+        TransmogrifierWorker.enqueue("user_upgrade", %{"user_id" => user.id})
       end
 
       {:ok, user}
diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
index 8f43066e3..1a2da014a 100644
--- a/lib/pleroma/web/federator/federator.ex
+++ b/lib/pleroma/web/federator/federator.ex
@@ -18,8 +18,6 @@ defmodule Pleroma.Web.Federator do
 
   require Logger
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   def init do
     # To do: consider removing this call in favor of scheduled execution (`quantum`-based)
     refresh_subscriptions(schedule_in: 60)
@@ -40,15 +38,11 @@ defmodule Pleroma.Web.Federator do
   # Client API
 
   def incoming_doc(doc) do
-    %{"op" => "incoming_doc", "body" => doc}
-    |> ReceiverWorker.new(worker_args(:federator_incoming))
-    |> Pleroma.Repo.insert()
+    ReceiverWorker.enqueue("incoming_doc", %{"body" => doc})
   end
 
   def incoming_ap_doc(params) do
-    %{"op" => "incoming_ap_doc", "params" => params}
-    |> ReceiverWorker.new(worker_args(:federator_incoming))
-    |> Pleroma.Repo.insert()
+    ReceiverWorker.enqueue("incoming_ap_doc", %{"params" => params})
   end
 
   def publish(%{id: "pleroma:fakeid"} = activity) do
@@ -56,27 +50,19 @@ defmodule Pleroma.Web.Federator do
   end
 
   def publish(activity) do
-    %{"op" => "publish", "activity_id" => activity.id}
-    |> PublisherWorker.new(worker_args(:federator_outgoing))
-    |> Pleroma.Repo.insert()
+    PublisherWorker.enqueue("publish", %{"activity_id" => activity.id})
   end
 
   def verify_websub(websub) do
-    %{"op" => "verify_websub", "websub_id" => websub.id}
-    |> SubscriberWorker.new(worker_args(:federator_outgoing))
-    |> Pleroma.Repo.insert()
+    SubscriberWorker.enqueue("verify_websub", %{"websub_id" => websub.id})
   end
 
   def request_subscription(websub) do
-    %{"op" => "request_subscription", "websub_id" => websub.id}
-    |> SubscriberWorker.new(worker_args(:federator_outgoing))
-    |> Pleroma.Repo.insert()
+    SubscriberWorker.enqueue("request_subscription", %{"websub_id" => websub.id})
   end
 
   def refresh_subscriptions(worker_args \\ []) do
-    %{"op" => "refresh_subscriptions"}
-    |> SubscriberWorker.new(worker_args ++ [max_attempts: 1] ++ worker_args(:federator_outgoing))
-    |> Pleroma.Repo.insert()
+    SubscriberWorker.enqueue("refresh_subscriptions", %{}, worker_args ++ [max_attempts: 1])
   end
 
   # Job Worker Callbacks
diff --git a/lib/pleroma/web/federator/publisher.ex b/lib/pleroma/web/federator/publisher.ex
index 42be109ab..937064638 100644
--- a/lib/pleroma/web/federator/publisher.ex
+++ b/lib/pleroma/web/federator/publisher.ex
@@ -31,11 +31,10 @@ defmodule Pleroma.Web.Federator.Publisher do
   """
   @spec enqueue_one(module(), Map.t()) :: :ok
   def enqueue_one(module, %{} = params) do
-    worker_args = Pleroma.Workers.WorkerHelper.worker_args(:federator_outgoing)
-
-    %{"op" => "publish_one", "module" => to_string(module), "params" => params}
-    |> PublisherWorker.new(worker_args)
-    |> Pleroma.Repo.insert()
+    PublisherWorker.enqueue(
+      "publish_one",
+      %{"module" => to_string(module), "params" => params}
+    )
   end
 
   @doc """
diff --git a/lib/pleroma/web/oauth/token/clean_worker.ex b/lib/pleroma/web/oauth/token/clean_worker.ex
index b150a68a7..eb94bf86f 100644
--- a/lib/pleroma/web/oauth/token/clean_worker.ex
+++ b/lib/pleroma/web/oauth/token/clean_worker.ex
@@ -16,12 +16,9 @@ defmodule Pleroma.Web.OAuth.Token.CleanWorker do
               @one_day
             )
 
-  alias Pleroma.Repo
   alias Pleroma.Web.OAuth.Token
   alias Pleroma.Workers.BackgroundWorker
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   def start_link(_), do: GenServer.start_link(__MODULE__, %{})
 
   def init(_) do
@@ -31,9 +28,7 @@ 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()
+    BackgroundWorker.enqueue("clean_expired_tokens", %{})
 
     Process.send_after(self(), :perform, @interval)
     {:noreply, state}
diff --git a/lib/pleroma/web/push/push.ex b/lib/pleroma/web/push/push.ex
index 4973b529c..7ef1532ac 100644
--- a/lib/pleroma/web/push/push.ex
+++ b/lib/pleroma/web/push/push.ex
@@ -3,13 +3,10 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Push do
-  alias Pleroma.Repo
   alias Pleroma.Workers.WebPusherWorker
 
   require Logger
 
-  import Pleroma.Workers.WorkerHelper, only: [worker_args: 1]
-
   def init do
     unless enabled() do
       Logger.warn("""
@@ -35,8 +32,6 @@ defmodule Pleroma.Web.Push do
   end
 
   def send(notification) do
-    %{"op" => "web_push", "notification_id" => notification.id}
-    |> WebPusherWorker.new(worker_args(:web_push))
-    |> Repo.insert()
+    WebPusherWorker.enqueue("web_push", %{"notification_id" => notification.id})
   end
 end
diff --git a/lib/pleroma/workers/activity_expiration_worker.ex b/lib/pleroma/workers/activity_expiration_worker.ex
index 0b491eabb..60dd3feba 100644
--- a/lib/pleroma/workers/activity_expiration_worker.ex
+++ b/lib/pleroma/workers/activity_expiration_worker.ex
@@ -8,6 +8,8 @@ defmodule Pleroma.Workers.ActivityExpirationWorker do
     queue: "activity_expiration",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "activity_expiration"
+
   @impl Oban.Worker
   def perform(
         %{
diff --git a/lib/pleroma/workers/background_worker.ex b/lib/pleroma/workers/background_worker.ex
index 7b5575a5f..b9aef3a92 100644
--- a/lib/pleroma/workers/background_worker.ex
+++ b/lib/pleroma/workers/background_worker.ex
@@ -13,6 +13,8 @@ defmodule Pleroma.Workers.BackgroundWorker do
     queue: "background",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "background"
+
   @impl Oban.Worker
   def perform(%{"op" => "fetch_initial_posts", "user_id" => user_id}, _job) do
     user = User.get_cached_by_id(user_id)
diff --git a/lib/pleroma/workers/digest_emails_worker.ex b/lib/pleroma/workers/digest_emails_worker.ex
new file mode 100644
index 000000000..ca073ce67
--- /dev/null
+++ b/lib/pleroma/workers/digest_emails_worker.ex
@@ -0,0 +1,21 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.DigestEmailsWorker do
+  alias Pleroma.User
+
+  # Note: `max_attempts` is intended to be overridden in `new/2` call
+  use Oban.Worker,
+    queue: "digest_emails",
+    max_attempts: 1
+
+  use Pleroma.Workers.WorkerHelper, queue: "digest_emails"
+
+  @impl Oban.Worker
+  def perform(%{"op" => "digest_email", "user_id" => user_id}, _job) do
+    user_id
+    |> User.get_cached_by_id()
+    |> Pleroma.DigestEmailWorker.perform()
+  end
+end
diff --git a/lib/pleroma/workers/mailer_worker.ex b/lib/pleroma/workers/mailer_worker.ex
index 4f73d61bc..a4bd54a6c 100644
--- a/lib/pleroma/workers/mailer_worker.ex
+++ b/lib/pleroma/workers/mailer_worker.ex
@@ -3,13 +3,13 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.MailerWorker do
-  alias Pleroma.User
-
   # Note: `max_attempts` is intended to be overridden in `new/2` call
   use Oban.Worker,
     queue: "mailer",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "mailer"
+
   @impl Oban.Worker
   def perform(%{"op" => "email", "encoded_email" => encoded_email, "config" => config}, _job) do
     encoded_email
@@ -17,10 +17,4 @@ defmodule Pleroma.Workers.MailerWorker do
     |> :erlang.binary_to_term()
     |> Pleroma.Emails.Mailer.deliver(config)
   end
-
-  def perform(%{"op" => "digest_email", "user_id" => user_id}, _job) do
-    user_id
-    |> User.get_cached_by_id()
-    |> Pleroma.DigestEmailWorker.perform()
-  end
 end
diff --git a/lib/pleroma/workers/publisher_worker.ex b/lib/pleroma/workers/publisher_worker.ex
index 5671d2a29..a3ac22635 100644
--- a/lib/pleroma/workers/publisher_worker.ex
+++ b/lib/pleroma/workers/publisher_worker.ex
@@ -11,6 +11,8 @@ defmodule Pleroma.Workers.PublisherWorker do
     queue: "federator_outgoing",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "federator_outgoing"
+
   def backoff(attempt) when is_integer(attempt) do
     Pleroma.Workers.WorkerHelper.sidekiq_backoff(attempt, 5)
   end
diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex
index cdce630f2..3cc415ce4 100644
--- a/lib/pleroma/workers/receiver_worker.ex
+++ b/lib/pleroma/workers/receiver_worker.ex
@@ -10,6 +10,8 @@ defmodule Pleroma.Workers.ReceiverWorker do
     queue: "federator_incoming",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "federator_incoming"
+
   @impl Oban.Worker
   def perform(%{"op" => "incoming_doc", "body" => doc}, _job) do
     Federator.perform(:incoming_doc, doc)
diff --git a/lib/pleroma/workers/scheduled_activity_worker.ex b/lib/pleroma/workers/scheduled_activity_worker.ex
index 4094411ae..936bb64d3 100644
--- a/lib/pleroma/workers/scheduled_activity_worker.ex
+++ b/lib/pleroma/workers/scheduled_activity_worker.ex
@@ -8,6 +8,8 @@ defmodule Pleroma.Workers.ScheduledActivityWorker do
     queue: "scheduled_activities",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "scheduled_activities"
+
   @impl Oban.Worker
   def perform(%{"op" => "execute", "activity_id" => activity_id}, _job) do
     Pleroma.ScheduledActivityWorker.perform(:execute, activity_id)
diff --git a/lib/pleroma/workers/subscriber_worker.ex b/lib/pleroma/workers/subscriber_worker.ex
index 22d1dc956..4fb994554 100644
--- a/lib/pleroma/workers/subscriber_worker.ex
+++ b/lib/pleroma/workers/subscriber_worker.ex
@@ -12,6 +12,8 @@ defmodule Pleroma.Workers.SubscriberWorker do
     queue: "federator_outgoing",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "federator_outgoing"
+
   @impl Oban.Worker
   def perform(%{"op" => "refresh_subscriptions"}, _job) do
     Federator.perform(:refresh_subscriptions)
diff --git a/lib/pleroma/workers/transmogrifier_worker.ex b/lib/pleroma/workers/transmogrifier_worker.ex
index 6f5c1a2f2..6fecc2bf9 100644
--- a/lib/pleroma/workers/transmogrifier_worker.ex
+++ b/lib/pleroma/workers/transmogrifier_worker.ex
@@ -10,6 +10,8 @@ defmodule Pleroma.Workers.TransmogrifierWorker do
     queue: "transmogrifier",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "transmogrifier"
+
   @impl Oban.Worker
   def perform(%{"op" => "user_upgrade", "user_id" => user_id}, _job) do
     user = User.get_cached_by_id(user_id)
diff --git a/lib/pleroma/workers/web_pusher_worker.ex b/lib/pleroma/workers/web_pusher_worker.ex
index 2b1d3b99a..4c2591a5c 100644
--- a/lib/pleroma/workers/web_pusher_worker.ex
+++ b/lib/pleroma/workers/web_pusher_worker.ex
@@ -11,6 +11,8 @@ defmodule Pleroma.Workers.WebPusherWorker do
     queue: "web_push",
     max_attempts: 1
 
+  use Pleroma.Workers.WorkerHelper, queue: "web_push"
+
   @impl Oban.Worker
   def perform(%{"op" => "web_push", "notification_id" => notification_id}, _job) do
     notification = Repo.get(Notification, notification_id)
diff --git a/lib/pleroma/workers/worker_helper.ex b/lib/pleroma/workers/worker_helper.ex
index f9ed2e64d..b12f198d4 100644
--- a/lib/pleroma/workers/worker_helper.ex
+++ b/lib/pleroma/workers/worker_helper.ex
@@ -4,6 +4,7 @@
 
 defmodule Pleroma.Workers.WorkerHelper do
   alias Pleroma.Config
+  alias Pleroma.Workers.WorkerHelper
 
   def worker_args(queue) do
     case Config.get([:workers, :retries, queue]) do
@@ -20,4 +21,21 @@ defmodule Pleroma.Workers.WorkerHelper do
 
     trunc(backoff)
   end
+
+  defmacro __using__(opts) do
+    caller_module = __CALLER__.module
+    queue = Keyword.fetch!(opts, :queue)
+
+    quote do
+      def enqueue(op, params, worker_args \\ []) do
+        params = Map.merge(%{"op" => op}, params)
+        queue_atom = String.to_atom(unquote(queue))
+        worker_args = worker_args ++ WorkerHelper.worker_args(queue_atom)
+
+        unquote(caller_module)
+        |> apply(:new, [params, worker_args])
+        |> Pleroma.Repo.insert()
+      end
+    end
+  end
 end
-- 
cgit v1.2.3


From b49085c156a6a4449c95c2c315f6250317122735 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Mon, 2 Sep 2019 14:57:40 +0300
Subject: [#1149] Refactoring: GenServer workers renamed to daemons, `use
 Oban.Worker` moved to helper.

---
 lib/pleroma/activity_expiration_worker.ex         | 66 -----------------------
 lib/pleroma/application.ex                        |  4 +-
 lib/pleroma/daemons/activity_expiration_daemon.ex | 66 +++++++++++++++++++++++
 lib/pleroma/daemons/digest_email_daemon.ex        | 42 +++++++++++++++
 lib/pleroma/daemons/scheduled_activity_daemon.ex  | 62 +++++++++++++++++++++
 lib/pleroma/digest_email_worker.ex                | 42 ---------------
 lib/pleroma/scheduled_activity_worker.ex          | 62 ---------------------
 lib/pleroma/workers/activity_expiration_worker.ex |  7 +--
 lib/pleroma/workers/background_worker.ex          |  5 --
 lib/pleroma/workers/digest_emails_worker.ex       |  7 +--
 lib/pleroma/workers/mailer_worker.ex              |  5 --
 lib/pleroma/workers/publisher_worker.ex           |  5 --
 lib/pleroma/workers/receiver_worker.ex            |  5 --
 lib/pleroma/workers/scheduled_activity_worker.ex  |  7 +--
 lib/pleroma/workers/subscriber_worker.ex          |  5 --
 lib/pleroma/workers/transmogrifier_worker.ex      |  5 --
 lib/pleroma/workers/web_pusher_worker.ex          |  5 --
 lib/pleroma/workers/worker_helper.ex              |  5 ++
 18 files changed, 180 insertions(+), 225 deletions(-)
 delete mode 100644 lib/pleroma/activity_expiration_worker.ex
 create mode 100644 lib/pleroma/daemons/activity_expiration_daemon.ex
 create mode 100644 lib/pleroma/daemons/digest_email_daemon.ex
 create mode 100644 lib/pleroma/daemons/scheduled_activity_daemon.ex
 delete mode 100644 lib/pleroma/digest_email_worker.ex
 delete mode 100644 lib/pleroma/scheduled_activity_worker.ex

(limited to 'lib')

diff --git a/lib/pleroma/activity_expiration_worker.ex b/lib/pleroma/activity_expiration_worker.ex
deleted file mode 100644
index c0820c202..000000000
--- a/lib/pleroma/activity_expiration_worker.ex
+++ /dev/null
@@ -1,66 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.ActivityExpirationWorker do
-  alias Pleroma.Activity
-  alias Pleroma.ActivityExpiration
-  alias Pleroma.Config
-  alias Pleroma.Repo
-  alias Pleroma.User
-  alias Pleroma.Web.CommonAPI
-
-  require Logger
-  use GenServer
-  import Ecto.Query
-
-  @schedule_interval :timer.minutes(1)
-
-  def start_link(_) do
-    GenServer.start_link(__MODULE__, nil)
-  end
-
-  @impl true
-  def init(_) do
-    if Config.get([ActivityExpiration, :enabled]) do
-      schedule_next()
-      {:ok, nil}
-    else
-      :ignore
-    end
-  end
-
-  def perform(:execute, expiration_id) do
-    try do
-      expiration =
-        ActivityExpiration
-        |> where([e], e.id == ^expiration_id)
-        |> Repo.one!()
-
-      activity = Activity.get_by_id_with_object(expiration.activity_id)
-      user = User.get_by_ap_id(activity.object.data["actor"])
-      CommonAPI.delete(activity.id, user)
-    rescue
-      error ->
-        Logger.error("#{__MODULE__} Couldn't delete expired activity: #{inspect(error)}")
-    end
-  end
-
-  @impl true
-  def handle_info(:perform, state) do
-    ActivityExpiration.due_expirations(@schedule_interval)
-    |> Enum.each(fn expiration ->
-      Pleroma.Workers.ActivityExpirationWorker.enqueue(
-        "activity_expiration",
-        %{"activity_expiration_id" => expiration.id}
-      )
-    end)
-
-    schedule_next()
-    {:noreply, state}
-  end
-
-  defp schedule_next do
-    Process.send_after(self(), :perform, @schedule_interval)
-  end
-end
diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index f8f866dbd..0c27027a0 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -36,8 +36,8 @@ defmodule Pleroma.Application do
         Pleroma.Emoji,
         Pleroma.Captcha,
         Pleroma.FlakeId,
-        Pleroma.ScheduledActivityWorker,
-        Pleroma.ActivityExpirationWorker
+        Pleroma.Daemons.ScheduledActivityDaemon,
+        Pleroma.Daemons.ActivityExpirationDaemon
       ] ++
         cachex_children() ++
         hackney_pool_children() ++
diff --git a/lib/pleroma/daemons/activity_expiration_daemon.ex b/lib/pleroma/daemons/activity_expiration_daemon.ex
new file mode 100644
index 000000000..cab7628c4
--- /dev/null
+++ b/lib/pleroma/daemons/activity_expiration_daemon.ex
@@ -0,0 +1,66 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Daemons.ActivityExpirationDaemon do
+  alias Pleroma.Activity
+  alias Pleroma.ActivityExpiration
+  alias Pleroma.Config
+  alias Pleroma.Repo
+  alias Pleroma.User
+  alias Pleroma.Web.CommonAPI
+
+  require Logger
+  use GenServer
+  import Ecto.Query
+
+  @schedule_interval :timer.minutes(1)
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, nil)
+  end
+
+  @impl true
+  def init(_) do
+    if Config.get([ActivityExpiration, :enabled]) do
+      schedule_next()
+      {:ok, nil}
+    else
+      :ignore
+    end
+  end
+
+  def perform(:execute, expiration_id) do
+    try do
+      expiration =
+        ActivityExpiration
+        |> where([e], e.id == ^expiration_id)
+        |> Repo.one!()
+
+      activity = Activity.get_by_id_with_object(expiration.activity_id)
+      user = User.get_by_ap_id(activity.object.data["actor"])
+      CommonAPI.delete(activity.id, user)
+    rescue
+      error ->
+        Logger.error("#{__MODULE__} Couldn't delete expired activity: #{inspect(error)}")
+    end
+  end
+
+  @impl true
+  def handle_info(:perform, state) do
+    ActivityExpiration.due_expirations(@schedule_interval)
+    |> Enum.each(fn expiration ->
+      Pleroma.Workers.ActivityExpirationWorker.enqueue(
+        "activity_expiration",
+        %{"activity_expiration_id" => expiration.id}
+      )
+    end)
+
+    schedule_next()
+    {:noreply, state}
+  end
+
+  defp schedule_next do
+    Process.send_after(self(), :perform, @schedule_interval)
+  end
+end
diff --git a/lib/pleroma/daemons/digest_email_daemon.ex b/lib/pleroma/daemons/digest_email_daemon.ex
new file mode 100644
index 000000000..462ad2c55
--- /dev/null
+++ b/lib/pleroma/daemons/digest_email_daemon.ex
@@ -0,0 +1,42 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Daemons.DigestEmailDaemon do
+  alias Pleroma.Repo
+  alias Pleroma.Workers.DigestEmailsWorker
+
+  import Ecto.Query
+
+  def perform do
+    config = Pleroma.Config.get([:email_notifications, :digest])
+    negative_interval = -Map.fetch!(config, :interval)
+    inactivity_threshold = Map.fetch!(config, :inactivity_threshold)
+    inactive_users_query = Pleroma.User.list_inactive_users_query(inactivity_threshold)
+
+    now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
+
+    from(u in inactive_users_query,
+      where: fragment(~s(? #> '{"email_notifications","digest"}' @> 'true'), u.info),
+      where: u.last_digest_emailed_at < datetime_add(^now, ^negative_interval, "day"),
+      select: u
+    )
+    |> Repo.all()
+    |> Enum.each(fn user ->
+      DigestEmailsWorker.enqueue("digest_email", %{"user_id" => user.id})
+    end)
+  end
+
+  @doc """
+  Send digest email to the given user.
+  Updates `last_digest_emailed_at` field for the user and returns the updated user.
+  """
+  @spec perform(Pleroma.User.t()) :: Pleroma.User.t()
+  def perform(user) do
+    with %Swoosh.Email{} = email <- Pleroma.Emails.UserEmail.digest_email(user) do
+      Pleroma.Emails.Mailer.deliver_async(email)
+    end
+
+    Pleroma.User.touch_last_digest_emailed_at(user)
+  end
+end
diff --git a/lib/pleroma/daemons/scheduled_activity_daemon.ex b/lib/pleroma/daemons/scheduled_activity_daemon.ex
new file mode 100644
index 000000000..aee5f723a
--- /dev/null
+++ b/lib/pleroma/daemons/scheduled_activity_daemon.ex
@@ -0,0 +1,62 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Daemons.ScheduledActivityDaemon do
+  @moduledoc """
+  Sends scheduled activities to the job queue.
+  """
+
+  alias Pleroma.Config
+  alias Pleroma.ScheduledActivity
+  alias Pleroma.User
+  alias Pleroma.Web.CommonAPI
+
+  use GenServer
+  require Logger
+
+  @schedule_interval :timer.minutes(1)
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, nil)
+  end
+
+  def init(_) do
+    if Config.get([ScheduledActivity, :enabled]) do
+      schedule_next()
+      {:ok, nil}
+    else
+      :ignore
+    end
+  end
+
+  def perform(:execute, scheduled_activity_id) do
+    try do
+      {:ok, scheduled_activity} = ScheduledActivity.delete(scheduled_activity_id)
+      %User{} = user = User.get_cached_by_id(scheduled_activity.user_id)
+      {:ok, _result} = CommonAPI.post(user, scheduled_activity.params)
+    rescue
+      error ->
+        Logger.error(
+          "#{__MODULE__} Couldn't create a status from the scheduled activity: #{inspect(error)}"
+        )
+    end
+  end
+
+  def handle_info(:perform, state) do
+    ScheduledActivity.due_activities(@schedule_interval)
+    |> Enum.each(fn scheduled_activity ->
+      Pleroma.Workers.ScheduledActivityWorker.enqueue(
+        "execute",
+        %{"activity_id" => scheduled_activity.id}
+      )
+    end)
+
+    schedule_next()
+    {:noreply, state}
+  end
+
+  defp schedule_next do
+    Process.send_after(self(), :perform, @schedule_interval)
+  end
+end
diff --git a/lib/pleroma/digest_email_worker.ex b/lib/pleroma/digest_email_worker.ex
deleted file mode 100644
index 5be7cf26b..000000000
--- a/lib/pleroma/digest_email_worker.ex
+++ /dev/null
@@ -1,42 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.DigestEmailWorker do
-  alias Pleroma.Repo
-  alias Pleroma.Workers.DigestEmailsWorker
-
-  import Ecto.Query
-
-  def perform do
-    config = Pleroma.Config.get([:email_notifications, :digest])
-    negative_interval = -Map.fetch!(config, :interval)
-    inactivity_threshold = Map.fetch!(config, :inactivity_threshold)
-    inactive_users_query = Pleroma.User.list_inactive_users_query(inactivity_threshold)
-
-    now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
-
-    from(u in inactive_users_query,
-      where: fragment(~s(? #> '{"email_notifications","digest"}' @> 'true'), u.info),
-      where: u.last_digest_emailed_at < datetime_add(^now, ^negative_interval, "day"),
-      select: u
-    )
-    |> Repo.all()
-    |> Enum.each(fn user ->
-      DigestEmailsWorker.enqueue("digest_email", %{"user_id" => user.id})
-    end)
-  end
-
-  @doc """
-  Send digest email to the given user.
-  Updates `last_digest_emailed_at` field for the user and returns the updated user.
-  """
-  @spec perform(Pleroma.User.t()) :: Pleroma.User.t()
-  def perform(user) do
-    with %Swoosh.Email{} = email <- Pleroma.Emails.UserEmail.digest_email(user) do
-      Pleroma.Emails.Mailer.deliver_async(email)
-    end
-
-    Pleroma.User.touch_last_digest_emailed_at(user)
-  end
-end
diff --git a/lib/pleroma/scheduled_activity_worker.ex b/lib/pleroma/scheduled_activity_worker.ex
deleted file mode 100644
index c41a542de..000000000
--- a/lib/pleroma/scheduled_activity_worker.ex
+++ /dev/null
@@ -1,62 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.ScheduledActivityWorker do
-  @moduledoc """
-  Sends scheduled activities to the job queue.
-  """
-
-  alias Pleroma.Config
-  alias Pleroma.ScheduledActivity
-  alias Pleroma.User
-  alias Pleroma.Web.CommonAPI
-
-  use GenServer
-  require Logger
-
-  @schedule_interval :timer.minutes(1)
-
-  def start_link(_) do
-    GenServer.start_link(__MODULE__, nil)
-  end
-
-  def init(_) do
-    if Config.get([ScheduledActivity, :enabled]) do
-      schedule_next()
-      {:ok, nil}
-    else
-      :ignore
-    end
-  end
-
-  def perform(:execute, scheduled_activity_id) do
-    try do
-      {:ok, scheduled_activity} = ScheduledActivity.delete(scheduled_activity_id)
-      %User{} = user = User.get_cached_by_id(scheduled_activity.user_id)
-      {:ok, _result} = CommonAPI.post(user, scheduled_activity.params)
-    rescue
-      error ->
-        Logger.error(
-          "#{__MODULE__} Couldn't create a status from the scheduled activity: #{inspect(error)}"
-        )
-    end
-  end
-
-  def handle_info(:perform, state) do
-    ScheduledActivity.due_activities(@schedule_interval)
-    |> Enum.each(fn scheduled_activity ->
-      Pleroma.Workers.ScheduledActivityWorker.enqueue(
-        "execute",
-        %{"activity_id" => scheduled_activity.id}
-      )
-    end)
-
-    schedule_next()
-    {:noreply, state}
-  end
-
-  defp schedule_next do
-    Process.send_after(self(), :perform, @schedule_interval)
-  end
-end
diff --git a/lib/pleroma/workers/activity_expiration_worker.ex b/lib/pleroma/workers/activity_expiration_worker.ex
index 60dd3feba..4e3e4195f 100644
--- a/lib/pleroma/workers/activity_expiration_worker.ex
+++ b/lib/pleroma/workers/activity_expiration_worker.ex
@@ -3,11 +3,6 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.ActivityExpirationWorker do
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "activity_expiration",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "activity_expiration"
 
   @impl Oban.Worker
@@ -18,6 +13,6 @@ defmodule Pleroma.Workers.ActivityExpirationWorker do
         },
         _job
       ) do
-    Pleroma.ActivityExpirationWorker.perform(:execute, activity_expiration_id)
+    Pleroma.Daemons.ActivityExpirationDaemon.perform(:execute, activity_expiration_id)
   end
 end
diff --git a/lib/pleroma/workers/background_worker.ex b/lib/pleroma/workers/background_worker.ex
index b9aef3a92..082f20ab7 100644
--- a/lib/pleroma/workers/background_worker.ex
+++ b/lib/pleroma/workers/background_worker.ex
@@ -8,11 +8,6 @@ defmodule Pleroma.Workers.BackgroundWorker do
   alias Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy
   alias Pleroma.Web.OAuth.Token.CleanWorker
 
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "background",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "background"
 
   @impl Oban.Worker
diff --git a/lib/pleroma/workers/digest_emails_worker.ex b/lib/pleroma/workers/digest_emails_worker.ex
index ca073ce67..3e5a836d0 100644
--- a/lib/pleroma/workers/digest_emails_worker.ex
+++ b/lib/pleroma/workers/digest_emails_worker.ex
@@ -5,17 +5,12 @@
 defmodule Pleroma.Workers.DigestEmailsWorker do
   alias Pleroma.User
 
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "digest_emails",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "digest_emails"
 
   @impl Oban.Worker
   def perform(%{"op" => "digest_email", "user_id" => user_id}, _job) do
     user_id
     |> User.get_cached_by_id()
-    |> Pleroma.DigestEmailWorker.perform()
+    |> Pleroma.Daemons.DigestEmailDaemon.perform()
   end
 end
diff --git a/lib/pleroma/workers/mailer_worker.ex b/lib/pleroma/workers/mailer_worker.ex
index a4bd54a6c..1b7a0eb3e 100644
--- a/lib/pleroma/workers/mailer_worker.ex
+++ b/lib/pleroma/workers/mailer_worker.ex
@@ -3,11 +3,6 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.MailerWorker do
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "mailer",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "mailer"
 
   @impl Oban.Worker
diff --git a/lib/pleroma/workers/publisher_worker.ex b/lib/pleroma/workers/publisher_worker.ex
index a3ac22635..455f7fc7e 100644
--- a/lib/pleroma/workers/publisher_worker.ex
+++ b/lib/pleroma/workers/publisher_worker.ex
@@ -6,11 +6,6 @@ defmodule Pleroma.Workers.PublisherWorker do
   alias Pleroma.Activity
   alias Pleroma.Web.Federator
 
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "federator_outgoing",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "federator_outgoing"
 
   def backoff(attempt) when is_integer(attempt) do
diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex
index 3cc415ce4..83d528a66 100644
--- a/lib/pleroma/workers/receiver_worker.ex
+++ b/lib/pleroma/workers/receiver_worker.ex
@@ -5,11 +5,6 @@
 defmodule Pleroma.Workers.ReceiverWorker do
   alias Pleroma.Web.Federator
 
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "federator_incoming",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "federator_incoming"
 
   @impl Oban.Worker
diff --git a/lib/pleroma/workers/scheduled_activity_worker.ex b/lib/pleroma/workers/scheduled_activity_worker.ex
index 936bb64d3..ca7d53af1 100644
--- a/lib/pleroma/workers/scheduled_activity_worker.ex
+++ b/lib/pleroma/workers/scheduled_activity_worker.ex
@@ -3,15 +3,10 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.ScheduledActivityWorker do
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "scheduled_activities",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "scheduled_activities"
 
   @impl Oban.Worker
   def perform(%{"op" => "execute", "activity_id" => activity_id}, _job) do
-    Pleroma.ScheduledActivityWorker.perform(:execute, activity_id)
+    Pleroma.Daemons.ScheduledActivityDaemon.perform(:execute, activity_id)
   end
 end
diff --git a/lib/pleroma/workers/subscriber_worker.ex b/lib/pleroma/workers/subscriber_worker.ex
index 4fb994554..fc490e300 100644
--- a/lib/pleroma/workers/subscriber_worker.ex
+++ b/lib/pleroma/workers/subscriber_worker.ex
@@ -7,11 +7,6 @@ defmodule Pleroma.Workers.SubscriberWorker do
   alias Pleroma.Web.Federator
   alias Pleroma.Web.Websub
 
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "federator_outgoing",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "federator_outgoing"
 
   @impl Oban.Worker
diff --git a/lib/pleroma/workers/transmogrifier_worker.ex b/lib/pleroma/workers/transmogrifier_worker.ex
index 6fecc2bf9..b581a2f86 100644
--- a/lib/pleroma/workers/transmogrifier_worker.ex
+++ b/lib/pleroma/workers/transmogrifier_worker.ex
@@ -5,11 +5,6 @@
 defmodule Pleroma.Workers.TransmogrifierWorker do
   alias Pleroma.User
 
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "transmogrifier",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "transmogrifier"
 
   @impl Oban.Worker
diff --git a/lib/pleroma/workers/web_pusher_worker.ex b/lib/pleroma/workers/web_pusher_worker.ex
index 4c2591a5c..bea2baffb 100644
--- a/lib/pleroma/workers/web_pusher_worker.ex
+++ b/lib/pleroma/workers/web_pusher_worker.ex
@@ -6,11 +6,6 @@ defmodule Pleroma.Workers.WebPusherWorker do
   alias Pleroma.Notification
   alias Pleroma.Repo
 
-  # Note: `max_attempts` is intended to be overridden in `new/2` call
-  use Oban.Worker,
-    queue: "web_push",
-    max_attempts: 1
-
   use Pleroma.Workers.WorkerHelper, queue: "web_push"
 
   @impl Oban.Worker
diff --git a/lib/pleroma/workers/worker_helper.ex b/lib/pleroma/workers/worker_helper.ex
index b12f198d4..358efa14a 100644
--- a/lib/pleroma/workers/worker_helper.ex
+++ b/lib/pleroma/workers/worker_helper.ex
@@ -27,6 +27,11 @@ defmodule Pleroma.Workers.WorkerHelper do
     queue = Keyword.fetch!(opts, :queue)
 
     quote do
+      # Note: `max_attempts` is intended to be overridden in `new/2` call
+      use Oban.Worker,
+        queue: unquote(queue),
+        max_attempts: 1
+
       def enqueue(op, params, worker_args \\ []) do
         params = Map.merge(%{"op" => op}, params)
         queue_atom = String.to_atom(unquote(queue))
-- 
cgit v1.2.3