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 35ef470d000c53e21c6f867d53ca3a83260d93b8 Mon Sep 17 00:00:00 2001
From: Sadposter <hannah+pleroma@coffee-and-dreams.uk>
Date: Mon, 2 Sep 2019 12:15:21 +0100
Subject: truncate fields for remote users instead

---
 lib/pleroma/user/info.ex | 7 +++++++
 1 file changed, 7 insertions(+)

(limited to 'lib')

diff --git a/lib/pleroma/user/info.ex b/lib/pleroma/user/info.ex
index 779bfbc18..0beb2f721 100644
--- a/lib/pleroma/user/info.ex
+++ b/lib/pleroma/user/info.ex
@@ -242,6 +242,7 @@ defmodule Pleroma.User.Info do
   end
 
   def remote_user_creation(info, params) do
+    params = Map.put(params, "fields", Enum.map(params["fields"], &truncate_field/1))
     info
     |> cast(params, [
       :ap_enabled,
@@ -326,6 +327,12 @@ defmodule Pleroma.User.Info do
 
   defp valid_field?(_), do: false
 
+  defp truncate_field(%{"name" => name, "value" => value}) do
+    {name, _chopped} = String.split_at(name, Pleroma.Config.get([:instance, :account_field_name_length], 255))
+    {value, _chopped} = String.split_at(value, Pleroma.Config.get([:instance, :account_field_value_length], 255))
+    %{"name" => name, "value" => value}
+  end
+
   @spec confirmation_changeset(Info.t(), keyword()) :: Changeset.t()
   def confirmation_changeset(info, opts) do
     need_confirmation? = Keyword.get(opts, :need_confirmation)
-- 
cgit v1.2.3


From 05c935c3961e4c1a20c7713611920318d45d4b57 Mon Sep 17 00:00:00 2001
From: Sadposter <hannah+pleroma@coffee-and-dreams.uk>
Date: Mon, 2 Sep 2019 12:15:40 +0100
Subject: mix format

---
 lib/pleroma/user/info.ex | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/user/info.ex b/lib/pleroma/user/info.ex
index 0beb2f721..ca1282d02 100644
--- a/lib/pleroma/user/info.ex
+++ b/lib/pleroma/user/info.ex
@@ -243,6 +243,7 @@ defmodule Pleroma.User.Info do
 
   def remote_user_creation(info, params) do
     params = Map.put(params, "fields", Enum.map(params["fields"], &truncate_field/1))
+
     info
     |> cast(params, [
       :ap_enabled,
@@ -328,8 +329,12 @@ defmodule Pleroma.User.Info do
   defp valid_field?(_), do: false
 
   defp truncate_field(%{"name" => name, "value" => value}) do
-    {name, _chopped} = String.split_at(name, Pleroma.Config.get([:instance, :account_field_name_length], 255))
-    {value, _chopped} = String.split_at(value, Pleroma.Config.get([:instance, :account_field_value_length], 255))
+    {name, _chopped} =
+      String.split_at(name, Pleroma.Config.get([:instance, :account_field_name_length], 255))
+
+    {value, _chopped} =
+      String.split_at(value, Pleroma.Config.get([:instance, :account_field_value_length], 255))
+
     %{"name" => name, "value" => value}
   end
 
-- 
cgit v1.2.3


From d0f07e55d28d25684130cb1090d0bdbb48807548 Mon Sep 17 00:00:00 2001
From: Sadposter <hannah+pleroma@coffee-and-dreams.uk>
Date: Mon, 2 Sep 2019 12:31:23 +0100
Subject: use atom key for fields

---
 lib/pleroma/user/info.ex | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/user/info.ex b/lib/pleroma/user/info.ex
index ca1282d02..151e025de 100644
--- a/lib/pleroma/user/info.ex
+++ b/lib/pleroma/user/info.ex
@@ -242,7 +242,12 @@ defmodule Pleroma.User.Info do
   end
 
   def remote_user_creation(info, params) do
-    params = Map.put(params, "fields", Enum.map(params["fields"], &truncate_field/1))
+    params =
+      if Map.has_key?(params, :fields) do
+        Map.put(params, :fields, Enum.map(params[:fields], &truncate_field/1))
+      else
+        params
+      end
 
     info
     |> cast(params, [
-- 
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


From 2975da284b75c846a99a56ce70a91ebc3cc43f33 Mon Sep 17 00:00:00 2001
From: Sadposter <hannah+pleroma@coffee-and-dreams.uk>
Date: Wed, 4 Sep 2019 15:45:40 +0100
Subject: truncate remote user bio/display name

---
 lib/pleroma/user.ex | 16 +++++++++++++++-
 1 file changed, 15 insertions(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 29fd6d2ea..87e56b5b4 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -174,11 +174,25 @@ defmodule Pleroma.User do
     |> Repo.aggregate(:count, :id)
   end
 
+  defp truncate_if_exists(params, key, max_length) do
+    if Map.has_key?(params, key) do
+      {value, _chopped} = String.split_at(params[key], max_length)
+      Map.put(params, key, value)
+    else
+      params
+    end
+  end
+
   def remote_user_creation(params) do
     bio_limit = Pleroma.Config.get([:instance, :user_bio_length], 5000)
     name_limit = Pleroma.Config.get([:instance, :user_name_length], 100)
 
-    params = Map.put(params, :info, params[:info] || %{})
+    params =
+      params
+      |> Map.put(:info, params[:info] || %{})
+      |> truncate_if_exists(:name, name_limit)
+      |> truncate_if_exists(:bio, bio_limit)
+
     info_cng = User.Info.remote_user_creation(%User.Info{}, params[:info])
 
     changes =
-- 
cgit v1.2.3


From cb99cfcc65f57f0044117ebd12d040488343d9ef Mon Sep 17 00:00:00 2001
From: Sadposter <hannah+pleroma@coffee-and-dreams.uk>
Date: Wed, 4 Sep 2019 15:57:42 +0100
Subject: don't try to truncate non-strings

---
 lib/pleroma/user.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 87e56b5b4..e2ebce6fc 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -175,7 +175,7 @@ defmodule Pleroma.User do
   end
 
   defp truncate_if_exists(params, key, max_length) do
-    if Map.has_key?(params, key) do
+    if Map.has_key?(params, key) and is_binary(params[key]) do
       {value, _chopped} = String.split_at(params[key], max_length)
       Map.put(params, key, value)
     else
-- 
cgit v1.2.3


From 40b3289c26137ee4d07c7fb79faf232714cc7592 Mon Sep 17 00:00:00 2001
From: Egor Kislitsyn <egor@kislitsyn.com>
Date: Fri, 6 Sep 2019 17:08:47 +0700
Subject: Refactor `add_link_headers/7` -> `add_link_headers/3`

---
 lib/pleroma/web/controller_helper.ex               | 95 ++++++----------------
 .../controllers/mastodon_api_controller.ex         | 28 +++----
 .../web/pleroma_api/pleroma_api_controller.ex      | 27 ++----
 3 files changed, 50 insertions(+), 100 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/controller_helper.ex b/lib/pleroma/web/controller_helper.ex
index eeac9f503..b53a01955 100644
--- a/lib/pleroma/web/controller_helper.ex
+++ b/lib/pleroma/web/controller_helper.ex
@@ -34,79 +34,38 @@ defmodule Pleroma.Web.ControllerHelper do
 
   defp param_to_integer(_, default), do: default
 
-  def add_link_headers(
-        conn,
-        method,
-        activities,
-        param \\ nil,
-        params \\ %{},
-        func3 \\ nil,
-        func4 \\ nil
-      ) do
-    params =
-      conn.params
-      |> Map.drop(["since_id", "max_id", "min_id"])
-      |> Map.merge(params)
+  def add_link_headers(conn, activities, extra_params \\ %{}) do
+    case List.last(activities) do
+      %{id: max_id} ->
+        params =
+          conn.params
+          |> Map.drop(Map.keys(conn.path_params))
+          |> Map.drop(["since_id", "max_id", "min_id"])
+          |> Map.merge(extra_params)
 
-    last = List.last(activities)
+        limit =
+          params
+          |> Map.get("limit", "20")
+          |> String.to_integer()
 
-    func3 = func3 || (&mastodon_api_url/3)
-    func4 = func4 || (&mastodon_api_url/4)
+        min_id =
+          if length(activities) <= limit do
+            activities
+            |> List.first()
+            |> Map.get(:id)
+          else
+            activities
+            |> Enum.at(limit * -1)
+            |> Map.get(:id)
+          end
 
-    if last do
-      max_id = last.id
+        next_url = current_url(conn, Map.merge(params, %{max_id: max_id}))
+        prev_url = current_url(conn, Map.merge(params, %{min_id: min_id}))
 
-      limit =
-        params
-        |> Map.get("limit", "20")
-        |> String.to_integer()
+        put_resp_header(conn, "link", "<#{next_url}>; rel=\"next\", <#{prev_url}>; rel=\"prev\"")
 
-      min_id =
-        if length(activities) <= limit do
-          activities
-          |> List.first()
-          |> Map.get(:id)
-        else
-          activities
-          |> Enum.at(limit * -1)
-          |> Map.get(:id)
-        end
-
-      {next_url, prev_url} =
-        if param do
-          {
-            func4.(
-              Pleroma.Web.Endpoint,
-              method,
-              param,
-              Map.merge(params, %{max_id: max_id})
-            ),
-            func4.(
-              Pleroma.Web.Endpoint,
-              method,
-              param,
-              Map.merge(params, %{min_id: min_id})
-            )
-          }
-        else
-          {
-            func3.(
-              Pleroma.Web.Endpoint,
-              method,
-              Map.merge(params, %{max_id: max_id})
-            ),
-            func3.(
-              Pleroma.Web.Endpoint,
-              method,
-              Map.merge(params, %{min_id: min_id})
-            )
-          }
-        end
-
-      conn
-      |> put_resp_header("link", "<#{next_url}>; rel=\"next\", <#{prev_url}>; rel=\"prev\"")
-    else
-      conn
+      _ ->
+        conn
     end
   end
 end
diff --git a/lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex b/lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex
index 8dfad7a54..f30a21bcc 100644
--- a/lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex
+++ b/lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex
@@ -6,7 +6,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
   use Pleroma.Web, :controller
 
   import Pleroma.Web.ControllerHelper,
-    only: [json_response: 3, add_link_headers: 5, add_link_headers: 4, add_link_headers: 3]
+    only: [json_response: 3, add_link_headers: 2, add_link_headers: 3]
 
   alias Ecto.Changeset
   alias Pleroma.Activity
@@ -365,7 +365,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       |> Enum.reverse()
 
     conn
-    |> add_link_headers(:home_timeline, activities)
+    |> add_link_headers(activities)
     |> put_view(StatusView)
     |> render("index.json", %{activities: activities, for: user, as: :activity})
   end
@@ -384,7 +384,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       |> Enum.reverse()
 
     conn
-    |> add_link_headers(:public_timeline, activities, false, %{"local" => local_only})
+    |> add_link_headers(activities, %{"local" => local_only})
     |> put_view(StatusView)
     |> render("index.json", %{activities: activities, for: user, as: :activity})
   end
@@ -398,7 +398,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       activities = ActivityPub.fetch_user_activities(user, reading_user, params)
 
       conn
-      |> add_link_headers(:user_statuses, activities, params["id"])
+      |> add_link_headers(activities)
       |> put_view(StatusView)
       |> render("index.json", %{
         activities: activities,
@@ -422,7 +422,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       |> Pagination.fetch_paginated(params)
 
     conn
-    |> add_link_headers(:dm_timeline, activities)
+    |> add_link_headers(activities)
     |> put_view(StatusView)
     |> render("index.json", %{activities: activities, for: user, as: :activity})
   end
@@ -523,7 +523,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
   def scheduled_statuses(%{assigns: %{user: user}} = conn, params) do
     with scheduled_activities <- MastodonAPI.get_scheduled_activities(user, params) do
       conn
-      |> add_link_headers(:scheduled_statuses, scheduled_activities)
+      |> add_link_headers(scheduled_activities)
       |> put_view(ScheduledActivityView)
       |> render("index.json", %{scheduled_activities: scheduled_activities})
     end
@@ -706,7 +706,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
     notifications = MastodonAPI.get_notifications(user, params)
 
     conn
-    |> add_link_headers(:notifications, notifications)
+    |> add_link_headers(notifications)
     |> put_view(NotificationView)
     |> render("index.json", %{notifications: notifications, for: user})
   end
@@ -894,7 +894,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       |> Enum.reverse()
 
     conn
-    |> add_link_headers(:hashtag_timeline, activities, params["tag"], %{"local" => local_only})
+    |> add_link_headers(activities, %{"local" => local_only})
     |> put_view(StatusView)
     |> render("index.json", %{activities: activities, for: user, as: :activity})
   end
@@ -910,7 +910,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
         end
 
       conn
-      |> add_link_headers(:followers, followers, user)
+      |> add_link_headers(followers)
       |> put_view(AccountView)
       |> render("accounts.json", %{for: for_user, users: followers, as: :user})
     end
@@ -927,7 +927,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
         end
 
       conn
-      |> add_link_headers(:following, followers, user)
+      |> add_link_headers(followers)
       |> put_view(AccountView)
       |> render("accounts.json", %{for: for_user, users: followers, as: :user})
     end
@@ -1152,7 +1152,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       |> Enum.reverse()
 
     conn
-    |> add_link_headers(:favourites, activities)
+    |> add_link_headers(activities)
     |> put_view(StatusView)
     |> render("index.json", %{activities: activities, for: user, as: :activity})
   end
@@ -1179,7 +1179,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
         |> Enum.reverse()
 
       conn
-      |> add_link_headers(:favourites, activities)
+      |> add_link_headers(activities)
       |> put_view(StatusView)
       |> render("index.json", %{activities: activities, for: for_user, as: :activity})
     else
@@ -1200,7 +1200,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       |> Enum.map(fn b -> Map.put(b.activity, :bookmark, Map.delete(b, :activity)) end)
 
     conn
-    |> add_link_headers(:bookmarks, bookmarks)
+    |> add_link_headers(bookmarks)
     |> put_view(StatusView)
     |> render("index.json", %{activities: activities, for: user, as: :activity})
   end
@@ -1640,7 +1640,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       end)
 
     conn
-    |> add_link_headers(:conversations, participations)
+    |> add_link_headers(participations)
     |> json(conversations)
   end
 
diff --git a/lib/pleroma/web/pleroma_api/pleroma_api_controller.ex b/lib/pleroma/web/pleroma_api/pleroma_api_controller.ex
index f4df3b024..d17ccf84d 100644
--- a/lib/pleroma/web/pleroma_api/pleroma_api_controller.ex
+++ b/lib/pleroma/web/pleroma_api/pleroma_api_controller.ex
@@ -5,7 +5,7 @@
 defmodule Pleroma.Web.PleromaAPI.PleromaAPIController do
   use Pleroma.Web, :controller
 
-  import Pleroma.Web.ControllerHelper, only: [add_link_headers: 7]
+  import Pleroma.Web.ControllerHelper, only: [add_link_headers: 2]
 
   alias Pleroma.Conversation.Participation
   alias Pleroma.Notification
@@ -27,31 +27,22 @@ defmodule Pleroma.Web.PleromaAPI.PleromaAPIController do
         %{assigns: %{user: user}} = conn,
         %{"id" => participation_id} = params
       ) do
-    params =
-      params
-      |> Map.put("blocking_user", user)
-      |> Map.put("muting_user", user)
-      |> Map.put("user", user)
-
-    participation =
-      participation_id
-      |> Participation.get(preload: [:conversation])
+    participation = Participation.get(participation_id, preload: [:conversation])
 
     if user.id == participation.user_id do
+      params =
+        params
+        |> Map.put("blocking_user", user)
+        |> Map.put("muting_user", user)
+        |> Map.put("user", user)
+
       activities =
         participation.conversation.ap_id
         |> ActivityPub.fetch_activities_for_context(params)
         |> Enum.reverse()
 
       conn
-      |> add_link_headers(
-        :conversation_statuses,
-        activities,
-        participation_id,
-        params,
-        nil,
-        &pleroma_api_url/4
-      )
+      |> add_link_headers(activities)
       |> put_view(StatusView)
       |> render("index.json", %{activities: activities, for: user, as: :activity})
     end
-- 
cgit v1.2.3


From 67e430093187c50f307810e88ed0e73afe825b75 Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Fri, 30 Aug 2019 13:21:48 +0300
Subject: description formatters

---
 lib/pleroma/docs/formatter.ex | 69 +++++++++++++++++++++++++++++++++++++++++++
 lib/pleroma/docs/json.ex      | 15 ++++++++++
 lib/pleroma/docs/markdown.ex  | 67 +++++++++++++++++++++++++++++++++++++++++
 3 files changed, 151 insertions(+)
 create mode 100644 lib/pleroma/docs/formatter.ex
 create mode 100644 lib/pleroma/docs/json.ex
 create mode 100644 lib/pleroma/docs/markdown.ex

(limited to 'lib')

diff --git a/lib/pleroma/docs/formatter.ex b/lib/pleroma/docs/formatter.ex
new file mode 100644
index 000000000..a1c757936
--- /dev/null
+++ b/lib/pleroma/docs/formatter.ex
@@ -0,0 +1,69 @@
+defmodule Pleroma.Docs.Formatter do
+  @callback process(keyword()) :: {:ok, String.t()}
+
+  @spec process(module(), keyword()) :: {:ok, String.t()}
+  def process(implementation, descriptions) do
+    implementation.process(descriptions)
+  end
+
+  def uploaders_list do
+    {:ok, modules} = :application.get_key(:pleroma, :modules)
+
+    Enum.filter(modules, fn module ->
+      name_as_list = Module.split(module)
+
+      List.starts_with?(name_as_list, ["Pleroma", "Uploaders"]) and
+        List.last(name_as_list) in ["S3", "Local", "MDII"]
+    end)
+  end
+
+  def filters_list do
+    {:ok, modules} = :application.get_key(:pleroma, :modules)
+
+    Enum.filter(modules, fn module ->
+      name_as_list = Module.split(module)
+
+      List.starts_with?(name_as_list, ["Pleroma", "Upload", "Filter"])
+    end)
+  end
+
+  def mrf_list do
+    {:ok, modules} = :application.get_key(:pleroma, :modules)
+
+    Enum.filter(modules, fn module ->
+      name_as_list = Module.split(module)
+
+      List.starts_with?(name_as_list, ["Pleroma", "Web", "ActivityPub", "MRF"]) and
+        length(name_as_list) > 4
+    end)
+  end
+
+  def richmedia_parsers do
+    {:ok, modules} = :application.get_key(:pleroma, :modules)
+
+    Enum.filter(modules, fn module ->
+      name_as_list = Module.split(module)
+
+      List.starts_with?(name_as_list, ["Pleroma", "Web", "RichMedia", "Parsers"]) and
+        length(name_as_list) == 5
+    end)
+  end
+end
+
+defimpl Jason.Encoder, for: Tuple do
+  def encode(tuple, opts) do
+    Jason.Encode.list(Tuple.to_list(tuple), opts)
+  end
+end
+
+defimpl Jason.Encoder, for: [Regex, Function] do
+  def encode(term, opts) do
+    Jason.Encode.string(inspect(term), opts)
+  end
+end
+
+defimpl String.Chars, for: Regex do
+  def to_string(term) do
+    inspect(term)
+  end
+end
diff --git a/lib/pleroma/docs/json.ex b/lib/pleroma/docs/json.ex
new file mode 100644
index 000000000..38f015017
--- /dev/null
+++ b/lib/pleroma/docs/json.ex
@@ -0,0 +1,15 @@
+defmodule Pleroma.Docs.JSON do
+  @behaviour Pleroma.Docs.Formatter
+  def process(descriptions) do
+    config_path = "docs/generate_config.json"
+    {:ok, file} = File.open(config_path, [:write])
+    json = generate_json(descriptions)
+    IO.write(file, json)
+    :ok = File.close(file)
+    {:ok, config_path}
+  end
+
+  def generate_json(descriptions) do
+    Jason.encode!(descriptions)
+  end
+end
diff --git a/lib/pleroma/docs/markdown.ex b/lib/pleroma/docs/markdown.ex
new file mode 100644
index 000000000..27a096631
--- /dev/null
+++ b/lib/pleroma/docs/markdown.ex
@@ -0,0 +1,67 @@
+defmodule Pleroma.Docs.Markdown do
+  @behaviour Pleroma.Docs.Formatter
+
+  def process(descriptions) do
+    config_path = "docs/config.md"
+    {:ok, file} = File.open(config_path, [:write])
+    IO.write(file, "# Generated configuration\r\n\r\n")
+    IO.write(file, "Date of generation: #{Date.utc_today()}\r\n\r\n")
+
+    IO.write(
+      file,
+      "This file describe the configuration, it is recommended to edit the relevant *.secret.exs file instead of the others founds in the ``config`` directory.
+If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherwise it is ``dev.secret.exs``.\r\n\r\n"
+    )
+
+    for group <- descriptions do
+      if is_nil(group[:key]) do
+        IO.write(file, "## #{inspect(group[:group])}\r\n\r\n")
+      else
+        IO.write(file, "## #{inspect(group[:key])}\r\n\r\n")
+      end
+
+      IO.write(file, "Type: `#{group[:type]}`  \r\n")
+      IO.write(file, "#{group[:description]}  \r\n\r\n")
+
+      for child <- group[:children] do
+        print_child_header(file, child)
+
+        print_suggestions(file, child[:suggestions])
+
+        if child[:children] do
+          for subchild <- child[:children] do
+            print_child_header(file, subchild)
+
+            print_suggestions(file, subchild[:suggestions])
+          end
+        end
+      end
+
+      IO.write(file, "\r\n")
+    end
+
+    :ok = File.close(file)
+    {:ok, config_path}
+  end
+
+  defp print_suggestion(file, suggestion) when is_function(suggestion) do
+    IO.write(file, "    `#{inspect(suggestion.())}`\r\n")
+  end
+
+  defp print_suggestion(file, suggestion) do
+    IO.write(file, "    `#{inspect(suggestion)}`\r\n")
+  end
+
+  defp print_suggestions(file, suggestions) do
+    IO.write(file, "Suggestions:  \r\n")
+
+    for suggestion <- suggestions do
+      print_suggestion(file, suggestion)
+    end
+  end
+
+  defp print_child_header(file, child) do
+    IO.write(file, "* `#{inspect(child[:key])}`: #{child[:description]}  \r\n")
+    IO.write(file, "Type: `#{inspect(child[:type])}`  \r\n")
+  end
+end
-- 
cgit v1.2.3


From 511d93fa5486a58b63576ce1b8af551bbafe703f Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Fri, 30 Aug 2019 13:22:21 +0300
Subject: mix docs generates config.md

---
 lib/mix/tasks/pleroma/docs.ex | 40 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 40 insertions(+)
 create mode 100644 lib/mix/tasks/pleroma/docs.ex

(limited to 'lib')

diff --git a/lib/mix/tasks/pleroma/docs.ex b/lib/mix/tasks/pleroma/docs.ex
new file mode 100644
index 000000000..5ae3b8716
--- /dev/null
+++ b/lib/mix/tasks/pleroma/docs.ex
@@ -0,0 +1,40 @@
+defmodule Mix.Tasks.Pleroma.Docs do
+  use Mix.Task
+  import Mix.Pleroma
+
+  @shortdoc "Generates docs from descriptions.exs"
+  @moduledoc """
+  Generates docs from `descriptions.exs`.
+
+  Supports two formats: `markdown` and `json`.
+
+  ## Generate markdown docs
+
+  `mix pleroma.docs`
+
+  ## Generate json docs
+
+  `mix pleroma.docs json`s
+  """
+
+  def run(["json"]) do
+    do_run(Pleroma.Docs.JSON)
+  end
+
+  def run(_) do
+    do_run(Pleroma.Docs.Markdown)
+  end
+
+  defp do_run(implementation) do
+    start_pleroma()
+    descriptions = Config.Reader.read!("config/description.exs")
+
+    {:ok, file_path} =
+      Pleroma.Docs.Formatter.process(
+        implementation,
+        descriptions[:pleroma][:config_description]
+      )
+
+    Mix.shell().info([:green, "Markdown docs successfully generated to #{file_path}."])
+  end
+end
-- 
cgit v1.2.3


From a1f2dfb10a777592ea85d4bf8c5f91c859ec225b Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Fri, 30 Aug 2019 14:04:21 +0300
Subject: expanding regex sigils to use modifiers

---
 lib/pleroma/web/admin_api/config.ex | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/admin_api/config.ex b/lib/pleroma/web/admin_api/config.ex
index a10cc779b..1917a5580 100644
--- a/lib/pleroma/web/admin_api/config.ex
+++ b/lib/pleroma/web/admin_api/config.ex
@@ -90,6 +90,8 @@ defmodule Pleroma.Web.AdminAPI.Config do
     for v <- entity, into: [], do: do_convert(v)
   end
 
+  defp do_convert(%Regex{} = entity), do: inspect(entity)
+
   defp do_convert(entity) when is_map(entity) do
     for {k, v} <- entity, into: %{}, do: {do_convert(k), do_convert(v)}
   end
@@ -122,7 +124,7 @@ defmodule Pleroma.Web.AdminAPI.Config do
 
   def transform(entity), do: :erlang.term_to_binary(entity)
 
-  defp do_transform(%Regex{} = entity) when is_map(entity), do: entity
+  defp do_transform(%Regex{} = entity), do: entity
 
   defp do_transform(%{"tuple" => [":dispatch", [entity]]}) do
     {dispatch_settings, []} = do_eval(entity)
@@ -154,8 +156,15 @@ defmodule Pleroma.Web.AdminAPI.Config do
   defp do_transform(entity), do: entity
 
   defp do_transform_string("~r/" <> pattern) do
-    pattern = String.trim_trailing(pattern, "/")
-    ~r/#{pattern}/
+    modificator = String.split(pattern, "/") |> List.last()
+    pattern = String.trim_trailing(pattern, "/" <> modificator)
+
+    case modificator do
+      "" -> ~r/#{pattern}/
+      "i" -> ~r/#{pattern}/i
+      "u" -> ~r/#{pattern}/u
+      "s" -> ~r/#{pattern}/s
+    end
   end
 
   defp do_transform_string(":" <> atom), do: String.to_atom(atom)
-- 
cgit v1.2.3


From 0559c82bdb39cba019099d404723f38fed6e2aba Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Fri, 30 Aug 2019 14:27:55 +0300
Subject: fix

---
 lib/mix/tasks/pleroma/docs.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/mix/tasks/pleroma/docs.ex b/lib/mix/tasks/pleroma/docs.ex
index 5ae3b8716..d68e02383 100644
--- a/lib/mix/tasks/pleroma/docs.ex
+++ b/lib/mix/tasks/pleroma/docs.ex
@@ -27,7 +27,7 @@ defmodule Mix.Tasks.Pleroma.Docs do
 
   defp do_run(implementation) do
     start_pleroma()
-    descriptions = Config.Reader.read!("config/description.exs")
+    {descriptions, _paths} = Mix.Config.eval!("config/description.exs")
 
     {:ok, file_path} =
       Pleroma.Docs.Formatter.process(
-- 
cgit v1.2.3


From 6721301086674afe721f9eea478a2037756be93c Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Fri, 30 Aug 2019 19:14:01 +0300
Subject: some changes

---
 lib/mix/tasks/pleroma/docs.ex |  6 ++--
 lib/pleroma/docs/formatter.ex | 69 ----------------------------------------
 lib/pleroma/docs/generator.ex | 73 +++++++++++++++++++++++++++++++++++++++++++
 lib/pleroma/docs/json.ex      |  5 ++-
 lib/pleroma/docs/markdown.ex  | 36 ++++++++++++++-------
 5 files changed, 104 insertions(+), 85 deletions(-)
 delete mode 100644 lib/pleroma/docs/formatter.ex
 create mode 100644 lib/pleroma/docs/generator.ex

(limited to 'lib')

diff --git a/lib/mix/tasks/pleroma/docs.ex b/lib/mix/tasks/pleroma/docs.ex
index d68e02383..821ee74f9 100644
--- a/lib/mix/tasks/pleroma/docs.ex
+++ b/lib/mix/tasks/pleroma/docs.ex
@@ -8,11 +8,11 @@ defmodule Mix.Tasks.Pleroma.Docs do
 
   Supports two formats: `markdown` and `json`.
 
-  ## Generate markdown docs
+  ## Generate Markdown docs
 
   `mix pleroma.docs`
 
-  ## Generate json docs
+  ## Generate JSON docs
 
   `mix pleroma.docs json`s
   """
@@ -30,7 +30,7 @@ defmodule Mix.Tasks.Pleroma.Docs do
     {descriptions, _paths} = Mix.Config.eval!("config/description.exs")
 
     {:ok, file_path} =
-      Pleroma.Docs.Formatter.process(
+      Pleroma.Docs.Generator.process(
         implementation,
         descriptions[:pleroma][:config_description]
       )
diff --git a/lib/pleroma/docs/formatter.ex b/lib/pleroma/docs/formatter.ex
deleted file mode 100644
index a1c757936..000000000
--- a/lib/pleroma/docs/formatter.ex
+++ /dev/null
@@ -1,69 +0,0 @@
-defmodule Pleroma.Docs.Formatter do
-  @callback process(keyword()) :: {:ok, String.t()}
-
-  @spec process(module(), keyword()) :: {:ok, String.t()}
-  def process(implementation, descriptions) do
-    implementation.process(descriptions)
-  end
-
-  def uploaders_list do
-    {:ok, modules} = :application.get_key(:pleroma, :modules)
-
-    Enum.filter(modules, fn module ->
-      name_as_list = Module.split(module)
-
-      List.starts_with?(name_as_list, ["Pleroma", "Uploaders"]) and
-        List.last(name_as_list) in ["S3", "Local", "MDII"]
-    end)
-  end
-
-  def filters_list do
-    {:ok, modules} = :application.get_key(:pleroma, :modules)
-
-    Enum.filter(modules, fn module ->
-      name_as_list = Module.split(module)
-
-      List.starts_with?(name_as_list, ["Pleroma", "Upload", "Filter"])
-    end)
-  end
-
-  def mrf_list do
-    {:ok, modules} = :application.get_key(:pleroma, :modules)
-
-    Enum.filter(modules, fn module ->
-      name_as_list = Module.split(module)
-
-      List.starts_with?(name_as_list, ["Pleroma", "Web", "ActivityPub", "MRF"]) and
-        length(name_as_list) > 4
-    end)
-  end
-
-  def richmedia_parsers do
-    {:ok, modules} = :application.get_key(:pleroma, :modules)
-
-    Enum.filter(modules, fn module ->
-      name_as_list = Module.split(module)
-
-      List.starts_with?(name_as_list, ["Pleroma", "Web", "RichMedia", "Parsers"]) and
-        length(name_as_list) == 5
-    end)
-  end
-end
-
-defimpl Jason.Encoder, for: Tuple do
-  def encode(tuple, opts) do
-    Jason.Encode.list(Tuple.to_list(tuple), opts)
-  end
-end
-
-defimpl Jason.Encoder, for: [Regex, Function] do
-  def encode(term, opts) do
-    Jason.Encode.string(inspect(term), opts)
-  end
-end
-
-defimpl String.Chars, for: Regex do
-  def to_string(term) do
-    inspect(term)
-  end
-end
diff --git a/lib/pleroma/docs/generator.ex b/lib/pleroma/docs/generator.ex
new file mode 100644
index 000000000..e788712cc
--- /dev/null
+++ b/lib/pleroma/docs/generator.ex
@@ -0,0 +1,73 @@
+defmodule Pleroma.Docs.Generator do
+  @callback process(keyword()) :: {:ok, String.t()}
+
+  @spec process(module(), keyword()) :: {:ok, String.t()}
+  def process(implementation, descriptions) do
+    implementation.process(descriptions)
+  end
+
+  @spec uploaders_list() :: [module()]
+  def uploaders_list do
+    {:ok, modules} = :application.get_key(:pleroma, :modules)
+
+    Enum.filter(modules, fn module ->
+      name_as_list = Module.split(module)
+
+      List.starts_with?(name_as_list, ["Pleroma", "Uploaders"]) and
+        List.last(name_as_list) in ["S3", "Local", "MDII"]
+    end)
+  end
+
+  @spec filters_list() :: [module()]
+  def filters_list do
+    {:ok, modules} = :application.get_key(:pleroma, :modules)
+
+    Enum.filter(modules, fn module ->
+      name_as_list = Module.split(module)
+
+      List.starts_with?(name_as_list, ["Pleroma", "Upload", "Filter"])
+    end)
+  end
+
+  @spec mrf_list() :: [module()]
+  def mrf_list do
+    {:ok, modules} = :application.get_key(:pleroma, :modules)
+
+    Enum.filter(modules, fn module ->
+      name_as_list = Module.split(module)
+
+      List.starts_with?(name_as_list, ["Pleroma", "Web", "ActivityPub", "MRF"]) and
+        length(name_as_list) > 4
+    end)
+  end
+
+  @spec richmedia_parsers() :: [module()]
+  def richmedia_parsers do
+    {:ok, modules} = :application.get_key(:pleroma, :modules)
+
+    Enum.filter(modules, fn module ->
+      name_as_list = Module.split(module)
+
+      List.starts_with?(name_as_list, ["Pleroma", "Web", "RichMedia", "Parsers"]) and
+        length(name_as_list) == 5
+    end)
+  end
+end
+
+defimpl Jason.Encoder, for: Tuple do
+  def encode(tuple, opts) do
+    Jason.Encode.list(Tuple.to_list(tuple), opts)
+  end
+end
+
+defimpl Jason.Encoder, for: [Regex, Function] do
+  def encode(term, opts) do
+    Jason.Encode.string(inspect(term), opts)
+  end
+end
+
+defimpl String.Chars, for: Regex do
+  def to_string(term) do
+    inspect(term)
+  end
+end
diff --git a/lib/pleroma/docs/json.ex b/lib/pleroma/docs/json.ex
index 38f015017..aed730e78 100644
--- a/lib/pleroma/docs/json.ex
+++ b/lib/pleroma/docs/json.ex
@@ -1,5 +1,7 @@
 defmodule Pleroma.Docs.JSON do
-  @behaviour Pleroma.Docs.Formatter
+  @behaviour Pleroma.Docs.Generator
+
+  @spec process(keyword()) :: {:ok, String.t()}
   def process(descriptions) do
     config_path = "docs/generate_config.json"
     {:ok, file} = File.open(config_path, [:write])
@@ -9,6 +11,7 @@ defmodule Pleroma.Docs.JSON do
     {:ok, config_path}
   end
 
+  @spec generate_json([keyword()]) :: String.t()
   def generate_json(descriptions) do
     Jason.encode!(descriptions)
   end
diff --git a/lib/pleroma/docs/markdown.ex b/lib/pleroma/docs/markdown.ex
index 27a096631..c66640bf1 100644
--- a/lib/pleroma/docs/markdown.ex
+++ b/lib/pleroma/docs/markdown.ex
@@ -1,16 +1,17 @@
 defmodule Pleroma.Docs.Markdown do
-  @behaviour Pleroma.Docs.Formatter
+  @behaviour Pleroma.Docs.Generator
 
+  @spec process(keyword()) :: {:ok, String.t()}
   def process(descriptions) do
     config_path = "docs/config.md"
     {:ok, file} = File.open(config_path, [:write])
-    IO.write(file, "# Generated configuration\r\n\r\n")
+    IO.write(file, "# Configuration\r\n\r\n")
     IO.write(file, "Date of generation: #{Date.utc_today()}\r\n\r\n")
 
     IO.write(
       file,
-      "This file describe the configuration, it is recommended to edit the relevant *.secret.exs file instead of the others founds in the ``config`` directory.
-If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherwise it is ``dev.secret.exs``.\r\n\r\n"
+      "This file describe the configuration, it is recommended to edit the relevant `*.secret.exs` file instead of the others founds in the ``config`` directory.  \r\n
+      If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherwise it is ``dev.secret.exs``.\r\n\r\n"
     )
 
     for group <- descriptions do
@@ -20,7 +21,6 @@ If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherw
         IO.write(file, "## #{inspect(group[:key])}\r\n\r\n")
       end
 
-      IO.write(file, "Type: `#{group[:type]}`  \r\n")
       IO.write(file, "#{group[:description]}  \r\n\r\n")
 
       for child <- group[:children] do
@@ -44,24 +44,36 @@ If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherw
     {:ok, config_path}
   end
 
+  defp print_suggestion(file, suggestion) when is_list(suggestion) do
+    IO.write(file, "    `#{inspect(suggestion)}`\r\n")
+  end
+
   defp print_suggestion(file, suggestion) when is_function(suggestion) do
     IO.write(file, "    `#{inspect(suggestion.())}`\r\n")
   end
 
-  defp print_suggestion(file, suggestion) do
-    IO.write(file, "    `#{inspect(suggestion)}`\r\n")
+  defp print_suggestion(file, suggestion, as_list \\ false) do
+    list_mark = if as_list, do: "*", else: ""
+    IO.write(file, "    #{list_mark} `#{inspect(suggestion)}`\r\n")
   end
 
+  defp print_suggestions(_file, nil), do: nil
+
   defp print_suggestions(file, suggestions) do
-    IO.write(file, "Suggestions:  \r\n")
+    IO.write(file, "    Suggestions:  \r\n")
 
-    for suggestion <- suggestions do
-      print_suggestion(file, suggestion)
+    if length(suggestions) > 1 do
+      for suggestion <- suggestions do
+        print_suggestion(file, suggestion, true)
+      end
+    else
+      print_suggestion(file, List.first(suggestions))
     end
   end
 
   defp print_child_header(file, child) do
-    IO.write(file, "* `#{inspect(child[:key])}`: #{child[:description]}  \r\n")
-    IO.write(file, "Type: `#{inspect(child[:type])}`  \r\n")
+    IO.write(file, "* `#{inspect(child[:key])}`  \r\n")
+    IO.write(file, "    #{child[:description]}  \r\n")
+    IO.write(file, "    Type: `#{inspect(child[:type])}`  \r\n")
   end
 end
-- 
cgit v1.2.3


From 8f5ee7db06e379e44505744fd21e59cc46432ac3 Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Fri, 30 Aug 2019 19:59:13 +0300
Subject: typo fix

---
 lib/mix/tasks/pleroma/docs.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/mix/tasks/pleroma/docs.ex b/lib/mix/tasks/pleroma/docs.ex
index 821ee74f9..fb8a2a014 100644
--- a/lib/mix/tasks/pleroma/docs.ex
+++ b/lib/mix/tasks/pleroma/docs.ex
@@ -14,7 +14,7 @@ defmodule Mix.Tasks.Pleroma.Docs do
 
   ## Generate JSON docs
 
-  `mix pleroma.docs json`s
+  `mix pleroma.docs json`
   """
 
   def run(["json"]) do
-- 
cgit v1.2.3


From 0624e06a9c981264f238041a2ad22392d4a8f792 Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Fri, 30 Aug 2019 20:14:18 +0300
Subject: little fix

---
 lib/pleroma/docs/markdown.ex | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/docs/markdown.ex b/lib/pleroma/docs/markdown.ex
index c66640bf1..797ce73bf 100644
--- a/lib/pleroma/docs/markdown.ex
+++ b/lib/pleroma/docs/markdown.ex
@@ -10,8 +10,8 @@ defmodule Pleroma.Docs.Markdown do
 
     IO.write(
       file,
-      "This file describe the configuration, it is recommended to edit the relevant `*.secret.exs` file instead of the others founds in the ``config`` directory.  \r\n
-      If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherwise it is ``dev.secret.exs``.\r\n\r\n"
+      "This file describe the configuration, it is recommended to edit the relevant `*.secret.exs` file instead of the others founds in the ``config`` directory.  \r\n\r\n" <>
+        " If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherwise it is ``dev.secret.exs``.\r\n\r\n"
     )
 
     for group <- descriptions do
-- 
cgit v1.2.3


From 35757b6d0eb7b59d511bfea6a166683e18d6aa97 Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Tue, 3 Sep 2019 09:45:54 +0300
Subject: don't add behaviour to suggestions

---
 lib/pleroma/docs/generator.ex | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/pleroma/docs/generator.ex b/lib/pleroma/docs/generator.ex
index e788712cc..aa578eee2 100644
--- a/lib/pleroma/docs/generator.ex
+++ b/lib/pleroma/docs/generator.ex
@@ -14,7 +14,7 @@ defmodule Pleroma.Docs.Generator do
       name_as_list = Module.split(module)
 
       List.starts_with?(name_as_list, ["Pleroma", "Uploaders"]) and
-        List.last(name_as_list) in ["S3", "Local", "MDII"]
+        List.last(name_as_list) != "Uploader"
     end)
   end
 
-- 
cgit v1.2.3


From 57dc59d98d0ae5b4315bbc5c8a9c7ca59f6341f9 Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Tue, 3 Sep 2019 12:31:43 +0300
Subject: little fix

---
 lib/mix/tasks/pleroma/docs.ex | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

(limited to 'lib')

diff --git a/lib/mix/tasks/pleroma/docs.ex b/lib/mix/tasks/pleroma/docs.ex
index fb8a2a014..4be53ce75 100644
--- a/lib/mix/tasks/pleroma/docs.ex
+++ b/lib/mix/tasks/pleroma/docs.ex
@@ -35,6 +35,8 @@ defmodule Mix.Tasks.Pleroma.Docs do
         descriptions[:pleroma][:config_description]
       )
 
-    Mix.shell().info([:green, "Markdown docs successfully generated to #{file_path}."])
+    type = if implementation == Pleroma.Docs.Markdown, do: "Markdown", else: "JSON"
+
+    Mix.shell().info([:green, "#{type} docs successfully generated to #{file_path}."])
   end
 end
-- 
cgit v1.2.3


From 5ff12e7df10dd70f70e5929ede5dc7570e066c57 Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Tue, 3 Sep 2019 19:22:25 +0300
Subject: some changes

---
 lib/pleroma/docs/markdown.ex | 33 ++++++++++++++++-----------------
 1 file changed, 16 insertions(+), 17 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/docs/markdown.ex b/lib/pleroma/docs/markdown.ex
index 797ce73bf..24930cc9f 100644
--- a/lib/pleroma/docs/markdown.ex
+++ b/lib/pleroma/docs/markdown.ex
@@ -4,24 +4,24 @@ defmodule Pleroma.Docs.Markdown do
   @spec process(keyword()) :: {:ok, String.t()}
   def process(descriptions) do
     config_path = "docs/config.md"
-    {:ok, file} = File.open(config_path, [:write])
-    IO.write(file, "# Configuration\r\n\r\n")
-    IO.write(file, "Date of generation: #{Date.utc_today()}\r\n\r\n")
+    {:ok, file} = File.open(config_path, [:utf8, :write])
+    IO.write(file, "# Configuration\n")
+    IO.write(file, "Date of generation: #{Date.utc_today()}\n\n")
 
     IO.write(
       file,
-      "This file describe the configuration, it is recommended to edit the relevant `*.secret.exs` file instead of the others founds in the ``config`` directory.  \r\n\r\n" <>
-        " If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherwise it is ``dev.secret.exs``.\r\n\r\n"
+      "This file describe the configuration, it is recommended to edit the relevant `*.secret.exs` file instead of the others founds in the ``config`` directory.\n\n" <>
+        "If you run Pleroma with ``MIX_ENV=prod`` the file is ``prod.secret.exs``, otherwise it is ``dev.secret.exs``.\n\n"
     )
 
     for group <- descriptions do
       if is_nil(group[:key]) do
-        IO.write(file, "## #{inspect(group[:group])}\r\n\r\n")
+        IO.write(file, "## #{inspect(group[:group])}\n")
       else
-        IO.write(file, "## #{inspect(group[:key])}\r\n\r\n")
+        IO.write(file, "## #{inspect(group[:key])}\n")
       end
 
-      IO.write(file, "#{group[:description]}  \r\n\r\n")
+      IO.write(file, "#{group[:description]}\n")
 
       for child <- group[:children] do
         print_child_header(file, child)
@@ -37,7 +37,7 @@ defmodule Pleroma.Docs.Markdown do
         end
       end
 
-      IO.write(file, "\r\n")
+      IO.write(file, "\n")
     end
 
     :ok = File.close(file)
@@ -45,22 +45,22 @@ defmodule Pleroma.Docs.Markdown do
   end
 
   defp print_suggestion(file, suggestion) when is_list(suggestion) do
-    IO.write(file, "    `#{inspect(suggestion)}`\r\n")
+    IO.write(file, "  `#{inspect(suggestion)}`\n")
   end
 
   defp print_suggestion(file, suggestion) when is_function(suggestion) do
-    IO.write(file, "    `#{inspect(suggestion.())}`\r\n")
+    IO.write(file, "  `#{inspect(suggestion.())}`\n")
   end
 
   defp print_suggestion(file, suggestion, as_list \\ false) do
-    list_mark = if as_list, do: "*", else: ""
-    IO.write(file, "    #{list_mark} `#{inspect(suggestion)}`\r\n")
+    list_mark = if as_list, do: "- ", else: ""
+    IO.write(file, "  #{list_mark}`#{inspect(suggestion)}`\n")
   end
 
   defp print_suggestions(_file, nil), do: nil
 
   defp print_suggestions(file, suggestions) do
-    IO.write(file, "    Suggestions:  \r\n")
+    IO.write(file, "Suggestions:\n")
 
     if length(suggestions) > 1 do
       for suggestion <- suggestions do
@@ -72,8 +72,7 @@ defmodule Pleroma.Docs.Markdown do
   end
 
   defp print_child_header(file, child) do
-    IO.write(file, "* `#{inspect(child[:key])}`  \r\n")
-    IO.write(file, "    #{child[:description]}  \r\n")
-    IO.write(file, "    Type: `#{inspect(child[:type])}`  \r\n")
+    IO.write(file, "- `#{inspect(child[:key])}` -`#{inspect(child[:type])}`  \n")
+    IO.write(file, "#{child[:description]}  \n")
   end
 end
-- 
cgit v1.2.3


From be32d90a0cdee77f4cd6ecbbbade1748c88637b8 Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Tue, 3 Sep 2019 20:06:22 +0300
Subject: little refactor

---
 lib/pleroma/docs/json.ex | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/docs/json.ex b/lib/pleroma/docs/json.ex
index aed730e78..18ba01d58 100644
--- a/lib/pleroma/docs/json.ex
+++ b/lib/pleroma/docs/json.ex
@@ -4,11 +4,13 @@ defmodule Pleroma.Docs.JSON do
   @spec process(keyword()) :: {:ok, String.t()}
   def process(descriptions) do
     config_path = "docs/generate_config.json"
-    {:ok, file} = File.open(config_path, [:write])
-    json = generate_json(descriptions)
-    IO.write(file, json)
-    :ok = File.close(file)
-    {:ok, config_path}
+
+    with {:ok, file} <- File.open(config_path, [:write]),
+         json <- generate_json(descriptions),
+         :ok <- IO.write(file, json),
+         :ok <- File.close(file) do
+      {:ok, config_path}
+    end
   end
 
   @spec generate_json([keyword()]) :: String.t()
-- 
cgit v1.2.3


From 38b29779c3372741ce1d7b652922058ad2fda79c Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Tue, 3 Sep 2019 20:11:32 +0300
Subject: refactoring

---
 lib/mix/tasks/pleroma/docs.ex | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

(limited to 'lib')

diff --git a/lib/mix/tasks/pleroma/docs.ex b/lib/mix/tasks/pleroma/docs.ex
index 4be53ce75..0d2663648 100644
--- a/lib/mix/tasks/pleroma/docs.ex
+++ b/lib/mix/tasks/pleroma/docs.ex
@@ -27,16 +27,16 @@ defmodule Mix.Tasks.Pleroma.Docs do
 
   defp do_run(implementation) do
     start_pleroma()
-    {descriptions, _paths} = Mix.Config.eval!("config/description.exs")
 
-    {:ok, file_path} =
-      Pleroma.Docs.Generator.process(
-        implementation,
-        descriptions[:pleroma][:config_description]
-      )
+    with {descriptions, _paths} <- Mix.Config.eval!("config/description.exs"),
+         {:ok, file_path} <-
+           Pleroma.Docs.Generator.process(
+             implementation,
+             descriptions[:pleroma][:config_description]
+           ) do
+      type = if implementation == Pleroma.Docs.Markdown, do: "Markdown", else: "JSON"
 
-    type = if implementation == Pleroma.Docs.Markdown, do: "Markdown", else: "JSON"
-
-    Mix.shell().info([:green, "#{type} docs successfully generated to #{file_path}."])
+      Mix.shell().info([:green, "#{type} docs successfully generated to #{file_path}."])
+    end
   end
 end
-- 
cgit v1.2.3


From 74e4c72c4a0a079e8e2a245ec9b1d22684baf2e1 Mon Sep 17 00:00:00 2001
From: Mark Felder <feld@FreeBSD.org>
Date: Wed, 11 Sep 2019 16:16:09 -0500
Subject: Fix double quotes in error logs

Example:

pleroma: [error] Couldn't fetch ""https://pleroma.soykaf.com/objects/6288a14b-0623-40fc-a26a-0d358f8a11ca"", error: nil
---
 lib/pleroma/web/activity_pub/transmogrifier.ex | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
index 468961bd0..350b83abb 100644
--- a/lib/pleroma/web/activity_pub/transmogrifier.ex
+++ b/lib/pleroma/web/activity_pub/transmogrifier.ex
@@ -185,12 +185,12 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
             |> Map.put("context", replied_object.data["context"] || object["conversation"])
           else
             e ->
-              Logger.error("Couldn't fetch \"#{inspect(in_reply_to_id)}\", error: #{inspect(e)}")
+              Logger.error("Couldn't fetch #{inspect(in_reply_to_id)}, error: #{inspect(e)}")
               object
           end
 
         e ->
-          Logger.error("Couldn't fetch \"#{inspect(in_reply_to_id)}\", error: #{inspect(e)}")
+          Logger.error("Couldn't fetch #{inspect(in_reply_to_id)}, error: #{inspect(e)}")
           object
       end
     else
-- 
cgit v1.2.3


From 39dc9b470c7ad8348a13f181039f11d14a42fa2b Mon Sep 17 00:00:00 2001
From: Egor Kislitsyn <egor@kislitsyn.com>
Date: Tue, 3 Sep 2019 21:58:30 +0700
Subject: Cleanup Pleroma.Activity and Pleroma.Web.ActivityPub.Utils

---
 lib/pleroma/activity.ex               | 193 ++++++++--------------------------
 lib/pleroma/activity/queries.ex       |  32 ++++--
 lib/pleroma/user.ex                   |   2 +-
 lib/pleroma/web/activity_pub/utils.ex | 167 +++++++++++------------------
 4 files changed, 127 insertions(+), 267 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/activity.ex b/lib/pleroma/activity.ex
index 44f1e3011..ec558168a 100644
--- a/lib/pleroma/activity.ex
+++ b/lib/pleroma/activity.ex
@@ -6,6 +6,7 @@ defmodule Pleroma.Activity do
   use Ecto.Schema
 
   alias Pleroma.Activity
+  alias Pleroma.Activity.Queries
   alias Pleroma.ActivityExpiration
   alias Pleroma.Bookmark
   alias Pleroma.Notification
@@ -65,8 +66,8 @@ defmodule Pleroma.Activity do
     timestamps()
   end
 
-  def with_joined_object(query) do
-    join(query, :inner, [activity], o in Object,
+  def with_joined_object(query, join_type \\ :inner) do
+    join(query, join_type, [activity], o in Object,
       on:
         fragment(
           "(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
@@ -78,10 +79,10 @@ defmodule Pleroma.Activity do
     )
   end
 
-  def with_preloaded_object(query) do
+  def with_preloaded_object(query, join_type \\ :inner) do
     query
     |> has_named_binding?(:object)
-    |> if(do: query, else: with_joined_object(query))
+    |> if(do: query, else: with_joined_object(query, join_type))
     |> preload([activity, object: object], object: object)
   end
 
@@ -107,12 +108,9 @@ defmodule Pleroma.Activity do
   def with_set_thread_muted_field(query, _), do: query
 
   def get_by_ap_id(ap_id) do
-    Repo.one(
-      from(
-        activity in Activity,
-        where: fragment("(?)->>'id' = ?", activity.data, ^to_string(ap_id))
-      )
-    )
+    ap_id
+    |> Queries.by_ap_id()
+    |> Repo.one()
   end
 
   def get_bookmark(%Activity{} = activity, %User{} = user) do
@@ -133,21 +131,10 @@ defmodule Pleroma.Activity do
   end
 
   def get_by_ap_id_with_object(ap_id) do
-    Repo.one(
-      from(
-        activity in Activity,
-        where: fragment("(?)->>'id' = ?", activity.data, ^to_string(ap_id)),
-        left_join: o in Object,
-        on:
-          fragment(
-            "(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
-            o.data,
-            activity.data,
-            activity.data
-          ),
-        preload: [object: o]
-      )
-    )
+    ap_id
+    |> Queries.by_ap_id()
+    |> with_preloaded_object(:left)
+    |> Repo.one()
   end
 
   def get_by_id(id) do
@@ -158,18 +145,9 @@ defmodule Pleroma.Activity do
   end
 
   def get_by_id_with_object(id) do
-    from(activity in Activity,
-      where: activity.id == ^id,
-      inner_join: o in Object,
-      on:
-        fragment(
-          "(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
-          o.data,
-          activity.data,
-          activity.data
-        ),
-      preload: [object: o]
-    )
+    Activity
+    |> where(id: ^id)
+    |> with_preloaded_object()
     |> Repo.one()
   end
 
@@ -180,51 +158,21 @@ defmodule Pleroma.Activity do
     |> Repo.all()
   end
 
-  def by_object_ap_id(ap_id) do
-    from(
-      activity in Activity,
-      where:
-        fragment(
-          "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-          activity.data,
-          activity.data,
-          ^to_string(ap_id)
-        )
-    )
-  end
-
-  def create_by_object_ap_id(ap_ids) when is_list(ap_ids) do
-    from(
-      activity in Activity,
-      where:
-        fragment(
-          "coalesce((?)->'object'->>'id', (?)->>'object') = ANY(?)",
-          activity.data,
-          activity.data,
-          ^ap_ids
-        ),
-      where: fragment("(?)->>'type' = 'Create'", activity.data)
-    )
-  end
-
-  def create_by_object_ap_id(ap_id) when is_binary(ap_id) do
-    from(
-      activity in Activity,
-      where:
-        fragment(
-          "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-          activity.data,
-          activity.data,
-          ^to_string(ap_id)
-        ),
-      where: fragment("(?)->>'type' = 'Create'", activity.data)
-    )
+  @doc """
+  Accepts `ap_id` or list of `ap_id`.
+  Returns a query.
+  """
+  @spec create_by_object_ap_id(String.t() | [String.t()]) :: Ecto.Queryable.t()
+  def create_by_object_ap_id(ap_id) do
+    ap_id
+    |> Queries.by_object_id()
+    |> Queries.by_type("Create")
   end
 
-  def create_by_object_ap_id(_), do: nil
-
   def get_all_create_by_object_ap_id(ap_id) do
-    Repo.all(create_by_object_ap_id(ap_id))
+    ap_id
+    |> create_by_object_ap_id()
+    |> Repo.all()
   end
 
   def get_create_by_object_ap_id(ap_id) when is_binary(ap_id) do
@@ -235,54 +183,17 @@ defmodule Pleroma.Activity do
 
   def get_create_by_object_ap_id(_), do: nil
 
-  def create_by_object_ap_id_with_object(ap_ids) when is_list(ap_ids) do
-    from(
-      activity in Activity,
-      where:
-        fragment(
-          "coalesce((?)->'object'->>'id', (?)->>'object') = ANY(?)",
-          activity.data,
-          activity.data,
-          ^ap_ids
-        ),
-      where: fragment("(?)->>'type' = 'Create'", activity.data),
-      inner_join: o in Object,
-      on:
-        fragment(
-          "(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
-          o.data,
-          activity.data,
-          activity.data
-        ),
-      preload: [object: o]
-    )
-  end
-
-  def create_by_object_ap_id_with_object(ap_id) when is_binary(ap_id) do
-    from(
-      activity in Activity,
-      where:
-        fragment(
-          "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-          activity.data,
-          activity.data,
-          ^to_string(ap_id)
-        ),
-      where: fragment("(?)->>'type' = 'Create'", activity.data),
-      inner_join: o in Object,
-      on:
-        fragment(
-          "(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
-          o.data,
-          activity.data,
-          activity.data
-        ),
-      preload: [object: o]
-    )
+  @doc """
+  Accepts `ap_id` or list of `ap_id`.
+  Returns a query.
+  """
+  @spec create_by_object_ap_id_with_object(String.t() | [String.t()]) :: Ecto.Queryable.t()
+  def create_by_object_ap_id_with_object(ap_id) do
+    ap_id
+    |> create_by_object_ap_id()
+    |> with_preloaded_object()
   end
 
-  def create_by_object_ap_id_with_object(_), do: nil
-
   def get_create_by_object_ap_id_with_object(ap_id) when is_binary(ap_id) do
     ap_id
     |> create_by_object_ap_id_with_object()
@@ -306,7 +217,8 @@ defmodule Pleroma.Activity do
   def normalize(_), do: nil
 
   def delete_by_ap_id(id) when is_binary(id) do
-    by_object_ap_id(id)
+    id
+    |> Queries.by_object_id()
     |> select([u], u)
     |> Repo.delete_all()
     |> elem(1)
@@ -350,31 +262,10 @@ defmodule Pleroma.Activity do
   end
 
   def follow_requests_for_actor(%Pleroma.User{ap_id: ap_id}) do
-    from(
-      a in Activity,
-      where:
-        fragment(
-          "? ->> 'type' = 'Follow'",
-          a.data
-        ),
-      where:
-        fragment(
-          "? ->> 'state' = 'pending'",
-          a.data
-        ),
-      where:
-        fragment(
-          "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-          a.data,
-          a.data,
-          ^ap_id
-        )
-    )
-  end
-
-  @spec query_by_actor(actor()) :: Ecto.Query.t()
-  def query_by_actor(actor) do
-    from(a in Activity, where: a.actor == ^actor)
+    ap_id
+    |> Queries.by_object_id()
+    |> Queries.by_type("Follow")
+    |> where([a], fragment("? ->> 'state' = 'pending'", a.data))
   end
 
   def restrict_deactivated_users(query) do
diff --git a/lib/pleroma/activity/queries.ex b/lib/pleroma/activity/queries.ex
index aa5b29566..13fa33831 100644
--- a/lib/pleroma/activity/queries.ex
+++ b/lib/pleroma/activity/queries.ex
@@ -13,6 +13,14 @@ defmodule Pleroma.Activity.Queries do
 
   alias Pleroma.Activity
 
+  @spec by_ap_id(query, String.t()) :: query
+  def by_ap_id(query \\ Activity, ap_id) do
+    from(
+      activity in query,
+      where: fragment("(?)->>'id' = ?", activity.data, ^to_string(ap_id))
+    )
+  end
+
   @spec by_actor(query, String.t()) :: query
   def by_actor(query \\ Activity, actor) do
     from(
@@ -21,8 +29,23 @@ defmodule Pleroma.Activity.Queries do
     )
   end
 
-  @spec by_object_id(query, String.t()) :: query
-  def by_object_id(query \\ Activity, object_id) do
+  @spec by_object_id(query, String.t() | [String.t()]) :: query
+  def by_object_id(query \\ Activity, object_id)
+
+  def by_object_id(query, object_ids) when is_list(object_ids) do
+    from(
+      activity in query,
+      where:
+        fragment(
+          "coalesce((?)->'object'->>'id', (?)->>'object') = ANY(?)",
+          activity.data,
+          activity.data,
+          ^object_ids
+        )
+    )
+  end
+
+  def by_object_id(query, object_id) when is_binary(object_id) do
     from(activity in query,
       where:
         fragment(
@@ -41,9 +64,4 @@ defmodule Pleroma.Activity.Queries do
       where: fragment("(?)->>'type' = ?", activity.data, ^activity_type)
     )
   end
-
-  @spec limit(query, pos_integer()) :: query
-  def limit(query \\ Activity, limit) do
-    from(activity in query, limit: ^limit)
-  end
 end
diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 3aa245f2a..ceca11def 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -1219,7 +1219,7 @@ defmodule Pleroma.User do
 
   def delete_user_activities(%User{ap_id: ap_id} = user) do
     ap_id
-    |> Activity.query_by_actor()
+    |> Activity.Queries.by_actor()
     |> RepoStreamer.chunk_stream(50)
     |> Stream.each(fn activities ->
       Enum.each(activities, &delete_activity(&1))
diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex
index c9c0c3763..47917f5d3 100644
--- a/lib/pleroma/web/activity_pub/utils.ex
+++ b/lib/pleroma/web/activity_pub/utils.ex
@@ -85,15 +85,13 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   defp extract_list(_), do: []
 
   def maybe_splice_recipient(ap_id, params) do
-    need_splice =
+    need_splice? =
       !recipient_in_collection(ap_id, params["to"]) &&
         !recipient_in_collection(ap_id, params["cc"])
 
-    cc_list = extract_list(params["cc"])
-
-    if need_splice do
-      params
-      |> Map.put("cc", [ap_id | cc_list])
+    if need_splice? do
+      cc_list = extract_list(params["cc"])
+      Map.put(params, "cc", [ap_id | cc_list])
     else
       params
     end
@@ -139,7 +137,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
       "object" => object
     }
 
-    Notification.get_notified_from_activity(%Activity{data: fake_create_activity}, false)
+    get_notified_from_object(fake_create_activity)
   end
 
   def get_notified_from_object(object) do
@@ -188,9 +186,9 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   Adds an id and a published data if they aren't there,
   also adds it to an included object
   """
-  def lazy_put_activity_defaults(map, fake \\ false) do
+  def lazy_put_activity_defaults(map, fake? \\ false) do
     map =
-      unless fake do
+      if not fake? do
         %{data: %{"id" => context}, id: context_id} = create_context(map["context"])
 
         map
@@ -207,7 +205,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
       end
 
     if is_map(map["object"]) do
-      object = lazy_put_object_defaults(map["object"], map, fake)
+      object = lazy_put_object_defaults(map["object"], map, fake?)
       %{map | "object" => object}
     else
       map
@@ -217,9 +215,9 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   @doc """
   Adds an id and published date if they aren't there.
   """
-  def lazy_put_object_defaults(map, activity \\ %{}, fake)
+  def lazy_put_object_defaults(map, activity \\ %{}, fake?)
 
-  def lazy_put_object_defaults(map, activity, true = _fake) do
+  def lazy_put_object_defaults(map, activity, true = _fake?) do
     map
     |> Map.put_new_lazy("published", &make_date/0)
     |> Map.put_new("id", "pleroma:fake_object_id")
@@ -228,7 +226,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
     |> Map.put_new("context_id", activity["context_id"])
   end
 
-  def lazy_put_object_defaults(map, activity, _fake) do
+  def lazy_put_object_defaults(map, activity, _fake?) do
     map
     |> Map.put_new_lazy("id", &generate_object_id/0)
     |> Map.put_new_lazy("published", &make_date/0)
@@ -242,9 +240,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   def insert_full_object(%{"object" => %{"type" => type} = object_data} = map)
       when is_map(object_data) and type in @supported_object_types do
     with {:ok, object} <- Object.create(object_data) do
-      map =
-        map
-        |> Map.put("object", object.data["id"])
+      map = Map.put(map, "object", object.data["id"])
 
       {:ok, map, object}
     end
@@ -263,7 +259,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do
     |> Activity.Queries.by_actor()
     |> Activity.Queries.by_object_id(id)
     |> Activity.Queries.by_type("Like")
-    |> Activity.Queries.limit(1)
+    |> limit(1)
     |> Repo.one()
   end
 
@@ -380,12 +376,11 @@ defmodule Pleroma.Web.ActivityPub.Utils do
         %Activity{data: %{"actor" => actor, "object" => object}} = activity,
         state
       ) do
-    with new_data <-
-           activity.data
-           |> Map.put("state", state),
-         changeset <- Changeset.change(activity, data: new_data),
-         {:ok, activity} <- Repo.update(changeset),
-         _ <- User.set_follow_state_cache(actor, object, state) do
+    new_data = Map.put(activity.data, "state", state)
+    changeset = Changeset.change(activity, data: new_data)
+
+    with {:ok, activity} <- Repo.update(changeset) do
+      User.set_follow_state_cache(actor, object, state)
       {:ok, activity}
     end
   end
@@ -410,28 +405,14 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   end
 
   def fetch_latest_follow(%User{ap_id: follower_id}, %User{ap_id: followed_id}) do
-    query =
-      from(
-        activity in Activity,
-        where:
-          fragment(
-            "? ->> 'type' = 'Follow'",
-            activity.data
-          ),
-        where: activity.actor == ^follower_id,
-        # this is to use the index
-        where:
-          fragment(
-            "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-            activity.data,
-            activity.data,
-            ^followed_id
-          ),
-        order_by: [fragment("? desc nulls last", activity.id)],
-        limit: 1
-      )
-
-    Repo.one(query)
+    "Follow"
+    |> Activity.Queries.by_type()
+    |> where(actor: ^follower_id)
+    # this is to use the index
+    |> Activity.Queries.by_object_id(followed_id)
+    |> order_by([activity], fragment("? desc nulls last", activity.id))
+    |> limit(1)
+    |> Repo.one()
   end
 
   #### Announce-related helpers
@@ -439,23 +420,13 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   @doc """
   Retruns an existing announce activity if the notice has already been announced
   """
-  def get_existing_announce(actor, %{data: %{"id" => id}}) do
-    query =
-      from(
-        activity in Activity,
-        where: activity.actor == ^actor,
-        # this is to use the index
-        where:
-          fragment(
-            "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-            activity.data,
-            activity.data,
-            ^id
-          ),
-        where: fragment("(?)->>'type' = 'Announce'", activity.data)
-      )
-
-    Repo.one(query)
+  def get_existing_announce(actor, %{data: %{"id" => ap_id}}) do
+    "Announce"
+    |> Activity.Queries.by_type()
+    |> where(actor: ^actor)
+    # this is to use the index
+    |> Activity.Queries.by_object_id(ap_id)
+    |> Repo.one()
   end
 
   @doc """
@@ -538,11 +509,13 @@ defmodule Pleroma.Web.ActivityPub.Utils do
         object
       ) do
     announcements =
-      if is_list(object.data["announcements"]), do: object.data["announcements"], else: []
+      if is_list(object.data["announcements"]) do
+        Enum.uniq([actor | object.data["announcements"]])
+      else
+        [actor]
+      end
 
-    with announcements <- [actor | announcements] |> Enum.uniq() do
-      update_element_in_object("announcement", announcements, object)
-    end
+    update_element_in_object("announcement", announcements, object)
   end
 
   def add_announce_to_object(_, object), do: {:ok, object}
@@ -570,28 +543,14 @@ defmodule Pleroma.Web.ActivityPub.Utils do
 
   #### Block-related helpers
   def fetch_latest_block(%User{ap_id: blocker_id}, %User{ap_id: blocked_id}) do
-    query =
-      from(
-        activity in Activity,
-        where:
-          fragment(
-            "? ->> 'type' = 'Block'",
-            activity.data
-          ),
-        where: activity.actor == ^blocker_id,
-        # this is to use the index
-        where:
-          fragment(
-            "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
-            activity.data,
-            activity.data,
-            ^blocked_id
-          ),
-        order_by: [fragment("? desc nulls last", activity.id)],
-        limit: 1
-      )
-
-    Repo.one(query)
+    "Block"
+    |> Activity.Queries.by_type()
+    |> where(actor: ^blocker_id)
+    # this is to use the index
+    |> Activity.Queries.by_object_id(blocked_id)
+    |> order_by([activity], fragment("? desc nulls last", activity.id))
+    |> limit(1)
+    |> Repo.one()
   end
 
   def make_block_data(blocker, blocked, activity_id) do
@@ -695,11 +654,11 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   #### Report-related helpers
 
   def update_report_state(%Activity{} = activity, state) when state in @supported_report_states do
-    with new_data <- Map.put(activity.data, "state", state),
-         changeset <- Changeset.change(activity, data: new_data),
-         {:ok, activity} <- Repo.update(changeset) do
-      {:ok, activity}
-    end
+    new_data = Map.put(activity.data, "state", state)
+
+    activity
+    |> Changeset.change(data: new_data)
+    |> Repo.update()
   end
 
   def update_report_state(_, _), do: {:error, "Unsupported state"}
@@ -766,21 +725,13 @@ defmodule Pleroma.Web.ActivityPub.Utils do
   end
 
   def get_existing_votes(actor, %{data: %{"id" => id}}) do
-    query =
-      from(
-        [activity, object: object] in Activity.with_preloaded_object(Activity),
-        where: fragment("(?)->>'type' = 'Create'", activity.data),
-        where: fragment("(?)->>'actor' = ?", activity.data, ^actor),
-        where:
-          fragment(
-            "(?)->>'inReplyTo' = ?",
-            object.data,
-            ^to_string(id)
-          ),
-        where: fragment("(?)->>'type' = 'Answer'", object.data)
-      )
-
-    Repo.all(query)
+    actor
+    |> Activity.Queries.by_actor()
+    |> Activity.Queries.by_type("Create")
+    |> Activity.with_preloaded_object()
+    |> where([a, object: o], fragment("(?)->>'inReplyTo' = ?", o.data, ^to_string(id)))
+    |> where([a, object: o], fragment("(?)->>'type' = 'Answer'", o.data))
+    |> Repo.all()
   end
 
   defp maybe_put(map, _key, nil), do: map
-- 
cgit v1.2.3


From 25d8216804c7742cd8549799a7785723f2a70afa Mon Sep 17 00:00:00 2001
From: Egor Kislitsyn <egor@kislitsyn.com>
Date: Fri, 13 Sep 2019 13:09:35 +0700
Subject: Add email change endpoint

---
 lib/pleroma/user.ex                                   |  9 +++++++++
 lib/pleroma/web/router.ex                             |  1 +
 .../web/twitter_api/controllers/util_controller.ex    | 19 +++++++++++++++++++
 3 files changed, 29 insertions(+)

(limited to 'lib')

diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 3aa245f2a..1f6a75d03 100644
--- a/lib/pleroma/user.ex
+++ b/lib/pleroma/user.ex
@@ -1624,4 +1624,13 @@ defmodule Pleroma.User do
   def is_internal_user?(%User{nickname: nil}), do: true
   def is_internal_user?(%User{local: true, nickname: "internal." <> _}), do: true
   def is_internal_user?(_), do: false
+
+  def change_email(user, email) do
+    user
+    |> cast(%{email: email}, [:email])
+    |> validate_required([:email])
+    |> unique_constraint(:email)
+    |> validate_format(:email, @email_regex)
+    |> update_and_set_cache()
+  end
 end
diff --git a/lib/pleroma/web/router.ex b/lib/pleroma/web/router.ex
index 7cd59acb2..b0464037e 100644
--- a/lib/pleroma/web/router.ex
+++ b/lib/pleroma/web/router.ex
@@ -224,6 +224,7 @@ defmodule Pleroma.Web.Router do
     scope [] do
       pipe_through(:oauth_write)
 
+      post("/change_email", UtilController, :change_email)
       post("/change_password", UtilController, :change_password)
       post("/delete_account", UtilController, :delete_account)
       put("/notification_settings", UtilController, :update_notificaton_settings)
diff --git a/lib/pleroma/web/twitter_api/controllers/util_controller.ex b/lib/pleroma/web/twitter_api/controllers/util_controller.ex
index 3405bd3b7..867787c57 100644
--- a/lib/pleroma/web/twitter_api/controllers/util_controller.ex
+++ b/lib/pleroma/web/twitter_api/controllers/util_controller.ex
@@ -314,6 +314,25 @@ defmodule Pleroma.Web.TwitterAPI.UtilController do
     end
   end
 
+  def change_email(%{assigns: %{user: user}} = conn, params) do
+    case CommonAPI.Utils.confirm_current_password(user, params["password"]) do
+      {:ok, user} ->
+        with {:ok, _user} <- User.change_email(user, params["email"]) do
+          json(conn, %{status: "success"})
+        else
+          {:error, changeset} ->
+            {_, {error, _}} = Enum.at(changeset.errors, 0)
+            json(conn, %{error: "Email #{error}."})
+
+          _ ->
+            json(conn, %{error: "Unable to change email."})
+        end
+
+      {:error, msg} ->
+        json(conn, %{error: msg})
+    end
+  end
+
   def delete_account(%{assigns: %{user: user}} = conn, params) do
     case CommonAPI.Utils.confirm_current_password(user, params["password"]) do
       {:ok, user} ->
-- 
cgit v1.2.3


From 69faec031d62f4e87a1791ae0c71ca4b0f02f12b Mon Sep 17 00:00:00 2001
From: Alex S <alex.strizhakov@gmail.com>
Date: Fri, 13 Sep 2019 19:02:42 +0300
Subject: markdown generation to the new file

---
 lib/pleroma/docs/markdown.ex | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

(limited to 'lib')

diff --git a/lib/pleroma/docs/markdown.ex b/lib/pleroma/docs/markdown.ex
index 24930cc9f..8386dc2fb 100644
--- a/lib/pleroma/docs/markdown.ex
+++ b/lib/pleroma/docs/markdown.ex
@@ -3,9 +3,9 @@ defmodule Pleroma.Docs.Markdown do
 
   @spec process(keyword()) :: {:ok, String.t()}
   def process(descriptions) do
-    config_path = "docs/config.md"
+    config_path = "docs/generated_config.md"
     {:ok, file} = File.open(config_path, [:utf8, :write])
-    IO.write(file, "# Configuration\n")
+    IO.write(file, "# Generated configuration\n")
     IO.write(file, "Date of generation: #{Date.utc_today()}\n\n")
 
     IO.write(
-- 
cgit v1.2.3


From 5c5ebd38619bb853a58374918fd8983569ba7c0b Mon Sep 17 00:00:00 2001
From: rinpatch <rinpatch@sdf.org>
Date: Sat, 14 Sep 2019 01:50:15 +0300
Subject: Mastodon API: Respect post privacy in favourited/reblogged endpoints

---
 lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex | 4 ++++
 1 file changed, 4 insertions(+)

(limited to 'lib')

diff --git a/lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex b/lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex
index 0940e07a6..060137b80 100644
--- a/lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex
+++ b/lib/pleroma/web/mastodon_api/controllers/mastodon_api_controller.ex
@@ -842,6 +842,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
 
   def favourited_by(%{assigns: %{user: user}} = conn, %{"id" => id}) do
     with %Activity{} = activity <- Activity.get_by_id_with_object(id),
+         {:visible, true} <- {:visible, Visibility.visible_for_user?(activity, user)},
          %Object{data: %{"likes" => likes}} <- Object.normalize(activity) do
       q = from(u in User, where: u.ap_id in ^likes)
 
@@ -853,12 +854,14 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       |> put_view(AccountView)
       |> render("accounts.json", %{for: user, users: users, as: :user})
     else
+      {:visible, false} -> {:error, :not_found}
       _ -> json(conn, [])
     end
   end
 
   def reblogged_by(%{assigns: %{user: user}} = conn, %{"id" => id}) do
     with %Activity{} = activity <- Activity.get_by_id_with_object(id),
+         {:visible, true} <- {:visible, Visibility.visible_for_user?(activity, user)},
          %Object{data: %{"announcements" => announces}} <- Object.normalize(activity) do
       q = from(u in User, where: u.ap_id in ^announces)
 
@@ -870,6 +873,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
       |> put_view(AccountView)
       |> render("accounts.json", %{for: user, users: users, as: :user})
     else
+      {:visible, false} -> {:error, :not_found}
       _ -> json(conn, [])
     end
   end
-- 
cgit v1.2.3