9 files changed, 160 insertions, 78 deletions
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