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)
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 """
"""
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
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()
alias Pleroma.Web.OStatus
alias Pleroma.Web.RelMe
alias Pleroma.Web.Websub
+ alias Pleroma.Workers.BackgroundWorker
require Logger
@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)
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
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
|> 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
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
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
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
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)
@behaviour Pleroma.Web.ActivityPub.MRF
alias Pleroma.HTTP
+ alias Pleroma.Repo
alias Pleroma.Web.MediaProxy
+ alias Pleroma.Workers.BackgroundWorker
require Logger
recv_timeout: 10_000
]
+ defdelegate worker_args(queue), to: Pleroma.Workers.Helper
+
def perform(:prefetch, url) do
Logger.info("Prefetching #{inspect(url)}")
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)}")
%{"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
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.
"""
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}
# 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)
end
def publish(%{id: "pleroma:fakeid"} = activity) do
- PublisherWorker.perform_publish(activity)
+ perform(:publish, activity)
end
def publish(activity) 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
[: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)
@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
# 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("""
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
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
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
--- /dev/null
+# 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
--- /dev/null
+# 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
--- /dev/null
+# 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
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,
@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
# 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,
@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
--- /dev/null
+# 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
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",
@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
--- /dev/null
+# 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
--- /dev/null
+# 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
use Pleroma.DataCase
alias Pleroma.Activity
alias Pleroma.Bookmark
- alias Pleroma.ObanHelpers
alias Pleroma.Object
+ alias Pleroma.Tests.ObanHelpers
alias Pleroma.ThreadMute
import Pleroma.Factory
{: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)
import Pleroma.Factory
alias Pleroma.Notification
+ alias Pleroma.Tests.ObanHelpers
alias Pleroma.User
alias Pleroma.Web.ActivityPub.Transmogrifier
alias Pleroma.Web.CommonAPI
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
}
{:ok, _delete_activity} = Transmogrifier.handle_incoming(delete_user_message)
+ ObanHelpers.perform_all()
assert Enum.empty?(Notification.for_user(local_user))
end
# 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.
"""
defmodule Pleroma.UserTest do
alias Pleroma.Activity
alias Pleroma.Builders.UserBuilder
- alias Pleroma.ObanHelpers
alias Pleroma.Object
alias Pleroma.Repo
+ alias Pleroma.Tests.ObanHelpers
alias Pleroma.User
alias Pleroma.Web.ActivityPub.ActivityPub
alias Pleroma.Web.CommonAPI
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
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
{: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)
{: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?(
%{
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}")
import Pleroma.Factory
alias Pleroma.Activity
alias Pleroma.Instances
- alias Pleroma.ObanHelpers
alias Pleroma.Object
+ alias Pleroma.Tests.ObanHelpers
alias Pleroma.User
alias Pleroma.Web.ActivityPub.ObjectView
alias Pleroma.Web.ActivityPub.UserView
use Pleroma.DataCase
alias Pleroma.HTTP
+ alias Pleroma.Tests.ObanHelpers
alias Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy
import Mock
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
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
|> Poison.decode!()
{:ok, _} = Transmogrifier.handle_incoming(data)
+ ObanHelpers.perform_all()
refute User.get_cached_by_ap_id(ap_id)
end
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"
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
defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
use Pleroma.Web.ConnCase
+ use Oban.Testing, repo: Pleroma.Repo
alias Pleroma.Notification
alias Pleroma.Repo
+ alias Pleroma.Tests.ObanHelpers
alias Pleroma.User
alias Pleroma.Web.CommonAPI
import Pleroma.Factory
{File, [],
read!: fn "follow_list.txt" ->
"Account address,Show boosts\n#{user2.ap_id},true"
- end},
- {PleromaJobQueue, [:passthrough], []}
+ end}
]) do
response =
conn
|> 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 ObanHelpers.member?(
+ %{
+ "op" => "follow_import",
+ "follower_id" => user1.id,
+ "followed_identifiers" => [user2.ap_id]
+ },
+ all_enqueued(worker: Pleroma.Workers.BackgroundWorker)
+ )
end
end
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
|> 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 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
|> json_response(:ok)
assert response == %{"status" => "success"}
+ ObanHelpers.perform_all()
user = User.get_cached_by_id(user.id)
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
projects / akkoma / commitdiff