diff options
Diffstat (limited to 'lib/pleroma/workers')
20 files changed, 105 insertions, 29 deletions
diff --git a/lib/pleroma/workers/attachments_cleanup_worker.ex b/lib/pleroma/workers/attachments_cleanup_worker.ex index 4c17640..4c17640 100755..100644 --- a/lib/pleroma/workers/attachments_cleanup_worker.ex +++ b/lib/pleroma/workers/attachments_cleanup_worker.ex diff --git a/lib/pleroma/workers/background_worker.ex b/lib/pleroma/workers/background_worker.ex index 7944176..dbf40ee 100755..100644 --- a/lib/pleroma/workers/background_worker.ex +++ b/lib/pleroma/workers/background_worker.ex @@ -28,7 +28,7 @@ defmodule Pleroma.Workers.BackgroundWorker do def perform(%Job{args: %{"op" => op, "user_id" => user_id, "identifiers" => identifiers}}) when op in ["blocks_import", "follow_import", "mutes_import"] do user = User.get_cached_by_id(user_id) - {:ok, User.Import.perform(String.to_atom(op), user, identifiers)} + {:ok, User.Import.perform(String.to_existing_atom(op), user, identifiers)} end def perform(%Job{ @@ -40,6 +40,11 @@ defmodule Pleroma.Workers.BackgroundWorker do Pleroma.FollowingRelationship.move_following(origin, target) end + def perform(%Job{args: %{"op" => "verify_fields_links", "user_id" => user_id}}) do + user = User.get_by_id(user_id) + User.perform(:verify_fields_links, user) + end + def perform(%Job{args: %{"op" => "delete_instance", "host" => host}}) do Instance.perform(:delete_instance, host) end diff --git a/lib/pleroma/workers/backup_worker.ex b/lib/pleroma/workers/backup_worker.ex index 12ee70f..a485ddb 100755..100644 --- a/lib/pleroma/workers/backup_worker.ex +++ b/lib/pleroma/workers/backup_worker.ex @@ -51,7 +51,7 @@ defmodule Pleroma.Workers.BackupWorker do end @impl Oban.Worker - def timeout(_job), do: :timer.seconds(900) + def timeout(_job), do: :infinity defp has_email?(user) do not is_nil(user.email) and user.email != "" diff --git a/lib/pleroma/workers/cron/digest_emails_worker.ex b/lib/pleroma/workers/cron/digest_emails_worker.ex index 1540c16..0292bbb 100755..100644 --- a/lib/pleroma/workers/cron/digest_emails_worker.ex +++ b/lib/pleroma/workers/cron/digest_emails_worker.ex @@ -7,7 +7,7 @@ defmodule Pleroma.Workers.Cron.DigestEmailsWorker do The worker to send digest emails. """ - use Oban.Worker, queue: "digest_emails" + use Oban.Worker, queue: "mailer" alias Pleroma.Config alias Pleroma.Emails diff --git a/lib/pleroma/workers/cron/new_users_digest_worker.ex b/lib/pleroma/workers/cron/new_users_digest_worker.ex index 267fe28..1c3e445 100755..100644 --- a/lib/pleroma/workers/cron/new_users_digest_worker.ex +++ b/lib/pleroma/workers/cron/new_users_digest_worker.ex @@ -9,7 +9,7 @@ defmodule Pleroma.Workers.Cron.NewUsersDigestWorker do import Ecto.Query - use Pleroma.Workers.WorkerHelper, queue: "new_users_digest" + use Pleroma.Workers.WorkerHelper, queue: "mailer" @impl Oban.Worker def perform(_job) do diff --git a/lib/pleroma/workers/mailer_worker.ex b/lib/pleroma/workers/mailer_worker.ex index 9407165..9407165 100755..100644 --- a/lib/pleroma/workers/mailer_worker.ex +++ b/lib/pleroma/workers/mailer_worker.ex diff --git a/lib/pleroma/workers/mute_expire_worker.ex b/lib/pleroma/workers/mute_expire_worker.ex index 8ce458d..8ce458d 100755..100644 --- a/lib/pleroma/workers/mute_expire_worker.ex +++ b/lib/pleroma/workers/mute_expire_worker.ex diff --git a/lib/pleroma/workers/poll_worker.ex b/lib/pleroma/workers/poll_worker.ex index 022d026..022d026 100755..100644 --- a/lib/pleroma/workers/poll_worker.ex +++ b/lib/pleroma/workers/poll_worker.ex diff --git a/lib/pleroma/workers/publisher_worker.ex b/lib/pleroma/workers/publisher_worker.ex index 598ae37..63fcf4a 100755..100644 --- a/lib/pleroma/workers/publisher_worker.ex +++ b/lib/pleroma/workers/publisher_worker.ex @@ -18,9 +18,9 @@ defmodule Pleroma.Workers.PublisherWorker do Federator.perform(:publish, activity) end - def perform(%Job{args: %{"op" => "publish_one", "module" => module_name, "params" => params}}) do + def perform(%Job{args: %{"op" => "publish_one", "params" => params}}) do params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end) - Federator.perform(:publish_one, String.to_atom(module_name), params) + Federator.perform(:publish_one, params) end @impl Oban.Worker diff --git a/lib/pleroma/workers/purge_expired_activity.ex b/lib/pleroma/workers/purge_expired_activity.ex index e554684..e554684 100755..100644 --- a/lib/pleroma/workers/purge_expired_activity.ex +++ b/lib/pleroma/workers/purge_expired_activity.ex diff --git a/lib/pleroma/workers/purge_expired_filter.ex b/lib/pleroma/workers/purge_expired_filter.ex index 9114aeb..9114aeb 100755..100644 --- a/lib/pleroma/workers/purge_expired_filter.ex +++ b/lib/pleroma/workers/purge_expired_filter.ex diff --git a/lib/pleroma/workers/purge_expired_token.ex b/lib/pleroma/workers/purge_expired_token.ex index 2ccd9e8..2ccd9e8 100755..100644 --- a/lib/pleroma/workers/purge_expired_token.ex +++ b/lib/pleroma/workers/purge_expired_token.ex diff --git a/lib/pleroma/workers/receiver_worker.ex b/lib/pleroma/workers/receiver_worker.ex index cf1bb62..8b2052c 100755..100644 --- a/lib/pleroma/workers/receiver_worker.ex +++ b/lib/pleroma/workers/receiver_worker.ex @@ -3,24 +3,57 @@ # SPDX-License-Identifier: AGPL-3.0-only defmodule Pleroma.Workers.ReceiverWorker do + alias Pleroma.Signature + alias Pleroma.User alias Pleroma.Web.Federator use Pleroma.Workers.WorkerHelper, queue: "federator_incoming" @impl Oban.Worker + + def perform(%Job{ + args: %{"op" => "incoming_ap_doc", "req_headers" => req_headers, "params" => params} + }) do + # Oban's serialization converts our tuple headers to lists. + # Revert it for the signature validation. + req_headers = Enum.into(req_headers, [], &List.to_tuple(&1)) + + conn_data = %{params: params, req_headers: req_headers} + + with {:ok, %User{} = _actor} <- User.get_or_fetch_by_ap_id(conn_data.params["actor"]), + {:ok, _public_key} <- Signature.refetch_public_key(conn_data), + {:signature, true} <- {:signature, HTTPSignatures.validate_conn(conn_data)}, + {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do + {:ok, res} + else + e -> process_errors(e) + end + end + def perform(%Job{args: %{"op" => "incoming_ap_doc", "params" => params}}) do with {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do {:ok, res} else + e -> process_errors(e) + end + end + + @impl Oban.Worker + def timeout(%_{args: %{"timeout" => timeout}}), do: timeout + + def timeout(_job), do: :timer.seconds(5) + + defp process_errors(errors) do + case errors do {:error, :origin_containment_failed} -> {:cancel, :origin_containment_failed} {:error, :already_present} -> {:cancel, :already_present} {:error, {:validate_object, reason}} -> {:cancel, reason} {:error, {:error, {:validate, reason}}} -> {:cancel, reason} {:error, {:reject, reason}} -> {:cancel, reason} - e -> e + {:signature, false} -> {:cancel, :invalid_signature} + {:error, {:error, reason = "Object has been deleted"}} -> {:cancel, reason} + {:error, _} = e -> e + e -> {:error, e} end end - - @impl Oban.Worker - def timeout(_job), do: :timer.seconds(5) end diff --git a/lib/pleroma/workers/remote_fetcher_worker.ex b/lib/pleroma/workers/remote_fetcher_worker.ex index d2a77aa..c264184 100755..100644 --- a/lib/pleroma/workers/remote_fetcher_worker.ex +++ b/lib/pleroma/workers/remote_fetcher_worker.ex @@ -9,7 +9,25 @@ defmodule Pleroma.Workers.RemoteFetcherWorker do @impl Oban.Worker def perform(%Job{args: %{"op" => "fetch_remote", "id" => id} = args}) do - {:ok, _object} = Fetcher.fetch_object_from_id(id, depth: args["depth"]) + case Fetcher.fetch_object_from_id(id, depth: args["depth"]) do + {:ok, _object} -> + :ok + + {:error, :forbidden} -> + {:discard, :forbidden} + + {:error, :not_found} -> + {:discard, :not_found} + + {:error, :allowed_depth} -> + {:discard, :allowed_depth} + + {:error, _} = e -> + e + + e -> + {:error, e} + end end @impl Oban.Worker diff --git a/lib/pleroma/workers/rich_media_expiration_worker.ex b/lib/pleroma/workers/rich_media_expiration_worker.ex new file mode 100644 index 0000000..d7ae497 --- /dev/null +++ b/lib/pleroma/workers/rich_media_expiration_worker.ex @@ -0,0 +1,15 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/> +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Workers.RichMediaExpirationWorker do + alias Pleroma.Web.RichMedia.Card + + use Oban.Worker, + queue: :rich_media_expiration + + @impl Oban.Worker + def perform(%Job{args: %{"url" => url} = _args}) do + Card.delete(url) + end +end diff --git a/lib/pleroma/workers/scheduled_activity_worker.ex b/lib/pleroma/workers/scheduled_activity_worker.ex index 4df84d0..4df84d0 100755..100644 --- a/lib/pleroma/workers/scheduled_activity_worker.ex +++ b/lib/pleroma/workers/scheduled_activity_worker.ex diff --git a/lib/pleroma/workers/search_indexing_worker.ex b/lib/pleroma/workers/search_indexing_worker.ex new file mode 100644 index 0000000..8476a2b --- /dev/null +++ b/lib/pleroma/workers/search_indexing_worker.ex @@ -0,0 +1,23 @@ +defmodule Pleroma.Workers.SearchIndexingWorker do + use Pleroma.Workers.WorkerHelper, queue: "search_indexing" + + @impl Oban.Worker + + alias Pleroma.Config.Getting, as: Config + + def perform(%Job{args: %{"op" => "add_to_index", "activity" => activity_id}}) do + activity = Pleroma.Activity.get_by_id_with_object(activity_id) + + search_module = Config.get([Pleroma.Search, :module]) + + search_module.add_to_index(activity) + end + + def perform(%Job{args: %{"op" => "remove_from_index", "object" => object_id}}) do + object = Pleroma.Object.get_by_id(object_id) + + search_module = Config.get([Pleroma.Search, :module]) + + search_module.remove_from_index(object) + end +end diff --git a/lib/pleroma/workers/transmogrifier_worker.ex b/lib/pleroma/workers/transmogrifier_worker.ex deleted file mode 100755 index 1f3f538..0000000 --- a/lib/pleroma/workers/transmogrifier_worker.ex +++ /dev/null @@ -1,18 +0,0 @@ -# Pleroma: A lightweight social networking server -# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/> -# SPDX-License-Identifier: AGPL-3.0-only - -defmodule Pleroma.Workers.TransmogrifierWorker do - alias Pleroma.User - - use Pleroma.Workers.WorkerHelper, queue: "transmogrifier" - - @impl Oban.Worker - def perform(%Job{args: %{"op" => "user_upgrade", "user_id" => user_id}}) do - user = User.get_cached_by_id(user_id) - Pleroma.Web.ActivityPub.Transmogrifier.perform(:user_upgrade, user) - end - - @impl Oban.Worker - def timeout(_job), do: :timer.seconds(5) -end diff --git a/lib/pleroma/workers/web_pusher_worker.ex b/lib/pleroma/workers/web_pusher_worker.ex index 67e84b0..67e84b0 100755..100644 --- a/lib/pleroma/workers/web_pusher_worker.ex +++ b/lib/pleroma/workers/web_pusher_worker.ex diff --git a/lib/pleroma/workers/worker_helper.ex b/lib/pleroma/workers/worker_helper.ex index 1d20cbd..1d20cbd 100755..100644 --- a/lib/pleroma/workers/worker_helper.ex +++ b/lib/pleroma/workers/worker_helper.ex |