[#1149] Added more oban workers. Refactoring.

diff --git a/lib/pleroma/digest_email_worker.ex b/lib/pleroma/digest_email_worker.ex
index 3b0e2bca695d3671653fa7ed50537c33e26d1799..6e44cc955aa2b22d3008bdd54a28cd487e3f58d6 100644 (file)
--- a/lib/pleroma/digest_email_worker.ex
+++ b/lib/pleroma/digest_email_worker.ex
@@ -1,6 +1,11 @@
 defmodule Pleroma.DigestEmailWorker do
 defmodule Pleroma.DigestEmailWorker do

+  alias Pleroma.Repo
+  alias Pleroma.Workers.Mailer, as: MailerWorker
+

   import Ecto.Query
 
   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)
   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()
       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 """
   end
 
   @doc """

diff --git a/lib/pleroma/scheduled_activity_worker.ex b/lib/pleroma/scheduled_activity_worker.ex
index 65b38622f212e2399ddecf9dc1d83af593bdb296..cabea51ca98b7b67a88be061bc7b308dd3ad6575 100644 (file)
--- 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.Config

+  alias Pleroma.Repo

   alias Pleroma.ScheduledActivity
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
   alias Pleroma.ScheduledActivity
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI

+

   use GenServer
   require Logger
 
   @schedule_interval :timer.minutes(1)
 
   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
   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 ->
   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()
     end)
 
     schedule_next()

diff --git a/lib/pleroma/user.ex b/lib/pleroma/user.ex
index 7d18f099e8914b3964f480b49d0be9d99fe3a5b3..bc2102ca72bb43fdbd03110c8670d88a2a4245ed 100644 (file)
--- 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.Web.OStatus
   alias Pleroma.Web.RelMe
   alias Pleroma.Web.Websub

+  alias Pleroma.Workers.BackgroundWorker

 
   require Logger
 
 
   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_-]+$/
 
   @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)
   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"
   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
 
   @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
   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
   end
 
   def deactivate(%User{} = user, status \\ true) do


@@ -1029,9 +1037,11 @@ defmodule Pleroma.User do
     |> update_and_set_cache()
   end
 
     |> 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
 
   @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
 
     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
 
   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 1a279a7df1b7f00e06c6ef3120d8f77585221f2d..8be8ac86f470cd21ce74a05d269acc335341c5f7 100644 (file)
--- 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.Web.ActivityPub.MRF
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.WebFinger

+  alias Pleroma.Workers.BackgroundWorker

 
   import Ecto.Query
   import Pleroma.Web.ActivityPub.Utils
 
   import Ecto.Query
   import Pleroma.Web.ActivityPub.Utils


@@ -25,6 +26,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   require Logger
   require Pleroma.Constants
 
   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
   # 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
 
           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)
 
 
       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 01d21a299aa702b3272b86d3d65228bfd0299eb6..1df3bb5b629e02282518931dbdcde1c0728db57b 100644 (file)
--- 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
   @behaviour Pleroma.Web.ActivityPub.MRF
 
   alias Pleroma.HTTP

+  alias Pleroma.Repo

   alias Pleroma.Web.MediaProxy
   alias Pleroma.Web.MediaProxy

+  alias Pleroma.Workers.BackgroundWorker

 
   require Logger
 
 
   require Logger
 


@@ -16,6 +18,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
     recv_timeout: 10_000
   ]
 
     recv_timeout: 10_000
   ]
 

+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+

   def perform(:prefetch, url) do
     Logger.info("Prefetching #{inspect(url)}")
 
   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} ->
         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)}")
 
           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
         %{"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
 
     {:ok, message}
   end

diff --git a/lib/pleroma/web/activity_pub/transmogrifier.ex b/lib/pleroma/web/activity_pub/transmogrifier.ex
index 5403b71d831e3ccc6b8cef38f14e1f83827742e7..0f117cd041918d574e8c723958652e4814a14406 100644 (file)
--- 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.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
 
 
   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.
   """
   @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
          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}
       end
 
       {:ok, user}

diff --git a/lib/pleroma/web/federator/federator.ex b/lib/pleroma/web/federator/federator.ex
index bb9eadfee6b620c5e24821ae46e4c4f4d6835f18..d85fe824fd8f7cd46dca214083ef0baef6e4a2f6 100644 (file)
--- 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
 # 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
 
   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)
   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
   end
 
   def publish(%{id: "pleroma:fakeid"} = activity) do

-    PublisherWorker.perform_publish(activity)
+    perform(:publish, activity)

   end
 
   def publish(activity) do
   end
 
   def publish(activity) do


@@ -68,11 +79,88 @@ defmodule Pleroma.Web.Federator do
     |> Pleroma.Repo.insert()
   end
 
     |> 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
     else

-      []
+      ActivityPub.make_user_from_ap_id(id)

     end
   end
 end
     end
   end
 end

diff --git a/lib/pleroma/web/oauth/token/clean_worker.ex b/lib/pleroma/web/oauth/token/clean_worker.ex
index dca85244931d5cd2c664ebb41f3bd9c305b113f9..c0c9c3653827f8036155d79de331bb4b2840e932 100644 (file)
--- 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
             )
               [:oauth2, :clean_expired_tokens_interval],
               86_400_000
             )

-  @queue :background

 
 

+  alias Pleroma.Repo

   alias Pleroma.Web.OAuth.Token
   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)
 
 
   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
 
   @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)
     Process.send_after(self(), :perform, @interval)

-    PleromaJobQueue.enqueue(@queue, __MODULE__, [:clean])

     {:noreply, state}
   end
 
     {:noreply, state}
   end
 

diff --git a/lib/pleroma/web/push/push.ex b/lib/pleroma/web/push/push.ex
index 729dad02a22a0cec0698c6501cf2671e939ec49f..b4f0e5127e37518efb603fde9dd1a307017aed65 100644 (file)
--- 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
 # 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
 
 
   require Logger
 

+  defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+

   def init do
     unless enabled() do
       Logger.warn("""
   def init do
     unless enabled() do
       Logger.warn("""


@@ -31,6 +34,9 @@ defmodule Pleroma.Web.Push do
     end
   end
 
     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
 end

diff --git a/lib/pleroma/web/twitter_api/controllers/util_controller.ex b/lib/pleroma/web/twitter_api/controllers/util_controller.ex
index 3405bd3b7f5c2ab5d551183ec7fd32666d76f9a9..7ba4ad305f288be53c0da3ec311bb81bed3ab9ba 100644 (file)
--- 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
              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
       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
 
   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
       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 (file)
index 0000000..3ab2b6b
--- /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 (file)
index 0000000..3286ce0
--- /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 (file)
index 0000000..da7fa6f
--- /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 67871977a82254b4500fcf6bb274d7efd277eed1..c890ffb79b4c00a44ac32412dbb7f711fad4efc3 100644 (file)
--- a/lib/pleroma/workers/publisher.ex
+++ b/lib/pleroma/workers/publisher.ex
@@ -4,7 +4,7 @@
 
 defmodule Pleroma.Workers.Publisher do
   alias Pleroma.Activity
 
 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,
 
   # 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
 
   @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
   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
   end
 end

diff --git a/lib/pleroma/workers/receiver.ex b/lib/pleroma/workers/receiver.ex
index 43558b4e6b71d3d0b4068edca70dde3c1499784f..d3de9571680130c907f645746f1cc24aad269682 100644 (file)
--- 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
 # 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,
 
   # 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
 
   @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
   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
   end
 end

diff --git a/lib/pleroma/workers/scheduled_activity_worker.ex b/lib/pleroma/workers/scheduled_activity_worker.ex
new file mode 100644 (file)
index 0000000..a49834f
--- /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 a8c01bb10d1e755ed1e8c03aa9c35052ea36694c..6af3ad0a19e99a31aeefd8cc54c68bfc51706bbf 100644 (file)
--- a/lib/pleroma/workers/subscriber.ex
+++ b/lib/pleroma/workers/subscriber.ex
@@ -4,11 +4,9 @@
 
 defmodule Pleroma.Workers.Subscriber do
   alias Pleroma.Repo
 
 defmodule Pleroma.Workers.Subscriber do
   alias Pleroma.Repo

-  alias Pleroma.Web.Websub
+  alias Pleroma.Web.Federator

   alias Pleroma.Web.Websub.WebsubClientSubscription
 
   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",
   # 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
 
   @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)
   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)
   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
   end
 end

diff --git a/lib/pleroma/workers/transmogrifier.ex b/lib/pleroma/workers/transmogrifier.ex
new file mode 100644 (file)
index 0000000..c6b4fab
--- /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 (file)
index 0000000..b99581e
--- /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

diff --git a/test/activity_test.exs b/test/activity_test.exs
index b9c12adb21b50fccdbf6cf8c5175ff19ead48660..658c478375097b8f7c9a9d193b7976d73c427606 100644 (file)
--- a/test/activity_test.exs
+++ b/test/activity_test.exs
@@ -6,8 +6,8 @@ defmodule Pleroma.ActivityTest do
   use Pleroma.DataCase
   alias Pleroma.Activity
   alias Pleroma.Bookmark
   use Pleroma.DataCase
   alias Pleroma.Activity
   alias Pleroma.Bookmark

-  alias Pleroma.ObanHelpers

   alias Pleroma.Object
   alias Pleroma.Object

+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.ThreadMute
   import Pleroma.Factory
 
   alias Pleroma.ThreadMute
   import Pleroma.Factory
 

diff --git a/test/conversation_test.exs b/test/conversation_test.exs
index 2ebbcab761aaf1a5c6d868a2978284dc805b0d75..f917aa69105d7c2369738f63e8b298abc2a49f30 100644 (file)
--- a/test/conversation_test.exs
+++ b/test/conversation_test.exs
@@ -28,7 +28,7 @@ defmodule Pleroma.ConversationTest do
     {:ok, _activity} =
       CommonAPI.post(user, %{"visibility" => "direct", "status" => "hey @#{other_user.nickname}"})
 
     {:ok, _activity} =
       CommonAPI.post(user, %{"visibility" => "direct", "status" => "hey @#{other_user.nickname}"})
 

-    Pleroma.ObanHelpers.perform_all()
+    Pleroma.Tests.ObanHelpers.perform_all()

 
     Repo.delete_all(Conversation)
     Repo.delete_all(Conversation.Participation)
 
     Repo.delete_all(Conversation)
     Repo.delete_all(Conversation.Participation)

diff --git a/test/notification_test.exs b/test/notification_test.exs
index 80ea2a085b5f1153109d3a4e95358c49437debbd..e1c9f4f93ba151d4587be144a1c27a16f84810af 100644 (file)
--- a/test/notification_test.exs
+++ b/test/notification_test.exs
@@ -8,6 +8,7 @@ defmodule Pleroma.NotificationTest do
   import Pleroma.Factory
 
   alias Pleroma.Notification
   import Pleroma.Factory
 
   alias Pleroma.Notification

+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.CommonAPI
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.CommonAPI


@@ -621,7 +622,8 @@ defmodule Pleroma.NotificationTest do
 
       refute Enum.empty?(Notification.for_user(other_user))
 
 
       refute Enum.empty?(Notification.for_user(other_user))
 

-      User.delete(user)
+      {:ok, job} = User.delete(user)
+      ObanHelpers.perform(job)

 
       assert Enum.empty?(Notification.for_user(other_user))
     end
 
       assert Enum.empty?(Notification.for_user(other_user))
     end


@@ -666,6 +668,7 @@ defmodule Pleroma.NotificationTest do
       }
 
       {:ok, _delete_activity} = Transmogrifier.handle_incoming(delete_user_message)
       }
 
       {:ok, _delete_activity} = Transmogrifier.handle_incoming(delete_user_message)

+      ObanHelpers.perform_all()

 
       assert Enum.empty?(Notification.for_user(local_user))
     end
 
       assert Enum.empty?(Notification.for_user(local_user))
     end

diff --git a/test/support/oban_helpers.ex b/test/support/oban_helpers.ex
index ecc03ba1ab87f23d64ae2b99c69367a9039bdbe1..d379c9ec76272d9cc1ae0ad54f42c26fea28b682 100644 (file)
--- a/test/support/oban_helpers.ex
+++ b/test/support/oban_helpers.ex
@@ -2,7 +2,7 @@
 # Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
 # SPDX-License-Identifier: AGPL-3.0-only
 
 # Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
 # SPDX-License-Identifier: AGPL-3.0-only
 

-defmodule Pleroma.ObanHelpers do
+defmodule Pleroma.Tests.ObanHelpers do

   @moduledoc """
   Oban test helpers.
   """
   @moduledoc """
   Oban test helpers.
   """

diff --git a/test/user_test.exs b/test/user_test.exs
index 8617752d752dc4d855b7a149f3813764c739394d..9c2117a0b994fcff28543e91867e0d8a92dc65c0 100644 (file)
--- a/test/user_test.exs
+++ b/test/user_test.exs
@@ -5,9 +5,9 @@
 defmodule Pleroma.UserTest do
   alias Pleroma.Activity
   alias Pleroma.Builders.UserBuilder
 defmodule Pleroma.UserTest do
   alias Pleroma.Activity
   alias Pleroma.Builders.UserBuilder

-  alias Pleroma.ObanHelpers

   alias Pleroma.Object
   alias Pleroma.Repo
   alias Pleroma.Object
   alias Pleroma.Repo

+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.CommonAPI
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.CommonAPI


@@ -676,7 +676,9 @@ defmodule Pleroma.UserTest do
         user3.nickname
       ]
 
         user3.nickname
       ]
 

-      result = User.follow_import(user1, identifiers)
+      {:ok, job} = User.follow_import(user1, identifiers)
+      result = ObanHelpers.perform(job)
+

       assert is_list(result)
       assert result == [user2, user3]
     end
       assert is_list(result)
       assert result == [user2, user3]
     end


@@ -887,7 +889,9 @@ defmodule Pleroma.UserTest do
         user3.nickname
       ]
 
         user3.nickname
       ]
 

-      result = User.blocks_import(user1, identifiers)
+      {:ok, job} = User.blocks_import(user1, identifiers)
+      result = ObanHelpers.perform(job)
+

       assert is_list(result)
       assert result == [user2, user3]
     end
       assert is_list(result)
       assert result == [user2, user3]
     end


@@ -1013,7 +1017,8 @@ defmodule Pleroma.UserTest do
       {:ok, like_two, _} = CommonAPI.favorite(activity.id, follower)
       {:ok, repeat, _} = CommonAPI.repeat(activity_two.id, user)
 
       {:ok, like_two, _} = CommonAPI.favorite(activity.id, follower)
       {:ok, repeat, _} = CommonAPI.repeat(activity_two.id, user)
 

-      {:ok, _} = User.delete(user)
+      {:ok, job} = User.delete(user)
+      {:ok, _user} = ObanHelpers.perform(job)

 
       follower = User.get_cached_by_id(follower.id)
 
 
       follower = User.get_cached_by_id(follower.id)
 


@@ -1043,7 +1048,8 @@ defmodule Pleroma.UserTest do
       {:ok, follower} = User.get_or_fetch_by_ap_id("http://mastodon.example.org/users/admin")
       {:ok, _} = User.follow(follower, user)
 
       {:ok, follower} = User.get_or_fetch_by_ap_id("http://mastodon.example.org/users/admin")
       {:ok, _} = User.follow(follower, user)
 

-      {:ok, _user} = User.delete(user)
+      {:ok, job} = User.delete(user)
+      {:ok, _user} = ObanHelpers.perform(job)

 
       assert ObanHelpers.member?(
                %{
 
       assert ObanHelpers.member?(
                %{


@@ -1100,7 +1106,8 @@ defmodule Pleroma.UserTest do
     test "User.delete() plugs any possible zombie objects" do
       user = insert(:user)
 
     test "User.delete() plugs any possible zombie objects" do
       user = insert(:user)
 

-      {:ok, _} = User.delete(user)
+      {:ok, job} = User.delete(user)
+      {:ok, _} = ObanHelpers.perform(job)

 
       {:ok, cached_user} = Cachex.get(:user_cache, "ap_id:#{user.ap_id}")
 
 
       {:ok, cached_user} = Cachex.get(:user_cache, "ap_id:#{user.ap_id}")
 

diff --git a/test/web/activity_pub/activity_pub_controller_test.exs b/test/web/activity_pub/activity_pub_controller_test.exs
index d7f0a8264ab83514d0fe57042dd201e33db88260..f46353fddb52a66ece35708a694930feae161b9c 100644 (file)
--- a/test/web/activity_pub/activity_pub_controller_test.exs
+++ b/test/web/activity_pub/activity_pub_controller_test.exs
@@ -9,8 +9,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
   import Pleroma.Factory
   alias Pleroma.Activity
   alias Pleroma.Instances
   import Pleroma.Factory
   alias Pleroma.Activity
   alias Pleroma.Instances

-  alias Pleroma.ObanHelpers

   alias Pleroma.Object
   alias Pleroma.Object

+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ObjectView
   alias Pleroma.Web.ActivityPub.UserView
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ObjectView
   alias Pleroma.Web.ActivityPub.UserView

diff --git a/test/web/activity_pub/mrf/mediaproxy_warming_policy_test.exs b/test/web/activity_pub/mrf/mediaproxy_warming_policy_test.exs
index 372e789be34e585da6d68e654e36f22bb958e883..95a809d25a1a5551bad1de39da6d46dc7208c678 100644 (file)
--- a/test/web/activity_pub/mrf/mediaproxy_warming_policy_test.exs
+++ b/test/web/activity_pub/mrf/mediaproxy_warming_policy_test.exs
@@ -6,6 +6,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicyTest do
   use Pleroma.DataCase
 
   alias Pleroma.HTTP
   use Pleroma.DataCase
 
   alias Pleroma.HTTP

+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy
 
   import Mock
   alias Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy
 
   import Mock


@@ -24,6 +25,11 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicyTest do
   test "it prefetches media proxy URIs" do
     with_mock HTTP, get: fn _, _, _ -> {:ok, []} end do
       MediaProxyWarmingPolicy.filter(@message)
   test "it prefetches media proxy URIs" do
     with_mock HTTP, get: fn _, _, _ -> {:ok, []} end do
       MediaProxyWarmingPolicy.filter(@message)

+
+      ObanHelpers.perform_all()
+      # Performing jobs which has been just enqueued
+      ObanHelpers.perform_all()
+

       assert called(HTTP.get(:_, :_, :_))
     end
   end
       assert called(HTTP.get(:_, :_, :_))
     end
   end

diff --git a/test/web/activity_pub/transmogrifier_test.exs b/test/web/activity_pub/transmogrifier_test.exs
index e7498e00570cf59d6541b8e1d01b8027671213f8..52f46c14167c640b0a3799fa551ef2567dbeaa77 100644 (file)
--- a/test/web/activity_pub/transmogrifier_test.exs
+++ b/test/web/activity_pub/transmogrifier_test.exs
@@ -8,6 +8,7 @@ defmodule Pleroma.Web.ActivityPub.TransmogrifierTest do
   alias Pleroma.Object
   alias Pleroma.Object.Fetcher
   alias Pleroma.Repo
   alias Pleroma.Object
   alias Pleroma.Object.Fetcher
   alias Pleroma.Repo

+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Transmogrifier


@@ -563,6 +564,7 @@ defmodule Pleroma.Web.ActivityPub.TransmogrifierTest do
         |> Poison.decode!()
 
       {:ok, _} = Transmogrifier.handle_incoming(data)
         |> Poison.decode!()
 
       {:ok, _} = Transmogrifier.handle_incoming(data)

+      ObanHelpers.perform_all()

 
       refute User.get_cached_by_ap_id(ap_id)
     end
 
       refute User.get_cached_by_ap_id(ap_id)
     end


@@ -1132,6 +1134,8 @@ defmodule Pleroma.Web.ActivityPub.TransmogrifierTest do
       assert user.info.note_count == 1
 
       {:ok, user} = Transmogrifier.upgrade_user_from_ap_id("https://niu.moe/users/rye")
       assert user.info.note_count == 1
 
       {:ok, user} = Transmogrifier.upgrade_user_from_ap_id("https://niu.moe/users/rye")

+      ObanHelpers.perform_all()
+

       assert user.info.ap_enabled
       assert user.info.note_count == 1
       assert user.follower_address == "https://niu.moe/users/rye/followers"
       assert user.info.ap_enabled
       assert user.info.note_count == 1
       assert user.follower_address == "https://niu.moe/users/rye/followers"

diff --git a/test/web/federator_test.exs b/test/web/federator_test.exs
index e0be4342b90a8f0ef974c123b465f8d072382cd4..9ca341b6da988d710bde2325b54d1e641f999a42 100644 (file)
--- a/test/web/federator_test.exs
+++ b/test/web/federator_test.exs
@@ -4,7 +4,7 @@
 
 defmodule Pleroma.Web.FederatorTest do
   alias Pleroma.Instances
 
 defmodule Pleroma.Web.FederatorTest do
   alias Pleroma.Instances

-  alias Pleroma.ObanHelpers
+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.Federator
   alias Pleroma.Workers.Publisher, as: PublisherWorker
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.Federator
   alias Pleroma.Workers.Publisher, as: PublisherWorker

diff --git a/test/web/twitter_api/util_controller_test.exs b/test/web/twitter_api/util_controller_test.exs
index 640579c09fe49e4e962b838d77ece35afd52c1b3..e3f129f72ada47723fc9ca6feb8d493dadc0ef26 100644 (file)
--- a/test/web/twitter_api/util_controller_test.exs
+++ b/test/web/twitter_api/util_controller_test.exs
@@ -4,9 +4,11 @@
 
 defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
   use Pleroma.Web.ConnCase
 
 defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
   use Pleroma.Web.ConnCase

+  use Oban.Testing, repo: Pleroma.Repo

 
   alias Pleroma.Notification
   alias Pleroma.Repo
 
   alias Pleroma.Notification
   alias Pleroma.Repo

+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
   import Pleroma.Factory
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
   import Pleroma.Factory


@@ -50,8 +52,7 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
         {File, [],
          read!: fn "follow_list.txt" ->
            "Account address,Show boosts\n#{user2.ap_id},true"
         {File, [],
          read!: fn "follow_list.txt" ->
            "Account address,Show boosts\n#{user2.ap_id},true"

-         end},
-        {PleromaJobQueue, [:passthrough], []}
+         end}

       ]) do
         response =
           conn
       ]) do
         response =
           conn


@@ -59,15 +60,16 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
           |> post("/api/pleroma/follow_import", %{"list" => %Plug.Upload{path: "follow_list.txt"}})
           |> json_response(:ok)
 
           |> post("/api/pleroma/follow_import", %{"list" => %Plug.Upload{path: "follow_list.txt"}})
           |> json_response(:ok)
 

-        assert called(
-                 PleromaJobQueue.enqueue(
-                   :background,
-                   User,
-                   [:follow_import, user1, [user2.ap_id]]
-                 )
-               )
-

         assert response == "job started"
         assert response == "job started"

+
+        assert ObanHelpers.member?(
+                 %{
+                   "op" => "follow_import",
+                   "follower_id" => user1.id,
+                   "followed_identifiers" => [user2.ap_id]
+                 },
+                 all_enqueued(worker: Pleroma.Workers.BackgroundWorker)
+               )

       end
     end
 
       end
     end
 


@@ -126,8 +128,7 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
       user3 = insert(:user)
 
       with_mocks([
       user3 = insert(:user)
 
       with_mocks([

-        {File, [], read!: fn "blocks_list.txt" -> "#{user2.ap_id} #{user3.ap_id}" end},
-        {PleromaJobQueue, [:passthrough], []}
+        {File, [], read!: fn "blocks_list.txt" -> "#{user2.ap_id} #{user3.ap_id}" end}

       ]) do
         response =
           conn
       ]) do
         response =
           conn


@@ -135,15 +136,16 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
           |> post("/api/pleroma/blocks_import", %{"list" => %Plug.Upload{path: "blocks_list.txt"}})
           |> json_response(:ok)
 
           |> post("/api/pleroma/blocks_import", %{"list" => %Plug.Upload{path: "blocks_list.txt"}})
           |> json_response(:ok)
 

-        assert called(
-                 PleromaJobQueue.enqueue(
-                   :background,
-                   User,
-                   [:blocks_import, user1, [user2.ap_id, user3.ap_id]]
-                 )
-               )
-

         assert response == "job started"
         assert response == "job started"

+
+        assert ObanHelpers.member?(
+                 %{
+                   "op" => "blocks_import",
+                   "blocker_id" => user1.id,
+                   "blocked_identifiers" => [user2.ap_id, user3.ap_id]
+                 },
+                 all_enqueued(worker: Pleroma.Workers.BackgroundWorker)
+               )

       end
     end
   end
       end
     end
   end


@@ -607,6 +609,7 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
         |> json_response(:ok)
 
       assert response == %{"status" => "success"}
         |> json_response(:ok)
 
       assert response == %{"status" => "success"}

+      ObanHelpers.perform_all()

 
       user = User.get_cached_by_id(user.id)
 
 
       user = User.get_cached_by_id(user.id)
 

diff --git a/test/web/websub/websub_test.exs b/test/web/websub/websub_test.exs
index b704a558a8c016960d288aeb288369759f2eef93..414610879629263e41d3320fc010b10896fa8bac 100644 (file)
--- a/test/web/websub/websub_test.exs
+++ b/test/web/websub/websub_test.exs
@@ -6,7 +6,7 @@ defmodule Pleroma.Web.WebsubTest do
   use Pleroma.DataCase
   use Oban.Testing, repo: Pleroma.Repo
 
   use Pleroma.DataCase
   use Oban.Testing, repo: Pleroma.Repo
 

-  alias Pleroma.ObanHelpers
+  alias Pleroma.Tests.ObanHelpers

   alias Pleroma.Web.Router.Helpers
   alias Pleroma.Web.Websub
   alias Pleroma.Web.Websub.WebsubClientSubscription
   alias Pleroma.Web.Router.Helpers
   alias Pleroma.Web.Websub
   alias Pleroma.Web.Websub.WebsubClientSubscription