Merge remote-tracking branch 'origin/develop' into translate-posts

lib/mix/pleroma.ex

@@ -14,7 +14,8 @@ defmodule Mix.Pleroma do
     :swoosh,
     :timex,
     :fast_html,
-    :oban
+    :oban,
+    :logger_backends
   ]
   @cachex_children ["object", "user", "scrubber", "web_resp"]
   @doc "Common functions to be reused in mix tasks"

lib/mix/tasks/pleroma/config.ex

@@ -205,6 +205,35 @@ defmodule Mix.Tasks.Pleroma.Config do
     end
   end
 
+  # Removes any policies that are not a real module
+  # as they will prevent the server from starting
+  def run(["fix_mrf_policies"]) do
+    check_configdb(fn ->
+      start_pleroma()
+
+      group = :pleroma
+      key = :mrf
+
+      %{value: value} =
+        group
+        |> ConfigDB.get_by_group_and_key(key)
+
+      policies =
+        Keyword.get(value, :policies, [])
+        |> Enum.filter(&is_atom(&1))
+        |> Enum.filter(fn mrf ->
+          case Code.ensure_compiled(mrf) do
+            {:module, _} -> true
+            {:error, _} -> false
+          end
+        end)
+
+      value = Keyword.put(value, :policies, policies)
+
+      ConfigDB.update_or_create(%{group: group, key: key, value: value})
+    end)
+  end
+
   @spec migrate_to_db(Path.t() | nil) :: any()
   def migrate_to_db(file_path \\ nil) do
     with :ok <- Pleroma.Config.DeprecationWarnings.warn() do

lib/mix/tasks/pleroma/database.ex

@@ -351,7 +351,7 @@ defmodule Mix.Tasks.Pleroma.Database do
         )
       end
 
-      shell_info('Done.')
+      shell_info(~c"Done.")
     end
   end
 

lib/mix/tasks/pleroma/test_runner.ex

@@ -0,0 +1,25 @@
+defmodule Mix.Tasks.Pleroma.TestRunner do
+  @shortdoc "Retries tests once if they fail"
+
+  use Mix.Task
+
+  def run(args \\ []) do
+    case System.cmd("mix", ["test"] ++ args, into: IO.stream(:stdio, :line)) do
+      {_, 0} ->
+        :ok
+
+      _ ->
+        retry(args)
+    end
+  end
+
+  def retry(args) do
+    case System.cmd("mix", ["test", "--failed"] ++ args, into: IO.stream(:stdio, :line)) do
+      {_, 0} ->
+        :ok
+
+      _ ->
+        exit(1)
+    end
+  end
+end

lib/pleroma/application.ex

@@ -14,7 +14,6 @@ defmodule Pleroma.Application do
   @name Mix.Project.config()[:name]
   @version Mix.Project.config()[:version]
   @repository Mix.Project.config()[:source_url]
-  @compile_env Mix.env()
 
   def name, do: @name
   def version, do: @version
@@ -53,7 +52,7 @@ defmodule Pleroma.Application do
     Pleroma.Config.Oban.warn()
     Config.DeprecationWarnings.warn()
 
-    if @compile_env != :test do
+    if Config.get([Pleroma.Web.Plugs.HTTPSecurityPlug, :enable], true) do
       Pleroma.Web.Plugs.HTTPSecurityPlug.warn_if_disabled()
     end
 

lib/pleroma/application_requirements.ex

@@ -241,10 +241,9 @@ defmodule Pleroma.ApplicationRequirements do
 
     missing_mrfs =
       Enum.reduce(mrfs, [], fn x, acc ->
-        if Code.ensure_compiled(x) do
-          acc
-        else
-          acc ++ [x]
+        case Code.ensure_compiled(x) do
+          {:module, _} -> acc
+          {:error, _} -> acc ++ [x]
         end
       end)
 

lib/pleroma/config/transfer_task.ex

@@ -1,5 +1,5 @@
 # Pleroma: A lightweight social networking server
-# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# Copyright © 2017-2023 Pleroma Authors <https://pleroma.social/>
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Config.TransferTask do
@@ -44,14 +44,9 @@ defmodule Pleroma.Config.TransferTask do
     with {_, true} <- {:configurable, Config.get(:configurable_from_database)} do
       # We need to restart applications for loaded settings take effect
 
-      {logger, other} =
+      settings =
         (Repo.all(ConfigDB) ++ deleted_settings)
         |> Enum.map(&merge_with_default/1)
-        |> Enum.split_with(fn {group, _, _, _} -> group in [:logger] end)
-
-      logger
-      |> Enum.sort()
-      |> Enum.each(&configure/1)
 
       started_applications = Application.started_applications()
 
@@ -64,7 +59,7 @@ defmodule Pleroma.Config.TransferTask do
           [:pleroma | reject]
         end
 
-      other
+      settings
       |> Enum.map(&update/1)
       |> Enum.uniq()
       |> Enum.reject(&(&1 in reject))
@@ -102,38 +97,6 @@ defmodule Pleroma.Config.TransferTask do
     {group, key, value, merged}
   end
 
-  # change logger configuration in runtime, without restart
-  defp configure({_, :backends, _, merged}) do
-    # removing current backends
-    Enum.each(Application.get_env(:logger, :backends), &Logger.remove_backend/1)
-
-    Enum.each(merged, &Logger.add_backend/1)
-
-    :ok = update_env(:logger, :backends, merged)
-  end
-
-  defp configure({_, key, _, merged}) when key in [:console, :ex_syslogger] do
-    merged =
-      if key == :console do
-        put_in(merged[:format], merged[:format] <> "\n")
-      else
-        merged
-      end
-
-    backend =
-      if key == :ex_syslogger,
-        do: {ExSyslogger, :ex_syslogger},
-        else: key
-
-    Logger.configure_backend(backend, merged)
-    :ok = update_env(:logger, key, merged)
-  end
-
-  defp configure({_, key, _, merged}) do
-    Logger.configure([{key, merged}])
-    :ok = update_env(:logger, key, merged)
-  end
-
   defp update({group, key, value, merged}) do
     try do
       :ok = update_env(group, key, merged)

lib/pleroma/config_db.ex

@@ -165,8 +165,7 @@ defmodule Pleroma.ConfigDB do
       {:pleroma, :ecto_repos},
       {:mime, :types},
       {:cors_plug, [:max_age, :methods, :expose, :headers]},
-      {:swarm, :node_blacklist},
-      {:logger, :backends}
+      {:swarm, :node_blacklist}
     ]
 
     Enum.any?(full_key_update, fn
@@ -385,7 +384,12 @@ defmodule Pleroma.ConfigDB do
 
   @spec module_name?(String.t()) :: boolean()
   def module_name?(string) do
-    Regex.match?(~r/^(Pleroma|Phoenix|Tesla|Ueberauth|Swoosh)\./, string) or
-      string in ["Oban", "Ueberauth", "ExSyslogger", "ConcurrentLimiter"]
+    if String.contains?(string, ".") do
+      [name | _] = String.split(string, ".", parts: 2)
+
+      name in ~w[Pleroma Phoenix Tesla Ueberauth Swoosh Logger LoggerBackends]
+    else
+      string in ~w[Oban Ueberauth ExSyslogger ConcurrentLimiter]
+    end
   end
 end

lib/pleroma/conversation/participation.ex

@@ -12,6 +12,8 @@ defmodule Pleroma.Conversation.Participation do
   import Ecto.Changeset
   import Ecto.Query
 
+  @type t :: %__MODULE__{}
+
   schema "conversation_participations" do
     belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
     belongs_to(:conversation, Conversation)

lib/pleroma/ecto_enums.ex

@@ -27,11 +27,3 @@ defenum(Pleroma.DataMigration.State,
   failed: 4,
   manual: 5
 )
-
-defenum(Pleroma.User.Backup.State,
-  pending: 1,
-  running: 2,
-  complete: 3,
-  failed: 4,
-  invalid: 5
-)

lib/pleroma/emails/user_email.ex

@@ -345,37 +345,22 @@ defmodule Pleroma.Emails.UserEmail do
     Router.Helpers.subscription_url(Endpoint, :unsubscribe, token)
   end
 
-  def backup_is_ready_email(backup, admin_user_id \\ nil) do
+  def backup_is_ready_email(backup) do
     %{user: user} = Pleroma.Repo.preload(backup, :user)
 
     Gettext.with_locale_or_default user.language do
       download_url = Pleroma.Web.PleromaAPI.BackupView.download_url(backup)
 
       html_body =
-        if is_nil(admin_user_id) do
-          Gettext.dpgettext(
-            "static_pages",
-            "account archive email body - self-requested",
-            """
-            <p>You requested a full backup of your Pleroma account. It's ready for download:</p>
-            <p><a href="%{download_url}">%{download_url}</a></p>
-            """,
-            download_url: download_url
-          )
-        else
-          admin = Pleroma.Repo.get(User, admin_user_id)
-
-          Gettext.dpgettext(
-            "static_pages",
-            "account archive email body - admin requested",
-            """
-            <p>Admin @%{admin_nickname} requested a full backup of your Pleroma account. It's ready for download:</p>
-            <p><a href="%{download_url}">%{download_url}</a></p>
-            """,
-            admin_nickname: admin.nickname,
-            download_url: download_url
-          )
-        end
+        Gettext.dpgettext(
+          "static_pages",
+          "account archive email body",
+          """
+          <p>A full backup of your Pleroma account was requested. It's ready for download:</p>
+          <p><a href="%{download_url}">%{download_url}</a></p>
+          """,
+          download_url: download_url
+        )
 
       new()
       |> to(recipient(user))

lib/pleroma/emoji/pack.ex

@@ -416,10 +416,10 @@ defmodule Pleroma.Emoji.Pack do
   end
 
   defp create_archive_and_cache(pack, hash) do
-    files = ['pack.json' | Enum.map(pack.files, fn {_, file} -> to_charlist(file) end)]
+    files = [~c"pack.json" | Enum.map(pack.files, fn {_, file} -> to_charlist(file) end)]
 
     {:ok, {_, result}} =
-      :zip.zip('#{pack.name}.zip', files, [:memory, cwd: to_charlist(pack.path)])
+      :zip.zip(~c"#{pack.name}.zip", files, [:memory, cwd: to_charlist(pack.path)])
 
     ttl_per_file = Pleroma.Config.get!([:emoji, :shared_pack_cache_seconds_per_file])
     overall_ttl = :timer.seconds(ttl_per_file * Enum.count(files))
@@ -586,7 +586,7 @@ defmodule Pleroma.Emoji.Pack do
     with :ok <- File.mkdir_p!(local_pack.path) do
       files = Enum.map(remote_pack["files"], fn {_, path} -> to_charlist(path) end)
       # Fallback cannot contain a pack.json file
-      files = if pack_info[:fallback], do: files, else: ['pack.json' | files]
+      files = if pack_info[:fallback], do: files, else: [~c"pack.json" | files]
 
       :zip.unzip(archive, cwd: to_charlist(local_pack.path), file_list: files)
     end

lib/pleroma/following_relationship.ex

@@ -199,8 +199,8 @@ defmodule Pleroma.FollowingRelationship do
     |> preload([:follower])
     |> Repo.all()
     |> Enum.map(fn following_relationship ->
-      Pleroma.Web.CommonAPI.follow(following_relationship.follower, target)
-      Pleroma.Web.CommonAPI.unfollow(following_relationship.follower, origin)
+      Pleroma.Web.CommonAPI.follow(target, following_relationship.follower)
+      Pleroma.Web.CommonAPI.unfollow(origin, following_relationship.follower)
     end)
     |> case do
       [] ->

lib/pleroma/frontend.ex

@@ -43,10 +43,6 @@ defmodule Pleroma.Frontend do
       {:download_or_unzip, _} ->
         Logger.info("Could not download or unzip the frontend")
         {:error, "Could not download or unzip the frontend"}
-
-      _e ->
-        Logger.info("Could not install the frontend")
-        {:error, "Could not install the frontend"}
     end
   end
 

lib/pleroma/gun/connection_pool/reclaimer.ex

@@ -9,7 +9,7 @@ defmodule Pleroma.Gun.ConnectionPool.Reclaimer do
 
   def start_monitor do
     pid =
-      case GenServer.start_link(__MODULE__, [], name: {:via, Registry, {registry(), "reclaimer"}}) do
+      case GenServer.start(__MODULE__, [], name: {:via, Registry, {registry(), "reclaimer"}}) do
         {:ok, pid} ->
           pid
 

lib/pleroma/gun/connection_pool/worker_supervisor.ex

@@ -5,6 +5,9 @@
 defmodule Pleroma.Gun.ConnectionPool.WorkerSupervisor do
   @moduledoc "Supervisor for pool workers. Does not do anything except enforce max connection limit"
 
+  alias Pleroma.Config
+  alias Pleroma.Gun.ConnectionPool.Worker
+
   use DynamicSupervisor
 
   def start_link(opts) do
@@ -14,21 +17,28 @@ defmodule Pleroma.Gun.ConnectionPool.WorkerSupervisor do
   def init(_opts) do
     DynamicSupervisor.init(
       strategy: :one_for_one,
-      max_children: Pleroma.Config.get([:connections_pool, :max_connections])
+      max_children: Config.get([:connections_pool, :max_connections])
     )
   end
 
-  def start_worker(opts, last_attempt \\ false) do
-    case DynamicSupervisor.start_child(__MODULE__, {Pleroma.Gun.ConnectionPool.Worker, opts}) do
-      {:error, :max_children} ->
-        funs = [fn -> last_attempt end, fn -> match?(:error, free_pool()) end]
+  def start_worker(opts, last_attempt \\ false)
 
-        if Enum.any?(funs, fn fun -> fun.() end) do
-          :telemetry.execute([:pleroma, :connection_pool, :provision_failure], %{opts: opts})
-          {:error, :pool_full}
-        else
-          start_worker(opts, true)
-        end
+  def start_worker(opts, true) do
+    case DynamicSupervisor.start_child(__MODULE__, {Worker, opts}) do
+      {:error, :max_children} ->
+        :telemetry.execute([:pleroma, :connection_pool, :provision_failure], %{opts: opts})
+        {:error, :pool_full}
+
+      res ->
+        res
+    end
+  end
+
+  def start_worker(opts, false) do
+    case DynamicSupervisor.start_child(__MODULE__, {Worker, opts}) do
+      {:error, :max_children} ->
+        free_pool()
+        start_worker(opts, true)
 
       res ->
         res

lib/pleroma/http.ex

@@ -68,7 +68,9 @@ defmodule Pleroma.HTTP do
 
     adapter = Application.get_env(:tesla, :adapter)
 
-    client = Tesla.client(adapter_middlewares(adapter), adapter)
+    extra_middleware = options[:tesla_middleware] || []
+
+    client = Tesla.client(adapter_middlewares(adapter, extra_middleware), adapter)
 
     maybe_limit(
       fn ->
@@ -102,20 +104,21 @@ defmodule Pleroma.HTTP do
     fun.()
   end
 
-  defp adapter_middlewares(Tesla.Adapter.Gun) do
-    [Tesla.Middleware.FollowRedirects, Pleroma.Tesla.Middleware.ConnectionPool]
+  defp adapter_middlewares(Tesla.Adapter.Gun, extra_middleware) do
+    [Tesla.Middleware.FollowRedirects, Pleroma.Tesla.Middleware.ConnectionPool] ++
+      extra_middleware
   end
 
-  defp adapter_middlewares({Tesla.Adapter.Finch, _}) do
-    [Tesla.Middleware.FollowRedirects]
+  defp adapter_middlewares({Tesla.Adapter.Finch, _}, extra_middleware) do
+    [Tesla.Middleware.FollowRedirects] ++ extra_middleware
   end
 
-  defp adapter_middlewares(_) do
+  defp adapter_middlewares(_, extra_middleware) do
     if Pleroma.Config.get(:env) == :test do
       # Emulate redirects in test env, which are handled by adapters in other environments
       [Tesla.Middleware.FollowRedirects]
     else
-      []
+      extra_middleware
     end
   end
 end

lib/pleroma/instances/instance.ex

@@ -10,7 +10,7 @@ defmodule Pleroma.Instances.Instance do
   alias Pleroma.Maps
   alias Pleroma.Repo
   alias Pleroma.User
-  alias Pleroma.Workers.BackgroundWorker
+  alias Pleroma.Workers.DeleteWorker
 
   use Ecto.Schema
 
@@ -297,7 +297,7 @@ defmodule Pleroma.Instances.Instance do
   all of those users' activities and notifications.
   """
   def delete_users_and_activities(host) when is_binary(host) do
-    BackgroundWorker.enqueue("delete_instance", %{"host" => host})
+    DeleteWorker.enqueue("delete_instance", %{"host" => host})
   end
 
   def perform(:delete_instance, host) when is_binary(host) do

lib/pleroma/notification.ex

@@ -734,7 +734,7 @@ defmodule Pleroma.Notification do
 
   def mark_as_read?(activity, target_user) do
     user = Activity.user_actor(activity)
-    User.mutes_user?(target_user, user) || CommonAPI.thread_muted?(target_user, activity)
+    User.mutes_user?(target_user, user) || CommonAPI.thread_muted?(activity, target_user)
   end
 
   def for_user_and_activity(user, activity) do

lib/pleroma/object.ex

@@ -99,21 +99,24 @@ defmodule Pleroma.Object do
   def get_by_id(nil), do: nil
   def get_by_id(id), do: Repo.get(Object, id)
 
+  @spec get_by_id_and_maybe_refetch(integer(), list()) :: Object.t() | nil
   def get_by_id_and_maybe_refetch(id, opts \\ []) do
-    %{updated_at: updated_at} = object = get_by_id(id)
+    with %Object{updated_at: updated_at} = object <- get_by_id(id) do
+      if opts[:interval] &&
+           NaiveDateTime.diff(NaiveDateTime.utc_now(), updated_at) > opts[:interval] do
+        case Fetcher.refetch_object(object) do
+          {:ok, %Object{} = object} ->
+            object
 
-    if opts[:interval] &&
-         NaiveDateTime.diff(NaiveDateTime.utc_now(), updated_at) > opts[:interval] do
-      case Fetcher.refetch_object(object) do
-        {:ok, %Object{} = object} ->
-          object
-
-        e ->
-          Logger.error("Couldn't refresh #{object.data["id"]}:\n#{inspect(e)}")
-          object
+          e ->
+            Logger.error("Couldn't refresh #{object.data["id"]}:\n#{inspect(e)}")
+            object
+        end
+      else
+        object
       end
     else
-      object
+      nil -> nil
     end
   end
 

lib/pleroma/object/fetcher.ex

@@ -59,6 +59,7 @@ defmodule Pleroma.Object.Fetcher do
   end
 
   # Note: will create a Create activity, which we need internally at the moment.
+  @spec fetch_object_from_id(String.t(), list()) :: {:ok, Object.t()} | {:error | :reject, any()}
   def fetch_object_from_id(id, options \\ []) do
     with {_, nil} <- {:fetch_object, Object.get_cached_by_ap_id(id)},
          {_, true} <- {:allowed_depth, Federator.allowed_thread_distance?(options[:depth])},

lib/pleroma/object/updater.ex

@@ -134,7 +134,10 @@ defmodule Pleroma.Object.Updater do
     else
       %{updated_object: updated_data} =
         updated_data
-        |> maybe_update_history(original_data, updated: updated, use_history_in_new_object?: false)
+        |> maybe_update_history(original_data,
+          updated: updated,
+          use_history_in_new_object?: false
+        )
 
       updated_data
       |> Map.put("updated", date)

lib/pleroma/search.ex

@@ -16,6 +16,6 @@ defmodule Pleroma.Search do
 
   def healthcheck_endpoints do
     search_module = Pleroma.Config.get([Pleroma.Search, :module])
-    search_module.healthcheck_endpoints
+    search_module.healthcheck_endpoints()
   end
 end

lib/pleroma/signature.ex

@@ -10,6 +10,14 @@ defmodule Pleroma.Signature do
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
 
+  import Plug.Conn, only: [put_req_header: 3]
+
+  @http_signatures_impl Application.compile_env(
+                          :pleroma,
+                          [__MODULE__, :http_signatures_impl],
+                          HTTPSignatures
+                        )
+
   @known_suffixes ["/publickey", "/main-key"]
 
   def key_id_to_actor_id(key_id) do
@@ -85,4 +93,48 @@ defmodule Pleroma.Signature do
   def signed_date(%NaiveDateTime{} = date) do
     Timex.format!(date, "{WDshort}, {0D} {Mshort} {YYYY} {h24}:{m}:{s} GMT")
   end
+
+  @spec validate_signature(Plug.Conn.t(), String.t()) :: boolean()
+  def validate_signature(%Plug.Conn{} = conn, request_target) do
+    # Newer drafts for HTTP signatures now use @request-target instead of the
+    # old (request-target). We'll now support both for incoming signatures.
+    conn =
+      conn
+      |> put_req_header("(request-target)", request_target)
+      |> put_req_header("@request-target", request_target)
+
+    @http_signatures_impl.validate_conn(conn)
+  end
+
+  @spec validate_signature(Plug.Conn.t()) :: boolean()
+  def validate_signature(%Plug.Conn{} = conn) do
+    # This (request-target) is non-standard, but many implementations do it
+    # this way due to a misinterpretation of
+    # https://datatracker.ietf.org/doc/html/draft-cavage-http-signatures-06
+    # "path" was interpreted as not having the query, though later examples
+    # show that it must be the absolute path + query. This behavior is kept to
+    # make sure most software (Pleroma itself, Mastodon, and probably others)
+    # do not break.
+    request_target = Enum.join([String.downcase(conn.method), conn.request_path], " ")
+
+    # This is the proper way to build the @request-target, as expected by
+    # many HTTP signature libraries, clarified in the following draft:
+    # https://www.ietf.org/archive/id/draft-ietf-httpbis-message-signatures-11.html#section-2.2.6
+    # It is the same as before, but containing the query part as well.
+    proper_target = Enum.join([request_target, "?", conn.query_string], "")
+
+    cond do
+      # Normal, non-standard behavior but expected by Pleroma and more.
+      validate_signature(conn, request_target) ->
+        true
+
+      # Has query string and the previous one failed: let's try the standard.
+      conn.query_string != "" ->
+        validate_signature(conn, proper_target)
+
+      # If there's no query string and signature fails, it's rotten.
+      true ->
+        false
+    end
+  end
 end

lib/pleroma/telemetry/logger.ex

@@ -39,7 +39,7 @@ defmodule Pleroma.Telemetry.Logger do
         _,
         _
       ) do
-    Logger.error(fn ->
+    Logger.debug(fn ->
       "Connection pool failed to reclaim any connections due to all of them being in use. It will have to drop requests for opening connections to new hosts"
     end)
   end
@@ -70,7 +70,7 @@ defmodule Pleroma.Telemetry.Logger do
         %{key: key},
         _
       ) do
-    Logger.warning(fn ->
+    Logger.debug(fn ->
       "Pool worker for #{key}: Client #{inspect(client_pid)} died before releasing the connection with #{inspect(reason)}"
     end)
   end

lib/pleroma/upload.ex

@@ -249,14 +249,16 @@ defmodule Pleroma.Upload do
 
   defp url_from_spec(_upload, _base_url, {:url, url}), do: url
 
+  @spec base_url() :: binary
   def base_url do
     uploader = @config_impl.get([Pleroma.Upload, :uploader])
-    upload_base_url = @config_impl.get([Pleroma.Upload, :base_url])
+    upload_fallback_url = Pleroma.Web.Endpoint.url() <> "/media/"
+    upload_base_url = @config_impl.get([Pleroma.Upload, :base_url]) || upload_fallback_url
     public_endpoint = @config_impl.get([uploader, :public_endpoint])
 
     case uploader do
       Pleroma.Uploaders.Local ->
-        upload_base_url || Pleroma.Web.Endpoint.url() <> "/media/"
+        upload_base_url
 
       Pleroma.Uploaders.S3 ->
         bucket = @config_impl.get([Pleroma.Uploaders.S3, :bucket])
@@ -268,11 +270,14 @@ defmodule Pleroma.Upload do
             !is_nil(truncated_namespace) ->
               truncated_namespace
 
-            !is_nil(namespace) ->
+            !is_nil(namespace) and !is_nil(bucket) ->
               namespace <> ":" <> bucket
 
-            true ->
+            !is_nil(bucket) ->
               bucket
+
+            true ->
+              ""
           end
 
         if public_endpoint do
@@ -285,7 +290,7 @@ defmodule Pleroma.Upload do
         @config_impl.get([Pleroma.Uploaders.IPFS, :get_gateway_url])
 
       _ ->
-        public_endpoint || upload_base_url || Pleroma.Web.Endpoint.url() <> "/media/"
+        public_endpoint || upload_base_url
     end
   end
 end

lib/pleroma/upload/filter/exiftool/strip_location.ex

@@ -16,7 +16,9 @@ defmodule Pleroma.Upload.Filter.Exiftool.StripLocation do
 
   def filter(%Pleroma.Upload{tempfile: file, content_type: "image" <> _}) do
     try do
-      case System.cmd("exiftool", ["-overwrite_original", "-gps:all=", file], parallelism: true) do
+      case System.cmd("exiftool", ["-overwrite_original", "-gps:all=", "-png:all=", file],
+             parallelism: true
+           ) do
         {_response, 0} -> {:ok, :filtered}
         {error, 1} -> {:error, error}
       end

lib/pleroma/user.ex

@@ -38,6 +38,8 @@ defmodule Pleroma.User do
   alias Pleroma.Web.OAuth
   alias Pleroma.Web.RelMe
   alias Pleroma.Workers.BackgroundWorker
+  alias Pleroma.Workers.DeleteWorker
+  alias Pleroma.Workers.UserRefreshWorker
 
   require Logger
   require Pleroma.Constants
@@ -1981,7 +1983,7 @@ defmodule Pleroma.User do
   def delete(%User{} = user) do
     # Purge the user immediately
     purge(user)
-    BackgroundWorker.enqueue("delete_user", %{"user_id" => user.id})
+    DeleteWorker.enqueue("delete_user", %{"user_id" => user.id})
   end
 
   # *Actually* delete the user from the DB
@@ -2154,20 +2156,20 @@ defmodule Pleroma.User do
 
   def fetch_by_ap_id(ap_id), do: ActivityPub.make_user_from_ap_id(ap_id)
 
+  @spec get_or_fetch_by_ap_id(String.t()) :: {:ok, User.t()} | {:error, any()}
   def get_or_fetch_by_ap_id(ap_id) do
-    cached_user = get_cached_by_ap_id(ap_id)
+    with cached_user = %User{} <- get_cached_by_ap_id(ap_id),
+         _ <- maybe_refresh(cached_user) do
+      {:ok, cached_user}
+    else
+      _ -> fetch_by_ap_id(ap_id)
+    end
+  end
 
-    maybe_fetched_user = needs_update?(cached_user) && fetch_by_ap_id(ap_id)
-
-    case {cached_user, maybe_fetched_user} do
-      {_, {:ok, %User{} = user}} ->
-        {:ok, user}
-
-      {%User{} = user, _} ->
-        {:ok, user}
-
-      _ ->
-        {:error, :not_found}
+  defp maybe_refresh(user) do
+    if needs_update?(user) do
+      UserRefreshWorker.new(%{"ap_id" => user.ap_id})
+      |> Oban.insert()
     end
   end
 

lib/pleroma/user/backup.ex

@@ -14,9 +14,10 @@ defmodule Pleroma.User.Backup do
 
   alias Pleroma.Activity
   alias Pleroma.Bookmark
+  alias Pleroma.Config
   alias Pleroma.Repo
+  alias Pleroma.Uploaders.Uploader
   alias Pleroma.User
-  alias Pleroma.User.Backup.State
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.ActivityPub.UserView
@@ -29,71 +30,111 @@ defmodule Pleroma.User.Backup do
     field(:file_name, :string)
     field(:file_size, :integer, default: 0)
     field(:processed, :boolean, default: false)
-    field(:state, State, default: :invalid)
-    field(:processed_number, :integer, default: 0)
+    field(:tempdir, :string)
 
     belongs_to(:user, User, type: FlakeId.Ecto.CompatType)
 
     timestamps()
   end
 
-  @config_impl Application.compile_env(:pleroma, [__MODULE__, :config_impl], Pleroma.Config)
+  @doc """
+  Schedules a job to backup a user if the number of backup requests has not exceeded the limit.
 
-  def create(user, admin_id \\ nil) do
-    with :ok <- validate_limit(user, admin_id),
-         {:ok, backup} <- user |> new() |> Repo.insert() do
-      BackupWorker.process(backup, admin_id)
+  Admins can directly call new/1 and schedule_backup/1 to bypass the limit.
+  """
+  @spec user(User.t()) :: {:ok, t()} | {:error, any()}
+  def user(user) do
+    days = Config.get([__MODULE__, :limit_days])
+
+    with true <- permitted?(user),
+         %__MODULE__{} = backup <- new(user),
+         {:ok, inserted_backup} <- Repo.insert(backup),
+         {:ok, %Oban.Job{}} <- schedule_backup(inserted_backup) do
+      {:ok, inserted_backup}
+    else
+      false ->
+        {:error,
+         dngettext(
+           "errors",
+           "Last export was less than a day ago",
+           "Last export was less than %{days} days ago",
+           days,
+           days: days
+         )}
+
+      e ->
+        {:error, e}
     end
   end
 
+  @doc "Generates a %Backup{} for a user with a random file name"
+  @spec new(User.t()) :: t()
   def new(user) do
     rand_str = :crypto.strong_rand_bytes(32) |> Base.url_encode64(padding: false)
     datetime = Calendar.NaiveDateTime.Format.iso8601_basic(NaiveDateTime.utc_now())
     name = "archive-#{user.nickname}-#{datetime}-#{rand_str}.zip"
 
     %__MODULE__{
-      user_id: user.id,
       content_type: "application/zip",
       file_name: name,
-      state: :pending
+      tempdir: tempdir(),
+      user: user
     }
   end
 
-  def delete(backup) do
-    uploader = Pleroma.Config.get([Pleroma.Upload, :uploader])
+  @doc "Schedules the execution of the provided backup"
+  @spec schedule_backup(t()) :: {:ok, Oban.Job.t()} | {:error, any()}
+  def schedule_backup(backup) do
+    with false <- is_nil(backup.id) do
+      %{"op" => "process", "backup_id" => backup.id}
+      |> BackupWorker.new()
+      |> Oban.insert()
+    else
+      true ->
+        {:error, "Backup is missing id. Please insert it into the Repo first."}
+
+      e ->
+        {:error, e}
+    end
+  end
+
+  @doc "Deletes the backup archive file and removes the database record"
+  @spec delete_archive(t()) :: {:ok, Ecto.Schema.t()} | {:error, Ecto.Changeset.t()}
+  def delete_archive(backup) do
+    uploader = Config.get([Pleroma.Upload, :uploader])
 
     with :ok <- uploader.delete_file(Path.join("backups", backup.file_name)) do
       Repo.delete(backup)
     end
   end
 
-  defp validate_limit(_user, admin_id) when is_binary(admin_id), do: :ok
+  @doc "Schedules a job to delete the backup archive"
+  @spec schedule_delete(t()) :: {:ok, Oban.Job.t()} | {:error, any()}
+  def schedule_delete(backup) do
+    days = Config.get([__MODULE__, :purge_after_days])
+    time = 60 * 60 * 24 * days
+    scheduled_at = Calendar.NaiveDateTime.add!(backup.inserted_at, time)
 
-  defp validate_limit(user, nil) do
-    case get_last(user.id) do
-      %__MODULE__{inserted_at: inserted_at} ->
-        days = Pleroma.Config.get([__MODULE__, :limit_days])
-        diff = Timex.diff(NaiveDateTime.utc_now(), inserted_at, :days)
+    %{"op" => "delete", "backup_id" => backup.id}
+    |> BackupWorker.new(scheduled_at: scheduled_at)
+    |> Oban.insert()
+  end
 
-        if diff > days do
-          :ok
-        else
-          {:error,
-           dngettext(
-             "errors",
-             "Last export was less than a day ago",
-             "Last export was less than %{days} days ago",
-             days,
-             days: days
-           )}
-        end
-
-      nil ->
-        :ok
+  defp permitted?(user) do
+    with {_, %__MODULE__{inserted_at: inserted_at}} <- {:last, get_last(user)},
+         days = Config.get([__MODULE__, :limit_days]),
+         diff = Timex.diff(NaiveDateTime.utc_now(), inserted_at, :days),
+         {_, true} <- {:diff, diff > days} do
+      true
+    else
+      {:last, nil} -> true
+      {:diff, false} -> false
     end
   end
 
-  def get_last(user_id) do
+  @doc "Returns last backup for the provided user"
+  @spec get_last(User.t()) :: t()
+  def get_last(%User{id: user_id}) do
     __MODULE__
     |> where(user_id: ^user_id)
     |> order_by(desc: :id)
@@ -101,6 +142,8 @@ defmodule Pleroma.User.Backup do
     |> Repo.one()
   end
 
+  @doc "Lists all existing backups for a user"
+  @spec list(User.t()) :: [Ecto.Schema.t() | term()]
   def list(%User{id: user_id}) do
     __MODULE__
     |> where(user_id: ^user_id)
@@ -108,149 +151,107 @@ defmodule Pleroma.User.Backup do
     |> Repo.all()
   end
 
-  def remove_outdated(%__MODULE__{id: latest_id, user_id: user_id}) do
-    __MODULE__
-    |> where(user_id: ^user_id)
-    |> where([b], b.id != ^latest_id)
-    |> Repo.all()
-    |> Enum.each(&BackupWorker.delete/1)
+  @doc "Schedules deletion of all but the the most recent backup"
+  @spec remove_outdated(User.t()) :: :ok
+  def remove_outdated(user) do
+    with %__MODULE__{} = latest_backup <- get_last(user) do
+      __MODULE__
+      |> where(user_id: ^user.id)
+      |> where([b], b.id != ^latest_backup.id)
+      |> Repo.all()
+      |> Enum.each(&schedule_delete/1)
+    else
+      _ -> :ok
+    end
   end
 
-  def get(id), do: Repo.get(__MODULE__, id)
-
-  defp set_state(backup, state, processed_number \\ nil) do
-    struct =
-      %{state: state}
-      |> Pleroma.Maps.put_if_present(:processed_number, processed_number)
+  def get_by_id(id), do: Repo.get(__MODULE__, id)
 
+  @doc "Generates changeset for %Pleroma.User.Backup{}"
+  @spec changeset(%__MODULE__{}, map()) :: %Ecto.Changeset{}
+  def changeset(backup \\ %__MODULE__{}, attrs) do
     backup
-    |> cast(struct, [:state, :processed_number])
+    |> cast(attrs, [:content_type, :file_name, :file_size, :processed, :tempdir])
+  end
+
+  @doc "Updates the backup record"
+  @spec update_record(%__MODULE__{}, map()) :: {:ok, %__MODULE__{}} | {:error, %Ecto.Changeset{}}
+  def update_record(%__MODULE__{} = backup, attrs) do
+    backup
+    |> changeset(attrs)
     |> Repo.update()
   end
 
-  def process(
-        %__MODULE__{} = backup,
-        processor_module \\ __MODULE__.Processor
-      ) do
-    set_state(backup, :running, 0)
-
-    current_pid = self()
-
-    task =
-      Task.Supervisor.async_nolink(
-        Pleroma.TaskSupervisor,
-        processor_module,
-        :do_process,
-        [backup, current_pid]
-      )
-
-    wait_backup(backup, backup.processed_number, task)
-  end
-
-  defp wait_backup(backup, current_processed, task) do
-    wait_time = @config_impl.get([__MODULE__, :process_wait_time])
-
-    receive do
-      {:progress, new_processed} ->
-        total_processed = current_processed + new_processed
-
-        set_state(backup, :running, total_processed)
-        wait_backup(backup, total_processed, task)
-
-      {:DOWN, _ref, _proc, _pid, reason} ->
-        backup = get(backup.id)
-
-        if reason != :normal do
-          Logger.error("Backup #{backup.id} process ended abnormally: #{inspect(reason)}")
-
-          {:ok, backup} = set_state(backup, :failed)
-
-          cleanup(backup)
-
-          {:error,
-           %{
-             backup: backup,
-             reason: :exit,
-             details: reason
-           }}
-        else
-          {:ok, backup}
-        end
-    after
-      wait_time ->
-        Logger.error(
-          "Backup #{backup.id} timed out after no response for #{wait_time}ms, terminating"
-        )
-
-        Task.Supervisor.terminate_child(Pleroma.TaskSupervisor, task.pid)
-
-        {:ok, backup} = set_state(backup, :failed)
-
-        cleanup(backup)
-
-        {:error,
-         %{
-           backup: backup,
-           reason: :timeout
-         }}
-    end
-  end
-
   @files [
-    'actor.json',
-    'outbox.json',
-    'likes.json',
-    'bookmarks.json',
-    'followers.json',
-    'following.json'
+    ~c"actor.json",
+    ~c"outbox.json",
+    ~c"likes.json",
+    ~c"bookmarks.json",
+    ~c"followers.json",
+    ~c"following.json"
   ]
-  @spec export(Pleroma.User.Backup.t(), pid()) :: {:ok, String.t()} | :error
-  def export(%__MODULE__{} = backup, caller_pid) do
-    backup = Repo.preload(backup, :user)
-    dir = backup_tempdir(backup)
 
-    with :ok <- File.mkdir(dir),
-         :ok <- actor(dir, backup.user, caller_pid),
-         :ok <- statuses(dir, backup.user, caller_pid),
-         :ok <- likes(dir, backup.user, caller_pid),
-         :ok <- bookmarks(dir, backup.user, caller_pid),
-         :ok <- followers(dir, backup.user, caller_pid),
-         :ok <- following(dir, backup.user, caller_pid),
-         {:ok, zip_path} <- :zip.create(backup.file_name, @files, cwd: dir),
-         {:ok, _} <- File.rm_rf(dir) do
-      {:ok, zip_path}
+  @spec run(t()) :: {:ok, t()} | {:error, :failed}
+  def run(%__MODULE__{} = backup) do
+    backup = Repo.preload(backup, :user)
+    tempfile = Path.join([backup.tempdir, backup.file_name])
+
+    with {_, :ok} <- {:mkdir, File.mkdir_p(backup.tempdir)},
+         {_, :ok} <- {:actor, actor(backup.tempdir, backup.user)},
+         {_, :ok} <- {:statuses, statuses(backup.tempdir, backup.user)},
+         {_, :ok} <- {:likes, likes(backup.tempdir, backup.user)},
+         {_, :ok} <- {:bookmarks, bookmarks(backup.tempdir, backup.user)},
+         {_, :ok} <- {:followers, followers(backup.tempdir, backup.user)},
+         {_, :ok} <- {:following, following(backup.tempdir, backup.user)},
+         {_, {:ok, _zip_path}} <-
+           {:zip, :zip.create(to_charlist(tempfile), @files, cwd: to_charlist(backup.tempdir))},
+         {_, {:ok, %File.Stat{size: zip_size}}} <- {:filestat, File.stat(tempfile)},
+         {:ok, updated_backup} <- update_record(backup, %{file_size: zip_size}) do
+      {:ok, updated_backup}
     else
-      _ -> :error
+      _ ->
+        File.rm_rf(backup.tempdir)
+        {:error, :failed}
     end
   end
 
-  def dir(name) do
-    dir = Pleroma.Config.get([__MODULE__, :dir]) || System.tmp_dir!()
-    Path.join(dir, name)
+  defp tempdir do
+    rand = :crypto.strong_rand_bytes(8) |> Base.url_encode64(padding: false)
+    subdir = "backup-#{rand}"
+
+    case Config.get([__MODULE__, :tempdir]) do
+      nil ->
+        Path.join([System.tmp_dir!(), subdir])
+
+      path ->
+        Path.join([path, subdir])
+    end
   end
 
-  def upload(%__MODULE__{} = backup, zip_path) do
-    uploader = Pleroma.Config.get([Pleroma.Upload, :uploader])
+  @doc "Uploads the completed backup and marks it as processed"
+  @spec upload(t()) :: {:ok, t()}
+  def upload(%__MODULE__{tempdir: tempdir} = backup) when is_binary(tempdir) do
+    uploader = Config.get([Pleroma.Upload, :uploader])
 
     upload = %Pleroma.Upload{
       name: backup.file_name,
-      tempfile: zip_path,
+      tempfile: Path.join([tempdir, backup.file_name]),
       content_type: backup.content_type,
       path: Path.join("backups", backup.file_name)
     }
 
-    with {:ok, _} <- Pleroma.Uploaders.Uploader.put_file(uploader, upload),
-         :ok <- File.rm(zip_path) do
-      {:ok, upload}
+    with {:ok, _} <- Uploader.put_file(uploader, upload),
+         {:ok, uploaded_backup} <- update_record(backup, %{processed: true}),
+         {:ok, _} <- File.rm_rf(tempdir) do
+      {:ok, uploaded_backup}
     end
   end
 
-  defp actor(dir, user, caller_pid) do
+  defp actor(dir, user) do
     with {:ok, json} <-
            UserView.render("user.json", %{user: user})
            |> Map.merge(%{"likes" => "likes.json", "bookmarks" => "bookmarks.json"})
            |> Jason.encode() do
-      send(caller_pid, {:progress, 1})
       File.write(Path.join(dir, "actor.json"), json)
     end
   end
@@ -269,22 +270,10 @@ defmodule Pleroma.User.Backup do
     )
   end
 
-  defp should_report?(num, chunk_size), do: rem(num, chunk_size) == 0
-
-  defp backup_tempdir(backup) do
-    name = String.trim_trailing(backup.file_name, ".zip")
-    dir(name)
-  end
-
-  defp cleanup(backup) do
-    dir = backup_tempdir(backup)
-    File.rm_rf(dir)
-  end
-
-  defp write(query, dir, name, fun, caller_pid) do
+  defp write(query, dir, name, fun) do
     path = Path.join(dir, "#{name}.json")
 
-    chunk_size = Pleroma.Config.get([__MODULE__, :process_chunk_size])
+    chunk_size = Config.get([__MODULE__, :process_chunk_size])
 
     with {:ok, file} <- File.open(path, [:write, :utf8]),
          :ok <- write_header(file, name) do
@@ -300,10 +289,6 @@ defmodule Pleroma.User.Backup do
                   end),
                {:ok, str} <- Jason.encode(data),
                :ok <- IO.write(file, str <> ",\n") do
-            if should_report?(acc + 1, chunk_size) do
-              send(caller_pid, {:progress, chunk_size})
-            end
-
             acc + 1
           else
             {:error, e} ->
@@ -318,31 +303,29 @@ defmodule Pleroma.User.Backup do
           end
         end)
 
-      send(caller_pid, {:progress, rem(total, chunk_size)})
-
       with :ok <- :file.pwrite(file, {:eof, -2}, "\n],\n  \"totalItems\": #{total}}") do
         File.close(file)
       end
     end
   end
 
-  defp bookmarks(dir, %{id: user_id} = _user, caller_pid) do
+  defp bookmarks(dir, %{id: user_id} = _user) do
     Bookmark
     |> where(user_id: ^user_id)
     |> join(:inner, [b], activity in assoc(b, :activity))
     |> select([b, a], %{id: b.id, object: fragment("(?)->>'object'", a.data)})
-    |> write(dir, "bookmarks", fn a -> {:ok, a.object} end, caller_pid)
+    |> write(dir, "bookmarks", fn a -> {:ok, a.object} end)
   end
 
-  defp likes(dir, user, caller_pid) do
+  defp likes(dir, user) do
     user.ap_id
     |> Activity.Queries.by_actor()
     |> Activity.Queries.by_type("Like")
     |> select([like], %{id: like.id, object: fragment("(?)->>'object'", like.data)})
-    |> write(dir, "likes", fn a -> {:ok, a.object} end, caller_pid)
+    |> write(dir, "likes", fn a -> {:ok, a.object} end)
   end
 
-  defp statuses(dir, user, caller_pid) do
+  defp statuses(dir, user) do
     opts =
       %{}
       |> Map.put(:type, ["Create", "Announce"])
@@ -362,52 +345,17 @@ defmodule Pleroma.User.Backup do
         with {:ok, activity} <- Transmogrifier.prepare_outgoing(a.data) do
           {:ok, Map.delete(activity, "@context")}
         end
-      end,
-      caller_pid
+      end
     )
   end
 
-  defp followers(dir, user, caller_pid) do
+  defp followers(dir, user) do
     User.get_followers_query(user)
-    |> write(dir, "followers", fn a -> {:ok, a.ap_id} end, caller_pid)
+    |> write(dir, "followers", fn a -> {:ok, a.ap_id} end)
   end
 
-  defp following(dir, user, caller_pid) do
+  defp following(dir, user) do
     User.get_friends_query(user)
-    |> write(dir, "following", fn a -> {:ok, a.ap_id} end, caller_pid)
-  end
-end
-
-defmodule Pleroma.User.Backup.ProcessorAPI do
-  @callback do_process(%Pleroma.User.Backup{}, pid()) ::
-              {:ok, %Pleroma.User.Backup{}} | {:error, any()}
-end
-
-defmodule Pleroma.User.Backup.Processor do
-  @behaviour Pleroma.User.Backup.ProcessorAPI
-
-  alias Pleroma.Repo
-  alias Pleroma.User.Backup
-
-  import Ecto.Changeset
-
-  @impl true
-  def do_process(backup, current_pid) do
-    with {:ok, zip_file} <- Backup.export(backup, current_pid),
-         {:ok, %{size: size}} <- File.stat(zip_file),
-         {:ok, _upload} <- Backup.upload(backup, zip_file) do
-      backup
-      |> cast(
-        %{
-          file_size: size,
-          processed: true,
-          state: :complete
-        },
-        [:file_size, :processed, :state]
-      )
-      |> Repo.update()
-    else
-      e -> {:error, e}
-    end
+    |> write(dir, "following", fn a -> {:ok, a.ap_id} end)
   end
 end

lib/pleroma/user/import.ex

@@ -31,7 +31,7 @@ defmodule Pleroma.User.Import do
       identifiers,
       fn identifier ->
         with {:ok, %User{} = blocked} <- User.get_or_fetch(identifier),
-             {:ok, _block} <- CommonAPI.block(blocker, blocked) do
+             {:ok, _block} <- CommonAPI.block(blocked, blocker) do
           blocked
         else
           error -> handle_error(:blocks_import, identifier, error)
@@ -46,7 +46,7 @@ defmodule Pleroma.User.Import do
       fn identifier ->
         with {:ok, %User{} = followed} <- User.get_or_fetch(identifier),
              {:ok, follower, followed} <- User.maybe_direct_follow(follower, followed),
-             {:ok, _, _, _} <- CommonAPI.follow(follower, followed) do
+             {:ok, _, _, _} <- CommonAPI.follow(followed, follower) do
           followed
         else
           error -> handle_error(:follow_import, identifier, error)

lib/pleroma/web.ex

@@ -163,7 +163,7 @@ defmodule Pleroma.Web do
       """
       def safe_render_many(collection, view, template, assigns \\ %{}) do
         Enum.map(collection, fn resource ->
-          as = Map.get(assigns, :as) || view.__resource__
+          as = Map.get(assigns, :as) || view.__resource__()
           assigns = Map.put(assigns, as, resource)
           safe_render(view, template, assigns)
         end)

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1661,7 +1661,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
        }}
     else
       {:error, _} = e -> e
-      e -> {:error, e}
     end
   end
 

lib/pleroma/web/activity_pub/activity_pub_controller.ex

@@ -293,8 +293,15 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubController do
     json(conn, "ok")
   end
 
-  def inbox(%{assigns: %{valid_signature: false}, req_headers: req_headers} = conn, params) do
-    Federator.incoming_ap_doc(%{req_headers: req_headers, params: params})
+  def inbox(%{assigns: %{valid_signature: false}} = conn, params) do
+    Federator.incoming_ap_doc(%{
+      method: conn.method,
+      req_headers: conn.req_headers,
+      request_path: conn.request_path,
+      params: params,
+      query_string: conn.query_string
+    })
+
     json(conn, "ok")
   end
 

lib/pleroma/web/activity_pub/mrf.ex

@@ -204,7 +204,7 @@ defmodule Pleroma.Web.ActivityPub.MRF do
       if function_exported?(policy, :config_description, 0) do
         description =
           @default_description
-          |> Map.merge(policy.config_description)
+          |> Map.merge(policy.config_description())
           |> Map.put(:group, :pleroma)
           |> Map.put(:tab, :mrf)
           |> Map.put(:type, :group)

lib/pleroma/web/activity_pub/mrf/follow_bot_policy.ex

@@ -49,7 +49,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.FollowBotPolicy do
           "#{__MODULE__}: Follow request from #{follower.nickname} to #{user.nickname}"
         )
 
-        CommonAPI.follow(follower, user)
+        CommonAPI.follow(user, follower)
       end
     end)
 

lib/pleroma/web/activity_pub/mrf/nsfw_api_policy.ex

@@ -137,7 +137,6 @@ defmodule Pleroma.Web.ActivityPub.MRF.NsfwApiPolicy do
       {:ok, object}
     else
       {:nsfw, _data} -> handle_nsfw(object)
-      _ -> {:reject, "NSFW: Attachment rejected"}
     end
   end
 

lib/pleroma/web/activity_pub/mrf/simple_policy.ex

@@ -220,9 +220,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.SimplePolicy do
          {:ok, object} <- check_object(object) do
       {:ok, object}
     else
-      {:reject, nil} -> {:reject, "[SimplePolicy]"}
       {:reject, _} = e -> e
-      _ -> {:reject, "[SimplePolicy]"}
     end
   end
 
@@ -236,9 +234,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.SimplePolicy do
          {:ok, object} <- check_banner_removal(actor_info, object) do
       {:ok, object}
     else
-      {:reject, nil} -> {:reject, "[SimplePolicy]"}
       {:reject, _} = e -> e
-      _ -> {:reject, "[SimplePolicy]"}
     end
   end
 
@@ -249,9 +245,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.SimplePolicy do
          {:ok, object} <- check_reject(uri, object) do
       {:ok, object}
     else
-      {:reject, nil} -> {:reject, "[SimplePolicy]"}
       {:reject, _} = e -> e
-      _ -> {:reject, "[SimplePolicy]"}
     end
   end
 

lib/pleroma/web/activity_pub/mrf/vocabulary_policy.ex

@@ -31,7 +31,6 @@ defmodule Pleroma.Web.ActivityPub.MRF.VocabularyPolicy do
       {:reject, _} = e -> e
       {:accepted, _} -> {:reject, "[VocabularyPolicy] #{message_type} not in accept list"}
       {:rejected, _} -> {:reject, "[VocabularyPolicy] #{message_type} in reject list"}
-      _ -> {:reject, "[VocabularyPolicy]"}
     end
   end
 

lib/pleroma/web/activity_pub/pipeline.ex

@@ -23,7 +23,7 @@ defmodule Pleroma.Web.ActivityPub.Pipeline do
   defp config, do: Config.get([:pipeline, :config], Config)
 
   @spec common_pipeline(map(), keyword()) ::
-          {:ok, Activity.t() | Object.t(), keyword()} | {:error, any()}
+          {:ok, Activity.t() | Object.t(), keyword()} | {:error | :reject, any()}
   def common_pipeline(object, meta) do
     case Repo.transaction(fn -> do_common_pipeline(object, meta) end, Utils.query_timeout()) do
       {:ok, {:ok, activity, meta}} ->

lib/pleroma/web/activity_pub/publisher.ex

@@ -80,13 +80,26 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
   parameters set:
 
   * `inbox`: the inbox to publish to
-  * `json`: the JSON message body representing the ActivityPub message
-  * `actor`: the actor which is signing the message
-  * `id`: the ActivityStreams URI of the message
+  * `activity_id`: the internal activity id
+  * `cc`: the cc recipients relevant to this inbox (optional)
   """
-  def publish_one(%{inbox: inbox, json: json, actor: %User{} = actor, id: id} = params) do
-    Logger.debug("Federating #{id} to #{inbox}")
+  def publish_one(%{inbox: inbox, activity_id: activity_id} = params) do
+    activity = Activity.get_by_id_with_user_actor(activity_id)
+    actor = activity.user_actor
+
+    ap_id = activity.data["id"]
+    Logger.debug("Federating #{ap_id} to #{inbox}")
     uri = %{path: path} = URI.parse(inbox)
+
+    {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
+
+    cc = Map.get(params, :cc)
+
+    json =
+      data
+      |> Map.put("cc", cc)
+      |> Jason.encode!()
+
     digest = "SHA-256=" <> (:crypto.hash(:sha256, json) |> Base.encode64())
 
     date = Pleroma.Signature.signed_date()
@@ -119,13 +132,14 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
     else
       {_post_result, %{status: code} = response} = e ->
         unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
-        Logger.metadata(activity: id, inbox: inbox, status: code)
+        Logger.metadata(activity: activity_id, inbox: inbox, status: code)
         Logger.error("Publisher failed to inbox #{inbox} with status #{code}")
 
         case response do
-          %{status: 403} -> {:discard, :forbidden}
-          %{status: 404} -> {:discard, :not_found}
-          %{status: 410} -> {:discard, :not_found}
+          %{status: 400} -> {:cancel, :bad_request}
+          %{status: 403} -> {:cancel, :forbidden}
+          %{status: 404} -> {:cancel, :not_found}
+          %{status: 410} -> {:cancel, :not_found}
           _ -> {:error, e}
         end
 
@@ -135,21 +149,12 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
 
       e ->
         unless params[:unreachable_since], do: Instances.set_unreachable(inbox)
-        Logger.metadata(activity: id, inbox: inbox)
+        Logger.metadata(activity: activity_id, inbox: inbox)
         Logger.error("Publisher failed to inbox #{inbox} #{inspect(e)}")
         {:error, e}
     end
   end
 
-  def publish_one(%{actor_id: actor_id} = params) do
-    actor = User.get_cached_by_id(actor_id)
-
-    params
-    |> Map.delete(:actor_id)
-    |> Map.put(:actor, actor)
-    |> publish_one()
-  end
-
   defp signature_host(%URI{port: port, scheme: scheme, host: host}) do
     if port == URI.default_port(scheme) do
       host
@@ -250,7 +255,6 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
   def publish(%User{} = actor, %{data: %{"bcc" => bcc}} = activity)
       when is_list(bcc) and bcc != [] do
     public = public?(activity)
-    {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
 
     [priority_recipients, recipients] = recipients(actor, activity)
 
@@ -275,16 +279,10 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
           # instance would only accept a first message for the first recipient and ignore the rest.
           cc = get_cc_ap_ids(ap_id, recipients)
 
-          json =
-            data
-            |> Map.put("cc", cc)
-            |> Jason.encode!()
-
           __MODULE__.enqueue_one(%{
             inbox: inbox,
-            json: json,
-            actor_id: actor.id,
-            id: activity.data["id"],
+            cc: cc,
+            activity_id: activity.id,
             unreachable_since: unreachable_since
           })
         end)
@@ -301,9 +299,6 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
       Relay.publish(activity)
     end
 
-    {:ok, data} = Transmogrifier.prepare_outgoing(activity.data)
-    json = Jason.encode!(data)
-
     [priority_inboxes, inboxes] =
       recipients(actor, activity)
       |> Enum.map(fn recipients ->
@@ -325,9 +320,7 @@ defmodule Pleroma.Web.ActivityPub.Publisher do
         __MODULE__.enqueue_one(
           %{
             inbox: inbox,
-            json: json,
-            actor_id: actor.id,
-            id: activity.data["id"],
+            activity_id: activity.id,
             unreachable_since: unreachable_since
           },
           priority: priority

lib/pleroma/web/activity_pub/relay.ex

@@ -22,7 +22,7 @@ defmodule Pleroma.Web.ActivityPub.Relay do
   def follow(target_instance) do
     with %User{} = local_user <- get_actor(),
          {:ok, %User{} = target_user} <- User.get_or_fetch_by_ap_id(target_instance),
-         {:ok, _, _, activity} <- CommonAPI.follow(local_user, target_user) do
+         {:ok, _, _, activity} <- CommonAPI.follow(target_user, local_user) do
       Logger.info("relay: followed instance: #{target_instance}; id=#{activity.data["id"]}")
       {:ok, activity}
     else

lib/pleroma/web/activity_pub/side_effects.ex

@@ -453,7 +453,7 @@ defmodule Pleroma.Web.ActivityPub.SideEffects do
        ) do
     orig_object_ap_id = updated_object["id"]
     orig_object = Object.get_by_ap_id(orig_object_ap_id)
-    orig_object_data = orig_object.data
+    orig_object_data = Map.get(orig_object, :data)
 
     updated_object =
       if meta[:local] do

lib/pleroma/web/activity_pub/transmogrifier.ex

@@ -534,6 +534,9 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
         else
           _ -> e
         end
+
+      e ->
+        {:error, e}
     end
   end
 

lib/pleroma/web/activity_pub/utils.ex

@@ -946,26 +946,14 @@ defmodule Pleroma.Web.ActivityPub.Utils do
     |> Repo.all()
   end
 
+  @spec maybe_handle_group_posts(Activity.t()) :: :ok
+  @doc "Automatically repeats posts for local group actor recipients"
   def maybe_handle_group_posts(activity) do
     poster = User.get_cached_by_ap_id(activity.actor)
 
-    mentions =
-      activity.data["to"]
-      |> Enum.filter(&(&1 != activity.actor))
-
-    mentioned_local_groups =
-      User.get_all_by_ap_id(mentions)
-      |> Enum.filter(fn user ->
-        user.actor_type == "Group" and
-          user.local and
-          not User.blocks?(user, poster)
-      end)
-
-    mentioned_local_groups
-    |> Enum.each(fn group ->
-      Pleroma.Web.CommonAPI.repeat(activity.id, group)
-    end)
-
-    :ok
+    User.get_recipients_from_activity(activity)
+    |> Enum.filter(&match?("Group", &1.actor_type))
+    |> Enum.reject(&User.blocks?(&1, poster))
+    |> Enum.each(&Pleroma.Web.CommonAPI.repeat(activity.id, &1))
   end
 end

lib/pleroma/web/admin_api/controllers/admin_api_controller.ex

@@ -13,6 +13,7 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIController do
   alias Pleroma.ModerationLog
   alias Pleroma.Stats
   alias Pleroma.User
+  alias Pleroma.User.Backup
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.AdminAPI
   alias Pleroma.Web.AdminAPI.AccountView
@@ -429,7 +430,9 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIController do
 
   def create_backup(%{assigns: %{user: admin}} = conn, %{"nickname" => nickname}) do
     with %User{} = user <- User.get_by_nickname(nickname),
-         {:ok, _} <- Pleroma.User.Backup.create(user, admin.id) do
+         %Backup{} = backup <- Backup.new(user),
+         {:ok, inserted_backup} <- Pleroma.Repo.insert(backup),
+         {:ok, %Oban.Job{}} <- Backup.schedule_backup(inserted_backup) do
       ModerationLog.insert_log(%{actor: admin, subject: user, action: "create_backup"})
 
       json(conn, "")

lib/pleroma/web/admin_api/controllers/invite_controller.ex

@@ -46,7 +46,6 @@ defmodule Pleroma.Web.AdminAPI.InviteController do
       render(conn, "show.json", invite: updated_invite)
     else
       nil -> {:error, :not_found}
-      error -> error
     end
   end
 

lib/pleroma/web/admin_api/controllers/rule_controller.ex

@@ -24,7 +24,7 @@ defmodule Pleroma.Web.AdminAPI.RuleController do
 
   plug(OAuthScopesPlug, %{scopes: ["admin:read"]} when action == :index)
 
-  action_fallback(AdminAPI.FallbackController)
+  action_fallback(Pleroma.Web.AdminAPI.FallbackController)
 
   defdelegate open_api_operation(action), to: Pleroma.Web.ApiSpec.Admin.RuleOperation
 

lib/pleroma/web/api_spec/operations/pleroma_backup_operation.ex

@@ -65,12 +65,7 @@ defmodule Pleroma.Web.ApiSpec.PleromaBackupOperation do
         file_name: %Schema{type: :string},
         file_size: %Schema{type: :integer},
         processed: %Schema{type: :boolean, description: "whether this backup has succeeded"},
-        state: %Schema{
-          type: :string,
-          description: "the state of the backup",
-          enum: ["pending", "running", "complete", "failed"]
-        },
-        processed_number: %Schema{type: :integer, description: "the number of records processed"}
+        tempdir: %Schema{type: :string}
       },
       example: %{
         "content_type" => "application/zip",
@@ -79,8 +74,7 @@ defmodule Pleroma.Web.ApiSpec.PleromaBackupOperation do
         "file_size" => 4105,
         "inserted_at" => "2020-09-08T16:42:07.000Z",
         "processed" => true,
-        "state" => "complete",
-        "processed_number" => 20
+        "tempdir" => "/tmp/PZIMw40vmpM"
       }
     }
   end

lib/pleroma/web/api_spec/operations/search_operation.ex

@@ -79,7 +79,9 @@ defmodule Pleroma.Web.ApiSpec.SearchOperation do
           %Schema{type: :string, enum: ["accounts", "hashtags", "statuses"]},
           "Search type"
         ),
-        Operation.parameter(:q, :query, %Schema{type: :string}, "The search query", required: true),
+        Operation.parameter(:q, :query, %Schema{type: :string}, "The search query",
+          required: true
+        ),
         Operation.parameter(
           :resolve,
           :query,

lib/pleroma/web/api_spec/operations/streaming_operation.ex

@@ -139,7 +139,7 @@ defmodule Pleroma.Web.ApiSpec.StreamingOperation do
   end
 
   defp get_schema(%Schema{} = schema), do: schema
-  defp get_schema(schema), do: schema.schema
+  defp get_schema(schema), do: schema.schema()
 
   defp server_sent_event_helper(name, description, type, payload, opts \\ []) do
     payload_type = Keyword.get(opts, :payload_type, :json)

lib/pleroma/web/api_spec/schemas/chat.ex

@@ -68,7 +68,7 @@ defmodule Pleroma.Web.ApiSpec.Schemas.Chat do
       },
       "id" => "1",
       "unread" => 2,
-      "last_message" => ChatMessage.schema().example(),
+      "last_message" => ChatMessage.schema().example,
       "updated_at" => "2020-04-21T15:06:45.000Z"
     }
   })

lib/pleroma/web/auth/ldap_authenticator.ex

@@ -91,7 +91,8 @@ defmodule Pleroma.Web.Auth.LDAPAuthenticator do
         end
 
       error ->
-        error
+        Logger.error("Could not bind LDAP user #{name}: #{inspect(error)}")
+        {:error, {:ldap_bind_error, error}}
     end
   end
 
@@ -102,28 +103,38 @@ defmodule Pleroma.Web.Auth.LDAPAuthenticator do
            {:scope, :eldap.wholeSubtree()},
            {:timeout, @search_timeout}
          ]) do
-      {:ok, {:eldap_search_result, [{:eldap_entry, _, attributes}], _}} ->
-        params = %{
-          name: name,
-          nickname: name,
-          password: nil
-        }
+      # The :eldap_search_result record structure changed in OTP 24.3 and added a controls field
+      # https://github.com/erlang/otp/pull/5538
+      {:ok, {:eldap_search_result, [{:eldap_entry, _object, attributes}], _referrals}} ->
+        try_register(name, attributes)
 
-        params =
-          case List.keyfind(attributes, 'mail', 0) do
-            {_, [mail]} -> Map.put_new(params, :email, :erlang.list_to_binary(mail))
-            _ -> params
-          end
-
-        changeset = User.register_changeset_ldap(%User{}, params)
-
-        case User.register(changeset) do
-          {:ok, user} -> user
-          error -> error
-        end
+      {:ok, {:eldap_search_result, [{:eldap_entry, _object, attributes}], _referrals, _controls}} ->
+        try_register(name, attributes)
 
       error ->
-        error
+        Logger.error("Couldn't register user because LDAP search failed: #{inspect(error)}")
+        {:error, {:ldap_search_error, error}}
+    end
+  end
+
+  defp try_register(name, attributes) do
+    params = %{
+      name: name,
+      nickname: name,
+      password: nil
+    }
+
+    params =
+      case List.keyfind(attributes, ~c"mail", 0) do
+        {_, [mail]} -> Map.put_new(params, :email, :erlang.list_to_binary(mail))
+        _ -> params
+      end
+
+    changeset = User.register_changeset_ldap(%User{}, params)
+
+    case User.register(changeset) do
+      {:ok, user} -> user
+      error -> error
     end
   end
 end

lib/pleroma/web/common_api.ex

@@ -19,19 +19,23 @@ defmodule Pleroma.Web.CommonAPI do
   alias Pleroma.Web.ActivityPub.Visibility
   alias Pleroma.Web.CommonAPI.ActivityDraft
 
+  import Ecto.Query, only: [where: 3]
   import Pleroma.Web.Gettext
   import Pleroma.Web.CommonAPI.Utils
 
   require Pleroma.Constants
   require Logger
 
-  def block(blocker, blocked) do
+  @spec block(User.t(), User.t()) :: {:ok, Activity.t()} | {:error, any()}
+  def block(blocked, blocker) do
     with {:ok, block_data, _} <- Builder.block(blocker, blocked),
          {:ok, block, _} <- Pipeline.common_pipeline(block_data, local: true) do
       {:ok, block}
     end
   end
 
+  @spec post_chat_message(User.t(), User.t(), String.t(), list()) ::
+          {:ok, Activity.t()} | {:error, any()}
   def post_chat_message(%User{} = user, %User{} = recipient, content, opts \\ []) do
     with maybe_attachment <- opts[:media_id] && Object.get_by_id(opts[:media_id]),
          :ok <- validate_chat_attachment_attribution(maybe_attachment, user),
@@ -95,7 +99,8 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def unblock(blocker, blocked) do
+  @spec unblock(User.t(), User.t()) :: {:ok, Activity.t()} | {:error, any()}
+  def unblock(blocked, blocker) do
     with {_, %Activity{} = block} <- {:fetch_block, Utils.fetch_latest_block(blocker, blocked)},
          {:ok, unblock_data, _} <- Builder.undo(blocker, block),
          {:ok, unblock, _} <- Pipeline.common_pipeline(unblock_data, local: true) do
@@ -114,7 +119,9 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def follow(follower, followed) do
+  @spec follow(User.t(), User.t()) ::
+          {:ok, User.t(), User.t(), Activity.t() | Object.t()} | {:error, :rejected}
+  def follow(followed, follower) do
     timeout = Pleroma.Config.get([:activitypub, :follow_handshake_timeout])
 
     with {:ok, follow_data, _} <- Builder.follow(follower, followed),
@@ -128,7 +135,8 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def unfollow(follower, unfollowed) do
+  @spec unfollow(User.t(), User.t()) :: {:ok, User.t()} | {:error, any()}
+  def unfollow(unfollowed, follower) do
     with {:ok, follower, _follow_activity} <- User.unfollow(follower, unfollowed),
          {:ok, _activity} <- ActivityPub.unfollow(follower, unfollowed),
          {:ok, _subscription} <- User.unsubscribe(follower, unfollowed),
@@ -137,6 +145,7 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec accept_follow_request(User.t(), User.t()) :: {:ok, User.t()} | {:error, any()}
   def accept_follow_request(follower, followed) do
     with %Activity{} = follow_activity <- Utils.fetch_latest_follow(follower, followed),
          {:ok, accept_data, _} <- Builder.accept(followed, follow_activity),
@@ -145,6 +154,7 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec reject_follow_request(User.t(), User.t()) :: {:ok, User.t()} | {:error, any()} | nil
   def reject_follow_request(follower, followed) do
     with %Activity{} = follow_activity <- Utils.fetch_latest_follow(follower, followed),
          {:ok, reject_data, _} <- Builder.reject(followed, follow_activity),
@@ -153,9 +163,11 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec delete(String.t(), User.t()) :: {:ok, Activity.t()} | {:error, any()}
   def delete(activity_id, user) do
     with {_, %Activity{data: %{"object" => _, "type" => "Create"}} = activity} <-
            {:find_activity, Activity.get_by_id(activity_id, filter: [])},
+         {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(activity)},
          {_, %Object{} = object, _} <-
            {:find_object, Object.normalize(activity, fetch: false), activity},
          true <- User.privileged?(user, :messages_delete) || user.ap_id == object.data["actor"],
@@ -201,6 +213,7 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec repeat(String.t(), User.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
   def repeat(id, user, params \\ %{}) do
     with %Activity{data: %{"type" => "Create"}} = activity <- Activity.get_by_id(id),
          object = %Object{} <- Object.normalize(activity, fetch: false),
@@ -218,11 +231,13 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec unrepeat(String.t(), User.t()) :: {:ok, Activity.t()} | {:error, any()}
   def unrepeat(id, user) do
     with {_, %Activity{data: %{"type" => "Create"}} = activity} <-
            {:find_activity, Activity.get_by_id(id)},
          %Object{} = note <- Object.normalize(activity, fetch: false),
          %Activity{} = announce <- Utils.get_existing_announce(user.ap_id, note),
+         {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(announce)},
          {:ok, undo, _} <- Builder.undo(user, announce),
          {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do
       {:ok, activity}
@@ -232,8 +247,8 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  @spec favorite(User.t(), binary()) :: {:ok, Activity.t() | :already_liked} | {:error, any()}
-  def favorite(%User{} = user, id) do
+  @spec favorite(String.t(), User.t()) :: {:ok, Activity.t()} | {:error, any()}
+  def favorite(id, %User{} = user) do
     case favorite_helper(user, id) do
       {:ok, _} = res ->
         res
@@ -247,7 +262,7 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def favorite_helper(user, id) do
+  defp favorite_helper(user, id) do
     with {_, %Activity{object: object}} <- {:find_object, Activity.get_by_id_with_object(id)},
          {_, {:ok, like_object, meta}} <- {:build_object, Builder.like(user, object)},
          {_, {:ok, %Activity{} = activity, _meta}} <-
@@ -270,11 +285,13 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec unfavorite(String.t(), User.t()) :: {:ok, Activity.t()} | {:error, any()}
   def unfavorite(id, user) do
     with {_, %Activity{data: %{"type" => "Create"}} = activity} <-
            {:find_activity, Activity.get_by_id(id)},
          %Object{} = note <- Object.normalize(activity, fetch: false),
          %Activity{} = like <- Utils.get_existing_like(user.ap_id, note),
+         {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(like)},
          {:ok, undo, _} <- Builder.undo(user, like),
          {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do
       {:ok, activity}
@@ -284,6 +301,8 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec react_with_emoji(String.t(), User.t(), String.t()) ::
+          {:ok, Activity.t()} | {:error, any()}
   def react_with_emoji(id, user, emoji) do
     with %Activity{} = activity <- Activity.get_by_id(id),
          object <- Object.normalize(activity, fetch: false),
@@ -296,8 +315,11 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec unreact_with_emoji(String.t(), User.t(), String.t()) ::
+          {:ok, Activity.t()} | {:error, any()}
   def unreact_with_emoji(id, user, emoji) do
     with %Activity{} = reaction_activity <- Utils.get_latest_reaction(id, user, emoji),
+         {_, {:ok, _}} <- {:cancel_jobs, maybe_cancel_jobs(reaction_activity)},
          {:ok, undo, _} <- Builder.undo(user, reaction_activity),
          {:ok, activity, _} <- Pipeline.common_pipeline(undo, local: true) do
       {:ok, activity}
@@ -307,7 +329,8 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def vote(user, %{data: %{"type" => "Question"}} = object, choices) do
+  @spec vote(Object.t(), User.t(), list()) :: {:ok, list(), Object.t()} | {:error, any()}
+  def vote(%Object{data: %{"type" => "Question"}} = object, %User{} = user, choices) do
     with :ok <- validate_not_author(object, user),
          :ok <- validate_existing_votes(user, object),
          {:ok, options, choices} <- normalize_and_validate_choices(choices, object) do
@@ -368,14 +391,16 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def public_announce?(_, %{visibility: visibility})
-      when visibility in ~w{public unlisted private direct},
-      do: visibility in ~w(public unlisted)
+  defp public_announce?(_, %{visibility: visibility})
+       when visibility in ~w{public unlisted private direct},
+       do: visibility in ~w(public unlisted)
 
-  def public_announce?(object, _) do
+  defp public_announce?(object, _) do
     Visibility.public?(object)
   end
 
+  @spec get_visibility(map(), map() | nil, Participation.t() | nil) ::
+          {String.t() | nil, String.t() | nil}
   def get_visibility(_, _, %Participation{}), do: {"direct", "direct"}
 
   def get_visibility(%{visibility: visibility}, in_reply_to, _)
@@ -394,6 +419,7 @@ defmodule Pleroma.Web.CommonAPI do
 
   def get_visibility(_, in_reply_to, _), do: {"public", get_replied_to_visibility(in_reply_to)}
 
+  @spec get_replied_to_visibility(Activity.t() | nil) :: String.t() | nil
   def get_replied_to_visibility(nil), do: nil
 
   def get_replied_to_visibility(activity) do
@@ -402,6 +428,8 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
+  @spec check_expiry_date({:ok, nil | integer()} | String.t()) ::
+          {:ok, boolean() | nil} | {:error, String.t()}
   def check_expiry_date({:ok, nil} = res), do: res
 
   def check_expiry_date({:ok, in_seconds}) do
@@ -419,19 +447,22 @@ defmodule Pleroma.Web.CommonAPI do
     |> check_expiry_date()
   end
 
+  @spec listen(User.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
   def listen(user, data) do
     with {:ok, draft} <- ActivityDraft.listen(user, data) do
       ActivityPub.listen(draft.changes)
     end
   end
 
+  @spec post(User.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
   def post(user, %{status: _} = data) do
     with {:ok, draft} <- ActivityDraft.create(user, data) do
       ActivityPub.create(draft.changes, draft.preview?)
     end
   end
 
-  def update(user, orig_activity, changes) do
+  @spec update(Activity.t(), User.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
+  def update(orig_activity, %User{} = user, changes) do
     with orig_object <- Object.normalize(orig_activity),
          {:ok, new_object} <- make_update_data(user, orig_object, changes),
          {:ok, update_data, _} <- Builder.update(user, new_object),
@@ -521,7 +552,8 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def add_mute(user, activity, params \\ %{}) do
+  @spec add_mute(Activity.t(), User.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
+  def add_mute(activity, user, params \\ %{}) do
     expires_in = Map.get(params, :expires_in, 0)
 
     with {:ok, _} <- ThreadMute.add_mute(user.id, activity.data["context"]),
@@ -540,15 +572,17 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def remove_mute(%User{} = user, %Activity{} = activity) do
+  @spec remove_mute(Activity.t(), User.t()) :: {:ok, Activity.t()} | {:error, any()}
+  def remove_mute(%Activity{} = activity, %User{} = user) do
     ThreadMute.remove_mute(user.id, activity.data["context"])
     {:ok, activity}
   end
 
-  def remove_mute(user_id, activity_id) do
+  @spec remove_mute(String.t(), String.t()) :: {:ok, Activity.t()} | {:error, any()}
+  def remove_mute(activity_id, user_id) do
     with {:user, %User{} = user} <- {:user, User.get_by_id(user_id)},
          {:activity, %Activity{} = activity} <- {:activity, Activity.get_by_id(activity_id)} do
-      remove_mute(user, activity)
+      remove_mute(activity, user)
     else
       {what, result} = error ->
         Logger.warning(
@@ -559,13 +593,15 @@ defmodule Pleroma.Web.CommonAPI do
     end
   end
 
-  def thread_muted?(%User{id: user_id}, %{data: %{"context" => context}})
+  @spec thread_muted?(Activity.t(), User.t()) :: boolean()
+  def thread_muted?(%{data: %{"context" => context}}, %User{id: user_id})
       when is_binary(context) do
     ThreadMute.exists?(user_id, context)
   end
 
   def thread_muted?(_, _), do: false
 
+  @spec report(User.t(), map()) :: {:ok, Activity.t()} | {:error, any()}
   def report(user, data) do
     with {:ok, account} <- get_reported_account(data.account_id),
          {:ok, {content_html, _, _}} <- make_report_content_html(data[:comment]),
@@ -599,6 +635,8 @@ defmodule Pleroma.Web.CommonAPI do
     |> Enum.filter(&Rule.exists?/1)
   end
 
+  @spec update_report_state(String.t() | [String.t()], String.t()) ::
+          {:ok, any()} | {:error, any()}
   def update_report_state(activity_ids, state) when is_list(activity_ids) do
     case Utils.update_report_state(activity_ids, state) do
       :ok -> {:ok, activity_ids}
@@ -611,17 +649,16 @@ defmodule Pleroma.Web.CommonAPI do
       Utils.update_report_state(activity, state)
     else
       nil -> {:error, :not_found}
-      _ -> {:error, dgettext("errors", "Could not update state")}
     end
   end
 
+  @spec update_activity_scope(String.t(), map()) :: {:ok, any()} | {:error, any()}
   def update_activity_scope(activity_id, opts \\ %{}) do
     with %Activity{} = activity <- Activity.get_by_id_with_object(activity_id),
          {:ok, activity} <- toggle_sensitive(activity, opts) do
       set_visibility(activity, opts)
     else
       nil -> {:error, :not_found}
-      {:error, reason} -> {:error, reason}
     end
   end
 
@@ -649,14 +686,17 @@ defmodule Pleroma.Web.CommonAPI do
 
   defp set_visibility(activity, _), do: {:ok, activity}
 
-  def hide_reblogs(%User{} = user, %User{} = target) do
+  @spec hide_reblogs(User.t(), User.t()) :: {:ok, any()} | {:error, any()}
+  def hide_reblogs(%User{} = target, %User{} = user) do
     UserRelationship.create_reblog_mute(user, target)
   end
 
-  def show_reblogs(%User{} = user, %User{} = target) do
+  @spec show_reblogs(User.t(), User.t()) :: {:ok, any()} | {:error, any()}
+  def show_reblogs(%User{} = target, %User{} = user) do
     UserRelationship.delete_reblog_mute(user, target)
   end
 
+  @spec get_user(String.t(), boolean()) :: User.t() | nil
   def get_user(ap_id, fake_record_fallback \\ true) do
     cond do
       user = User.get_cached_by_ap_id(ap_id) ->
@@ -673,4 +713,14 @@ defmodule Pleroma.Web.CommonAPI do
         nil
     end
   end
+
+  defp maybe_cancel_jobs(%Activity{id: activity_id}) do
+    Oban.Job
+    |> where([j], j.worker == "Pleroma.Workers.PublisherWorker")
+    |> where([j], j.args["op"] == "publish_one")
+    |> where([j], j.args["params"]["activity_id"] == ^activity_id)
+    |> Oban.cancel_all_jobs()
+  end
+
+  defp maybe_cancel_jobs(_), do: {:ok, 0}
 end

lib/pleroma/web/federator.ex

@@ -35,10 +35,12 @@ defmodule Pleroma.Web.Federator do
   end
 
   # Client API
-  def incoming_ap_doc(%{params: params, req_headers: req_headers}) do
+  def incoming_ap_doc(%{params: _params, req_headers: _req_headers} = args) do
+    job_args = Enum.into(args, %{}, fn {k, v} -> {Atom.to_string(k), v} end)
+
     ReceiverWorker.enqueue(
       "incoming_ap_doc",
-      %{"req_headers" => req_headers, "params" => params, "timeout" => :timer.seconds(20)},
+      Map.put(job_args, "timeout", :timer.seconds(20)),
       priority: 2
     )
   end

lib/pleroma/web/mastodon_api/controllers/account_controller.ex

@@ -460,7 +460,7 @@ defmodule Pleroma.Web.MastodonAPI.AccountController do
   end
 
   def unfollow(%{assigns: %{user: follower, account: followed}} = conn, _params) do
-    with {:ok, follower} <- CommonAPI.unfollow(follower, followed) do
+    with {:ok, follower} <- CommonAPI.unfollow(followed, follower) do
       render(conn, "relationship.json", user: follower, target: followed)
     end
   end
@@ -495,7 +495,7 @@ defmodule Pleroma.Web.MastodonAPI.AccountController do
 
   @doc "POST /api/v1/accounts/:id/block"
   def block(%{assigns: %{user: blocker, account: blocked}} = conn, _params) do
-    with {:ok, _activity} <- CommonAPI.block(blocker, blocked) do
+    with {:ok, _activity} <- CommonAPI.block(blocked, blocker) do
       render(conn, "relationship.json", user: blocker, target: blocked)
     else
       {:error, message} -> json_response(conn, :forbidden, %{error: message})
@@ -504,7 +504,7 @@ defmodule Pleroma.Web.MastodonAPI.AccountController do
 
   @doc "POST /api/v1/accounts/:id/unblock"
   def unblock(%{assigns: %{user: blocker, account: blocked}} = conn, _params) do
-    with {:ok, _activity} <- CommonAPI.unblock(blocker, blocked) do
+    with {:ok, _activity} <- CommonAPI.unblock(blocked, blocker) do
       render(conn, "relationship.json", user: blocker, target: blocked)
     else
       {:error, message} -> json_response(conn, :forbidden, %{error: message})

lib/pleroma/web/mastodon_api/controllers/poll_controller.ex

@@ -51,7 +51,7 @@ defmodule Pleroma.Web.MastodonAPI.PollController do
     with %Object{data: %{"type" => "Question"}} = object <- Object.get_by_id(id),
          %Activity{} = activity <- Activity.get_create_by_object_ap_id(object.data["id"]),
          true <- Visibility.visible_for_user?(activity, user),
-         {:ok, _activities, object} <- get_cached_vote_or_vote(user, object, choices) do
+         {:ok, _activities, object} <- get_cached_vote_or_vote(object, user, choices) do
       try_render(conn, "show.json", %{object: object, for: user})
     else
       nil -> render_error(conn, :not_found, "Record not found")
@@ -60,11 +60,11 @@ defmodule Pleroma.Web.MastodonAPI.PollController do
     end
   end
 
-  defp get_cached_vote_or_vote(user, object, choices) do
+  defp get_cached_vote_or_vote(object, user, choices) do
     idempotency_key = "polls:#{user.id}:#{object.data["id"]}"
 
     @cachex.fetch!(:idempotency_cache, idempotency_key, fn _ ->
-      case CommonAPI.vote(user, object, choices) do
+      case CommonAPI.vote(object, user, choices) do
         {:error, _message} = res -> {:ignore, res}
         res -> {:commit, res}
       end

lib/pleroma/web/mastodon_api/controllers/status_controller.ex

@@ -279,7 +279,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusController do
          actor <- Activity.user_actor(activity),
          {_, true} <- {:own_status, actor.id == user.id},
          changes <- body_params |> put_application(conn),
-         {_, {:ok, _update_activity}} <- {:pipeline, CommonAPI.update(user, activity, changes)},
+         {_, {:ok, _update_activity}} <- {:pipeline, CommonAPI.update(activity, user, changes)},
          {_, %Activity{}} = {_, activity} <- {:refetched, Activity.get_by_id_with_object(id)} do
       try_render(conn, "show.json",
         activity: activity,
@@ -360,7 +360,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusController do
           conn,
         _
       ) do
-    with {:ok, _fav} <- CommonAPI.favorite(user, activity_id),
+    with {:ok, _fav} <- CommonAPI.favorite(activity_id, user),
          %Activity{} = activity <- Activity.get_by_id(activity_id) do
       try_render(conn, "show.json", activity: activity, for: user, as: :activity)
     end
@@ -456,7 +456,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusController do
         _
       ) do
     with %Activity{} = activity <- Activity.get_by_id(id),
-         {:ok, activity} <- CommonAPI.add_mute(user, activity, params) do
+         {:ok, activity} <- CommonAPI.add_mute(activity, user, params) do
       try_render(conn, "show.json", activity: activity, for: user, as: :activity)
     end
   end
@@ -470,7 +470,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusController do
         _
       ) do
     with %Activity{} = activity <- Activity.get_by_id(id),
-         {:ok, activity} <- CommonAPI.remove_mute(user, activity) do
+         {:ok, activity} <- CommonAPI.remove_mute(activity, user) do
       try_render(conn, "show.json", activity: activity, for: user, as: :activity)
     end
   end

lib/pleroma/web/mastodon_api/mastodon_api.ex

@@ -16,7 +16,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPI do
   def follow(follower, followed, params \\ %{}) do
     result =
       if not User.following?(follower, followed) do
-        CommonAPI.follow(follower, followed)
+        CommonAPI.follow(followed, follower)
       else
         {:ok, follower, followed, nil}
       end
@@ -30,11 +30,11 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPI do
   end
 
   defp set_reblogs_visibility(false, {:ok, follower, followed, _}) do
-    CommonAPI.hide_reblogs(follower, followed)
+    CommonAPI.hide_reblogs(followed, follower)
   end
 
   defp set_reblogs_visibility(_, {:ok, follower, followed, _}) do
-    CommonAPI.show_reblogs(follower, followed)
+    CommonAPI.show_reblogs(followed, follower)
   end
 
   defp set_subscription(true, {:ok, follower, followed, _}) do

lib/pleroma/web/mastodon_api/views/status_view.ex

@@ -30,7 +30,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do
   # pagination is restricted to 40 activities at a time
   defp fetch_rich_media_for_activities(activities) do
     Enum.each(activities, fn activity ->
-      spawn(fn -> Card.get_by_activity(activity) end)
+      Card.get_by_activity(activity)
     end)
   end
 
@@ -293,7 +293,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do
       cond do
         is_nil(opts[:for]) -> false
         is_boolean(activity.thread_muted?) -> activity.thread_muted?
-        true -> CommonAPI.thread_muted?(opts[:for], activity)
+        true -> CommonAPI.thread_muted?(activity, opts[:for])
       end
 
     attachment_data = object.data["attachment"] || []

lib/pleroma/web/mastodon_api/websocket_handler.ex

@@ -104,7 +104,7 @@ defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
   end
 
   def handle_info(:close, state) do
-    {:stop, {:closed, 'connection closed by server'}, state}
+    {:stop, {:closed, ~c"connection closed by server"}, state}
   end
 
   def handle_info(msg, state) do

lib/pleroma/web/media_proxy.ex

@@ -75,8 +75,7 @@ defmodule Pleroma.Web.MediaProxy do
     %{host: domain} = URI.parse(url)
 
     mediaproxy_whitelist_domains =
-      [:media_proxy, :whitelist]
-      |> Config.get()
+      Config.get([:media_proxy, :whitelist], [])
       |> Kernel.++(["#{Upload.base_url()}"])
       |> Enum.map(&maybe_get_domain_from_url/1)
 

lib/pleroma/web/metadata/utils.ex

@@ -25,11 +25,14 @@ defmodule Pleroma.Web.Metadata.Utils do
     |> scrub_html_and_truncate_object_field(object)
   end
 
-  def scrub_html_and_truncate(%{data: %{"content" => content}} = object) do
+  def scrub_html_and_truncate(%{data: %{"content" => content}} = object)
+      when is_binary(content) and content != "" do
     content
     |> scrub_html_and_truncate_object_field(object)
   end
 
+  def scrub_html_and_truncate(%{}), do: ""
+
   def scrub_html_and_truncate(content, max_length \\ 200, omission \\ "...")
       when is_binary(content) do
     content

lib/pleroma/web/o_auth/app.ex

@@ -62,7 +62,7 @@ defmodule Pleroma.Web.OAuth.App do
     |> Repo.insert()
   end
 
-  @spec update(pos_integer(), map()) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
+  @spec update(pos_integer(), map()) :: {:ok, t()} | {:error, Ecto.Changeset.t()} | nil
   def update(id, params) do
     with %__MODULE__{} = app <- Repo.get(__MODULE__, id) do
       app

lib/pleroma/web/pleroma_api/controllers/backup_controller.ex

@@ -20,7 +20,7 @@ defmodule Pleroma.Web.PleromaAPI.BackupController do
   end
 
   def create(%{assigns: %{user: user}} = conn, _params) do
-    with {:ok, _} <- Backup.create(user) do
+    with {:ok, _} <- Backup.user(user) do
       backups = Backup.list(user)
       render(conn, "index.json", backups: backups)
     end

lib/pleroma/web/pleroma_api/views/backup_view.ex

@@ -9,22 +9,12 @@ defmodule Pleroma.Web.PleromaAPI.BackupView do
   alias Pleroma.Web.CommonAPI.Utils
 
   def render("show.json", %{backup: %Backup{} = backup}) do
-    # To deal with records before the migration
-    state =
-      if backup.state == :invalid do
-        if backup.processed, do: :complete, else: :failed
-      else
-        backup.state
-      end
-
     %{
       id: backup.id,
       content_type: backup.content_type,
       url: download_url(backup),
       file_size: backup.file_size,
       processed: backup.processed,
-      state: to_string(state),
-      processed_number: backup.processed_number,
       inserted_at: Utils.to_masto_date(backup.inserted_at)
     }
   end

lib/pleroma/web/plugs/http_signature_plug.ex

@@ -8,16 +8,12 @@ defmodule Pleroma.Web.Plugs.HTTPSignaturePlug do
   import Plug.Conn
   import Phoenix.Controller, only: [get_format: 1, text: 2]
 
+  alias Pleroma.Signature
   alias Pleroma.Web.ActivityPub.MRF
 
   require Logger
 
   @config_impl Application.compile_env(:pleroma, [__MODULE__, :config_impl], Pleroma.Config)
-  @http_signatures_impl Application.compile_env(
-                          :pleroma,
-                          [__MODULE__, :http_signatures_impl],
-                          HTTPSignatures
-                        )
 
   def init(options) do
     options
@@ -39,48 +35,6 @@ defmodule Pleroma.Web.Plugs.HTTPSignaturePlug do
     end
   end
 
-  defp validate_signature(conn, request_target) do
-    # Newer drafts for HTTP signatures now use @request-target instead of the
-    # old (request-target). We'll now support both for incoming signatures.
-    conn =
-      conn
-      |> put_req_header("(request-target)", request_target)
-      |> put_req_header("@request-target", request_target)
-
-    @http_signatures_impl.validate_conn(conn)
-  end
-
-  defp validate_signature(conn) do
-    # This (request-target) is non-standard, but many implementations do it
-    # this way due to a misinterpretation of
-    # https://datatracker.ietf.org/doc/html/draft-cavage-http-signatures-06
-    # "path" was interpreted as not having the query, though later examples
-    # show that it must be the absolute path + query. This behavior is kept to
-    # make sure most software (Pleroma itself, Mastodon, and probably others)
-    # do not break.
-    request_target = String.downcase("#{conn.method}") <> " #{conn.request_path}"
-
-    # This is the proper way to build the @request-target, as expected by
-    # many HTTP signature libraries, clarified in the following draft:
-    # https://www.ietf.org/archive/id/draft-ietf-httpbis-message-signatures-11.html#section-2.2.6
-    # It is the same as before, but containing the query part as well.
-    proper_target = request_target <> "?#{conn.query_string}"
-
-    cond do
-      # Normal, non-standard behavior but expected by Pleroma and more.
-      validate_signature(conn, request_target) ->
-        true
-
-      # Has query string and the previous one failed: let's try the standard.
-      conn.query_string != "" ->
-        validate_signature(conn, proper_target)
-
-      # If there's no query string and signature fails, it's rotten.
-      true ->
-        false
-    end
-  end
-
   defp maybe_assign_valid_signature(conn) do
     if has_signature_header?(conn) do
       # we replace the digest header with the one we computed in DigestPlug
@@ -90,7 +44,7 @@ defmodule Pleroma.Web.Plugs.HTTPSignaturePlug do
           conn -> conn
         end
 
-      assign(conn, :valid_signature, validate_signature(conn))
+      assign(conn, :valid_signature, Signature.validate_signature(conn))
     else
       Logger.debug("No signature header!")
       conn

lib/pleroma/web/plugs/o_auth_plug.ex

@@ -52,7 +52,7 @@ defmodule Pleroma.Web.Plugs.OAuthPlug do
         where: t.token == ^token
       )
 
-    with %Token{user_id: user_id} = token_record <- Repo.one(token_query),
+    with %Token{user_id: user_id} = token_record <- Repo.one(token_query) |> Repo.preload(:user),
          false <- is_nil(user_id),
          %User{} = user <- User.get_cached_by_id(user_id) do
       {:ok, user, token_record}

lib/pleroma/web/plugs/o_auth_scopes_plug.ex

@@ -34,7 +34,9 @@ defmodule Pleroma.Web.Plugs.OAuthScopesPlug do
         permissions = Enum.join(missing_scopes, " #{op} ")
 
         error_message =
-          dgettext("errors", "Insufficient permissions: %{permissions}.", permissions: permissions)
+          dgettext("errors", "Insufficient permissions: %{permissions}.",
+            permissions: permissions
+          )
 
         conn
         |> put_resp_content_type("application/json")

lib/pleroma/web/rich_media/backfill.ex

@@ -4,75 +4,50 @@
 
 defmodule Pleroma.Web.RichMedia.Backfill do
   alias Pleroma.Web.RichMedia.Card
+  alias Pleroma.Web.RichMedia.Helpers
   alias Pleroma.Web.RichMedia.Parser
   alias Pleroma.Web.RichMedia.Parser.TTL
-  alias Pleroma.Workers.RichMediaExpirationWorker
+  alias Pleroma.Workers.RichMediaWorker
 
   require Logger
 
-  @backfiller Pleroma.Config.get([__MODULE__, :provider], Pleroma.Web.RichMedia.Backfill.Task)
   @cachex Pleroma.Config.get([:cachex, :provider], Cachex)
-  @max_attempts 3
-  @retry 5_000
+  @stream_out_impl Pleroma.Config.get(
+                     [__MODULE__, :stream_out],
+                     Pleroma.Web.ActivityPub.ActivityPub
+                   )
 
-  def start(%{url: url} = args) when is_binary(url) do
+  @spec run(map()) :: :ok | Parser.parse_errors() | Helpers.get_errors()
+  def run(%{"url" => url} = args) do
     url_hash = Card.url_to_hash(url)
 
-    args =
-      args
-      |> Map.put(:attempt, 1)
-      |> Map.put(:url_hash, url_hash)
-
-    @backfiller.run(args)
-  end
-
-  def run(%{url: url, url_hash: url_hash, attempt: attempt} = args)
-      when attempt <= @max_attempts do
     case Parser.parse(url) do
       {:ok, fields} ->
         {:ok, card} = Card.create(url, fields)
 
         maybe_schedule_expiration(url, fields)
 
-        if Map.has_key?(args, :activity_id) do
+        with %{"activity_id" => activity_id} <- args,
+             false <- is_nil(activity_id) do
           stream_update(args)
         end
 
         warm_cache(url_hash, card)
+        :ok
 
-      {:error, {:invalid_metadata, fields}} ->
-        Logger.debug("Rich media incomplete or invalid metadata for #{url}: #{inspect(fields)}")
+      {:error, type} = error
+      when type in [:invalid_metadata, :body_too_large, :content_type, :validate, :get, :head] ->
         negative_cache(url_hash)
-
-      {:error, :body_too_large} ->
-        Logger.error("Rich media error for #{url}: :body_too_large")
-        negative_cache(url_hash)
-
-      {:error, {:content_type, type}} ->
-        Logger.debug("Rich media error for #{url}: :content_type is #{type}")
-        negative_cache(url_hash)
-
-      e ->
-        Logger.debug("Rich media error for #{url}: #{inspect(e)}")
-
-        :timer.sleep(@retry * attempt)
-
-        run(%{args | attempt: attempt + 1})
+        error
     end
   end
 
-  def run(%{url: url, url_hash: url_hash}) do
-    Logger.debug("Rich media failure for #{url}")
-
-    negative_cache(url_hash, :timer.minutes(15))
-  end
-
   defp maybe_schedule_expiration(url, fields) do
     case TTL.process(fields, url) do
       {:ok, ttl} when is_number(ttl) ->
         timestamp = DateTime.from_unix!(ttl)
 
-        RichMediaExpirationWorker.new(%{"url" => url}, scheduled_at: timestamp)
+        RichMediaWorker.new(%{"op" => "expire", "url" => url}, scheduled_at: timestamp)
         |> Oban.insert()
 
       _ ->
@@ -80,22 +55,14 @@ defmodule Pleroma.Web.RichMedia.Backfill do
     end
   end
 
-  defp stream_update(%{activity_id: activity_id}) do
+  defp stream_update(%{"activity_id" => activity_id}) do
     Pleroma.Activity.get_by_id(activity_id)
     |> Pleroma.Activity.normalize()
-    |> Pleroma.Web.ActivityPub.ActivityPub.stream_out()
+    |> @stream_out_impl.stream_out()
   end
 
   defp warm_cache(key, val), do: @cachex.put(:rich_media_cache, key, val)
-  defp negative_cache(key, ttl \\ nil), do: @cachex.put(:rich_media_cache, key, nil, ttl: ttl)
-end
 
-defmodule Pleroma.Web.RichMedia.Backfill.Task do
-  alias Pleroma.Web.RichMedia.Backfill
-
-  def run(args) do
-    Task.Supervisor.start_child(Pleroma.TaskSupervisor, Backfill, :run, [args],
-      name: {:global, {:rich_media, args.url_hash}}
-    )
-  end
+  defp negative_cache(key, ttl \\ :timer.minutes(15)),
+    do: @cachex.put(:rich_media_cache, key, :error, ttl: ttl)
 end

lib/pleroma/web/rich_media/card.ex

@@ -7,8 +7,8 @@ defmodule Pleroma.Web.RichMedia.Card do
   alias Pleroma.HTML
   alias Pleroma.Object
   alias Pleroma.Repo
-  alias Pleroma.Web.RichMedia.Backfill
   alias Pleroma.Web.RichMedia.Parser
+  alias Pleroma.Workers.RichMediaWorker
 
   @cachex Pleroma.Config.get([:cachex, :provider], Cachex)
   @config_impl Application.compile_env(:pleroma, [__MODULE__, :config_impl], Pleroma.Config)
@@ -75,17 +75,18 @@ defmodule Pleroma.Web.RichMedia.Card do
 
   def get_by_url(nil), do: nil
 
-  @spec get_or_backfill_by_url(String.t(), map()) :: t() | nil
-  def get_or_backfill_by_url(url, backfill_opts \\ %{}) do
+  @spec get_or_backfill_by_url(String.t(), keyword()) :: t() | nil
+  def get_or_backfill_by_url(url, opts \\ []) do
     if @config_impl.get([:rich_media, :enabled]) do
       case get_by_url(url) do
         %__MODULE__{} = card ->
           card
 
         nil ->
-          backfill_opts = Map.put(backfill_opts, :url, url)
+          activity_id = Keyword.get(opts, :activity_id, nil)
 
-          Backfill.start(backfill_opts)
+          RichMediaWorker.new(%{"op" => "backfill", "url" => url, "activity_id" => activity_id})
+          |> Oban.insert()
 
           nil
 
@@ -137,7 +138,7 @@ defmodule Pleroma.Web.RichMedia.Card do
       nil
     else
       {:cached, url} ->
-        get_or_backfill_by_url(url, %{activity_id: activity.id})
+        get_or_backfill_by_url(url, activity_id: activity.id)
 
       _ ->
         :error

lib/pleroma/web/rich_media/helpers.ex

@@ -5,26 +5,38 @@
 defmodule Pleroma.Web.RichMedia.Helpers do
   alias Pleroma.Config
 
+  require Logger
+
+  @type get_errors :: {:error, :body_too_large | :content_type | :head | :get}
+
+  @spec rich_media_get(String.t()) :: {:ok, String.t()} | get_errors()
   def rich_media_get(url) do
     headers = [{"user-agent", Pleroma.Application.user_agent() <> "; Bot"}]
 
-    head_check =
-      case Pleroma.HTTP.head(url, headers, http_options()) do
-        # If the HEAD request didn't reach the server for whatever reason,
-        # we assume the GET that comes right after won't either
-        {:error, _} = e ->
-          e
+    with {_, {:ok, %Tesla.Env{status: 200, headers: headers}}} <-
+           {:head, Pleroma.HTTP.head(url, headers, http_options())},
+         {_, :ok} <- {:content_type, check_content_type(headers)},
+         {_, :ok} <- {:content_length, check_content_length(headers)},
+         {_, {:ok, %Tesla.Env{status: 200, body: body}}} <-
+           {:get, Pleroma.HTTP.get(url, headers, http_options())} do
+      {:ok, body}
+    else
+      {:head, _} ->
+        Logger.debug("Rich media error for #{url}: HTTP HEAD failed")
+        {:error, :head}
 
-        {:ok, %Tesla.Env{status: 200, headers: headers}} ->
-          with :ok <- check_content_type(headers),
-               :ok <- check_content_length(headers),
-               do: :ok
+      {:content_type, {_, type}} ->
+        Logger.debug("Rich media error for #{url}: content-type is #{type}")
+        {:error, :content_type}
 
-        _ ->
-          :ok
-      end
+      {:content_length, {_, length}} ->
+        Logger.debug("Rich media error for #{url}: content-length is #{length}")
+        {:error, :body_too_large}
 
-    with :ok <- head_check, do: Pleroma.HTTP.get(url, headers, http_options())
+      {:get, _} ->
+        Logger.debug("Rich media error for #{url}: HTTP GET failed")
+        {:error, :get}
+    end
   end
 
   defp check_content_type(headers) do
@@ -32,7 +44,7 @@ defmodule Pleroma.Web.RichMedia.Helpers do
       {_, content_type} ->
         case Plug.Conn.Utils.media_type(content_type) do
           {:ok, "text", "html", _} -> :ok
-          _ -> {:error, {:content_type, content_type}}
+          _ -> {:error, content_type}
         end
 
       _ ->
@@ -47,7 +59,7 @@ defmodule Pleroma.Web.RichMedia.Helpers do
       {_, maybe_content_length} ->
         case Integer.parse(maybe_content_length) do
           {content_length, ""} when content_length <= max_body -> :ok
-          {_, ""} -> {:error, :body_too_large}
+          {_, ""} -> {:error, maybe_content_length}
           _ -> :ok
         end
 
@@ -57,9 +69,12 @@ defmodule Pleroma.Web.RichMedia.Helpers do
   end
 
   defp http_options do
+    timeout = Config.get!([:rich_media, :timeout])
+
     [
       pool: :rich_media,
-      max_body: Config.get([:rich_media, :max_body], 5_000_000)
+      max_body: Config.get([:rich_media, :max_body], 5_000_000),
+      tesla_middleware: [{Tesla.Middleware.Timeout, timeout: timeout}]
     ]
   end
 end

lib/pleroma/web/rich_media/parser.ex

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.RichMedia.Parser do
+  alias Pleroma.Web.RichMedia.Helpers
   require Logger
 
   @config_impl Application.compile_env(:pleroma, [__MODULE__, :config_impl], Pleroma.Config)
@@ -11,24 +12,26 @@ defmodule Pleroma.Web.RichMedia.Parser do
     Pleroma.Config.get([:rich_media, :parsers])
   end
 
-  def parse(nil), do: nil
+  @type parse_errors :: {:error, :rich_media_disabled | :validate}
 
-  @spec parse(String.t()) :: {:ok, map()} | {:error, any()}
-  def parse(url) do
+  @spec parse(String.t()) ::
+          {:ok, map()} | parse_errors() | Helpers.get_errors()
+  def parse(url) when is_binary(url) do
     with {_, true} <- {:config, @config_impl.get([:rich_media, :enabled])},
-         :ok <- validate_page_url(url),
-         {:ok, data} <- parse_url(url) do
+         {_, :ok} <- {:validate, validate_page_url(url)},
+         {_, {:ok, data}} <- {:parse, parse_url(url)} do
       data = Map.put(data, "url", url)
       {:ok, data}
     else
       {:config, _} -> {:error, :rich_media_disabled}
-      e -> e
+      {:validate, _} -> {:error, :validate}
+      {:parse, error} -> error
     end
   end
 
   defp parse_url(url) do
-    with {:ok, %Tesla.Env{body: html}} <- Pleroma.Web.RichMedia.Helpers.rich_media_get(url),
-         {:ok, html} <- Floki.parse_document(html) do
+    with {:ok, body} <- Helpers.rich_media_get(url),
+         {:ok, html} <- Floki.parse_document(body) do
       html
       |> maybe_parse()
       |> clean_parsed_data()
@@ -50,8 +53,8 @@ defmodule Pleroma.Web.RichMedia.Parser do
     {:ok, data}
   end
 
-  defp check_parsed_data(data) do
-    {:error, {:invalid_metadata, data}}
+  defp check_parsed_data(_data) do
+    {:error, :invalid_metadata}
   end
 
   defp clean_parsed_data(data) do

lib/pleroma/web/rich_media/parsers/o_embed.ex

@@ -22,7 +22,7 @@ defmodule Pleroma.Web.RichMedia.Parsers.OEmbed do
   end
 
   defp get_oembed_data(url) do
-    with {:ok, %Tesla.Env{body: json}} <- Pleroma.Web.RichMedia.Helpers.rich_media_get(url) do
+    with {:ok, json} <- Pleroma.Web.RichMedia.Helpers.rich_media_get(url) do
       Jason.decode(json)
     end
   end

lib/pleroma/web/router.ex

@@ -990,7 +990,7 @@ defmodule Pleroma.Web.Router do
 
   scope "/" do
     pipe_through([:pleroma_html, :authenticate, :require_admin])
-    live_dashboard("/phoenix/live_dashboard")
+    live_dashboard("/phoenix/live_dashboard", additional_pages: [oban: Oban.LiveDashboard])
   end
 
   # Test-only routes needed to test action dispatching and plug chain execution

lib/pleroma/web/streamer.ex

@@ -172,7 +172,13 @@ defmodule Pleroma.Web.Streamer do
   def stream(topics, items) do
     if should_env_send?() do
       for topic <- List.wrap(topics), item <- List.wrap(items) do
-        spawn(fn -> do_stream(topic, item) end)
+        fun = fn -> do_stream(topic, item) end
+
+        if Config.get([__MODULE__, :sync_streaming], false) do
+          fun.()
+        else
+          spawn(fun)
+        end
       end
     end
   end
@@ -200,7 +206,7 @@ defmodule Pleroma.Web.Streamer do
          false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
          false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
          true <- thread_containment(item, user),
-         false <- CommonAPI.thread_muted?(user, parent) do
+         false <- CommonAPI.thread_muted?(parent, user) do
       false
     else
       _ -> true

lib/pleroma/web/templates/feed/feed/tag.atom.eex

@@ -12,7 +12,7 @@
   <subtitle><%= Gettext.dpgettext("static_pages", "tag feed description", "These are public toots tagged with #%{tag}. You can interact with them if you have an account anywhere in the fediverse.", tag: @tag) %></subtitle>
   <logo><%= feed_logo() %></logo>
   <updated><%= most_recent_update(@activities) %></updated>
-  <link rel="self" href="<%= '#{Routes.tag_feed_url(@conn, :feed, @tag)}.atom'  %>" type="application/atom+xml"/>
+  <link rel="self" href="<%= "#{Routes.tag_feed_url(@conn, :feed, @tag)}.atom"  %>" type="application/atom+xml"/>
 
   <%= for activity <- @activities do %>
   <%= render Phoenix.Controller.view_module(@conn), "_tag_activity.atom", Map.merge(assigns, prepare_activity(activity, actor: true)) %>

lib/pleroma/web/templates/feed/feed/tag.rss.eex

@@ -6,7 +6,7 @@
 
     <title>#<%= @tag %></title>
     <description><%= Gettext.dpgettext("static_pages", "tag feed description", "These are public toots tagged with #%{tag}. You can interact with them if you have an account anywhere in the fediverse.", tag: @tag) %></description>
-    <link><%= '#{Routes.tag_feed_url(@conn, :feed, @tag)}.rss' %></link>
+    <link><%= "#{Routes.tag_feed_url(@conn, :feed, @tag)}.rss" %></link>
     <webfeeds:logo><%= feed_logo() %></webfeeds:logo>
     <webfeeds:accentColor>2b90d9</webfeeds:accentColor>
     <%= for activity <- @activities do %>

lib/pleroma/web/templates/feed/feed/user.atom.eex

@@ -11,12 +11,12 @@
   <subtitle><%= escape(@user.bio) %></subtitle>
   <updated><%= most_recent_update(@activities, @user, :atom) %></updated>
   <logo><%= logo(@user) %></logo>
-  <link rel="self" href="<%= '#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.atom' %>" type="application/atom+xml"/>
+  <link rel="self" href="<%= "#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.atom" %>" type="application/atom+xml"/>
 
   <%= render Phoenix.Controller.view_module(@conn), "_author.atom", assigns %>
 
   <%= if last_activity(@activities) do %>
-    <link rel="next" href="<%= '#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.atom?max_id=#{last_activity(@activities).id}' %>" type="application/atom+xml"/>
+    <link rel="next" href="<%= "#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.atom?max_id=#{last_activity(@activities).id}" %>" type="application/atom+xml"/>
   <% end %>
 
   <%= for activity <- @activities do %>

lib/pleroma/web/templates/feed/feed/user.rss.eex

@@ -7,20 +7,20 @@
      xmlns:poco="http://portablecontacts.net/spec/1.0">
   <channel>
     <title><%= @user.nickname <> "'s timeline" %></title>
-    <link><%= '#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.rss' %></link>
+    <link><%= "#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.rss" %></link>
     <atom:link href="<%= Routes.user_feed_url(@conn, :feed, @user.nickname) <> ".atom" %>"
                rel="self" type="application/rss+xml" />
     <description><%= escape(@user.bio) %></description>
     <image>
       <url><%= logo(@user) %></url>
       <title><%= @user.nickname <> "'s timeline" %></title>
-      <link><%= '#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.rss' %></link>
+      <link><%= "#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.rss" %></link>
     </image>
 
     <%= render Phoenix.Controller.view_module(@conn), "_author.rss", assigns %>
 
     <%= if last_activity(@activities) do %>
-      <link rel="next"><%= '#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.rss?max_id=#{last_activity(@activities).id}' %></link>
+      <link rel="next"><%= "#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.rss?max_id=#{last_activity(@activities).id}" %></link>
     <% end %>
 
     <%= for activity <- @activities do %>

lib/pleroma/web/twitter_api/controllers/remote_follow_controller.ex

@@ -73,7 +73,7 @@ defmodule Pleroma.Web.TwitterAPI.RemoteFollowController do
   #
   def do_follow(%{assigns: %{user: %User{} = user}} = conn, %{"user" => %{"id" => id}}) do
     with {:fetch_user, %User{} = followee} <- {:fetch_user, User.get_cached_by_id(id)},
-         {:ok, _, _, _} <- CommonAPI.follow(user, followee) do
+         {:ok, _, _, _} <- CommonAPI.follow(followee, user) do
       redirect(conn, to: "/users/#{followee.id}")
     else
       error ->
@@ -90,7 +90,7 @@ defmodule Pleroma.Web.TwitterAPI.RemoteFollowController do
     with {_, %User{} = followee} <- {:fetch_user, User.get_cached_by_id(id)},
          {_, {:ok, user}, _} <- {:auth, WrapperAuthenticator.get_user(conn), followee},
          {_, _, _, false} <- {:mfa_required, followee, user, MFA.require?(user)},
-         {:ok, _, _, _} <- CommonAPI.follow(user, followee) do
+         {:ok, _, _, _} <- CommonAPI.follow(followee, user) do
       redirect(conn, to: "/users/#{followee.id}")
     else
       error ->
@@ -108,7 +108,7 @@ defmodule Pleroma.Web.TwitterAPI.RemoteFollowController do
          {_, _, {:ok, %{user: user}}} <- {:mfa_token, followee, MFA.Token.validate(token)},
          {_, _, _, {:ok, _}} <-
            {:verify_mfa_code, followee, token, TOTPAuthenticator.verify(code, user)},
-         {:ok, _, _, _} <- CommonAPI.follow(user, followee) do
+         {:ok, _, _, _} <- CommonAPI.follow(followee, user) do
       redirect(conn, to: "/users/#{followee.id}")
     else
       error ->

lib/pleroma/web/twitter_api/controllers/util_controller.ex

@@ -207,9 +207,6 @@ defmodule Pleroma.Web.TwitterAPI.UtilController do
           {:error, changeset} ->
             {_, {error, _}} = Enum.at(changeset.errors, 0)
             json(conn, %{error: "New password #{error}."})
-
-          _ ->
-            json(conn, %{error: "Unable to change password."})
         end
 
       {:error, msg} ->

lib/pleroma/web/xml.ex

@@ -9,7 +9,7 @@ defmodule Pleroma.Web.XML do
 
   def string_from_xpath(xpath, doc) do
     try do
-      {:xmlObj, :string, res} = :xmerl_xpath.string('string(#{xpath})', doc)
+      {:xmlObj, :string, res} = :xmerl_xpath.string(~c"string(#{xpath})", doc)
 
       res =
         res

lib/pleroma/workers/background_worker.ex

@@ -3,7 +3,6 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.BackgroundWorker do
-  alias Pleroma.Instances.Instance
   alias Pleroma.User
 
   use Pleroma.Workers.WorkerHelper, queue: "background"
@@ -15,11 +14,6 @@ defmodule Pleroma.Workers.BackgroundWorker do
     User.perform(:set_activation_async, user, status)
   end
 
-  def perform(%Job{args: %{"op" => "delete_user", "user_id" => user_id}}) do
-    user = User.get_cached_by_id(user_id)
-    User.perform(:delete, user)
-  end
-
   def perform(%Job{args: %{"op" => "force_password_reset", "user_id" => user_id}}) do
     user = User.get_cached_by_id(user_id)
     User.perform(:force_password_reset, user)
@@ -45,10 +39,6 @@ defmodule Pleroma.Workers.BackgroundWorker do
     User.perform(:verify_fields_links, user)
   end
 
-  def perform(%Job{args: %{"op" => "delete_instance", "host" => host}}) do
-    Instance.perform(:delete_instance, host)
-  end
-
   @impl Oban.Worker
-  def timeout(_job), do: :timer.seconds(900)
+  def timeout(_job), do: :timer.seconds(15)
 end

lib/pleroma/workers/backup_worker.ex

@@ -6,64 +6,46 @@ defmodule Pleroma.Workers.BackupWorker do
   use Oban.Worker, queue: :slow, max_attempts: 1
 
   alias Oban.Job
+  alias Pleroma.Config.Getting, as: Config
   alias Pleroma.User.Backup
 
-  def process(backup, admin_user_id \\ nil) do
-    %{"op" => "process", "backup_id" => backup.id, "admin_user_id" => admin_user_id}
-    |> new()
-    |> Oban.insert()
-  end
-
-  def schedule_deletion(backup) do
-    days = Pleroma.Config.get([Backup, :purge_after_days])
-    time = 60 * 60 * 24 * days
-    scheduled_at = Calendar.NaiveDateTime.add!(backup.inserted_at, time)
-
-    %{"op" => "delete", "backup_id" => backup.id}
-    |> new(scheduled_at: scheduled_at)
-    |> Oban.insert()
-  end
-
-  def delete(backup) do
-    %{"op" => "delete", "backup_id" => backup.id}
-    |> new()
-    |> Oban.insert()
-  end
-
   @impl Oban.Worker
   def perform(%Job{
-        args: %{"op" => "process", "backup_id" => backup_id, "admin_user_id" => admin_user_id}
+        args: %{"op" => "process", "backup_id" => backup_id}
       }) do
-    with {:ok, %Backup{} = backup} <-
-           backup_id |> Backup.get() |> Backup.process(),
-         {:ok, _job} <- schedule_deletion(backup),
-         :ok <- Backup.remove_outdated(backup),
-         :ok <- maybe_deliver_email(backup, admin_user_id) do
-      {:ok, backup}
+    with {_, %Backup{} = backup} <- {:get, Backup.get_by_id(backup_id)},
+         {_, {:ok, updated_backup}} <- {:run, Backup.run(backup)},
+         {_, {:ok, uploaded_backup}} <- {:upload, Backup.upload(updated_backup)},
+         {_, {:ok, _job}} <- {:delete, Backup.schedule_delete(uploaded_backup)},
+         {_, :ok} <- {:outdated, Backup.remove_outdated(uploaded_backup.user)},
+         {_, :ok} <- {:email, maybe_deliver_email(uploaded_backup)} do
+      {:ok, uploaded_backup}
+    else
+      e -> {:error, e}
     end
   end
 
   def perform(%Job{args: %{"op" => "delete", "backup_id" => backup_id}}) do
-    case Backup.get(backup_id) do
-      %Backup{} = backup -> Backup.delete(backup)
+    case Backup.get_by_id(backup_id) do
+      %Backup{} = backup -> Backup.delete_archive(backup)
       nil -> :ok
     end
   end
 
   @impl Oban.Worker
-  def timeout(_job), do: :infinity
+  def timeout(_job), do: Config.get([Backup, :timeout], :timer.minutes(30))
 
   defp has_email?(user) do
     not is_nil(user.email) and user.email != ""
   end
 
-  defp maybe_deliver_email(backup, admin_user_id) do
+  defp maybe_deliver_email(backup) do
     has_mailer = Pleroma.Config.get([Pleroma.Emails.Mailer, :enabled])
     backup = backup |> Pleroma.Repo.preload(:user)
 
     if has_email?(backup.user) and has_mailer do
       backup
-      |> Pleroma.Emails.UserEmail.backup_is_ready_email(admin_user_id)
+      |> Pleroma.Emails.UserEmail.backup_is_ready_email()
       |> Pleroma.Emails.Mailer.deliver()
 
       :ok

lib/pleroma/workers/cron/digest_emails_worker.ex

@@ -58,4 +58,7 @@ defmodule Pleroma.Workers.Cron.DigestEmailsWorker do
 
     User.touch_last_digest_emailed_at(user)
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/cron/new_users_digest_worker.ex

@@ -60,4 +60,7 @@ defmodule Pleroma.Workers.Cron.NewUsersDigestWorker do
 
     :ok
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/delete_worker.ex

@@ -0,0 +1,24 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.DeleteWorker do
+  alias Pleroma.Instances.Instance
+  alias Pleroma.User
+
+  use Pleroma.Workers.WorkerHelper, queue: "slow"
+
+  @impl Oban.Worker
+
+  def perform(%Job{args: %{"op" => "delete_user", "user_id" => user_id}}) do
+    user = User.get_cached_by_id(user_id)
+    User.perform(:delete, user)
+  end
+
+  def perform(%Job{args: %{"op" => "delete_instance", "host" => host}}) do
+    Instance.perform(:delete_instance, host)
+  end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(900)
+end

lib/pleroma/workers/mute_expire_worker.ex

@@ -14,7 +14,7 @@ defmodule Pleroma.Workers.MuteExpireWorker do
   def perform(%Job{
         args: %{"op" => "unmute_conversation", "user_id" => user_id, "activity_id" => activity_id}
       }) do
-    Pleroma.Web.CommonAPI.remove_mute(user_id, activity_id)
+    Pleroma.Web.CommonAPI.remove_mute(activity_id, user_id)
     :ok
   end
 

lib/pleroma/workers/poll_worker.ex

@@ -17,6 +17,9 @@ defmodule Pleroma.Workers.PollWorker do
     with %Activity{} = activity <- find_poll_activity(activity_id),
          {:ok, notifications} <- Notification.create_poll_notifications(activity) do
       Notification.stream(notifications)
+    else
+      {:error, :poll_activity_not_found} = e -> {:cancel, e}
+      e -> {:error, e}
     end
   end
 

lib/pleroma/workers/purge_expired_activity.ex

@@ -6,8 +6,8 @@ defmodule Pleroma.Workers.PurgeExpiredActivity do
   @moduledoc """
   Worker which purges expired activity.
   """
-
-  use Oban.Worker, queue: :slow, max_attempts: 1, unique: [period: :infinity]
+  @queue :background
+  use Oban.Worker, queue: @queue, max_attempts: 1, unique: [period: :infinity]
 
   import Ecto.Query
 
@@ -46,20 +46,22 @@ defmodule Pleroma.Workers.PurgeExpiredActivity do
 
   defp find_activity(id) do
     with nil <- Activity.get_by_id_with_object(id) do
-      {:error, :activity_not_found}
+      {:cancel, :activity_not_found}
     end
   end
 
   defp find_user(ap_id) do
     with nil <- Pleroma.User.get_by_ap_id(ap_id) do
-      {:error, :user_not_found}
+      {:cancel, :user_not_found}
     end
   end
 
   def get_expiration(id) do
+    queue = Atom.to_string(@queue)
+
     from(j in Oban.Job,
       where: j.state == "scheduled",
-      where: j.queue == "slow",
+      where: j.queue == ^queue,
       where: fragment("?->>'activity_id' = ?", j.args, ^id)
     )
     |> Pleroma.Repo.one()

lib/pleroma/workers/receiver_worker.ex

@@ -12,17 +12,30 @@ defmodule Pleroma.Workers.ReceiverWorker do
   @impl Oban.Worker
 
   def perform(%Job{
-        args: %{"op" => "incoming_ap_doc", "req_headers" => req_headers, "params" => params}
+        args: %{
+          "op" => "incoming_ap_doc",
+          "method" => method,
+          "params" => params,
+          "req_headers" => req_headers,
+          "request_path" => request_path,
+          "query_string" => query_string
+        }
       }) 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}
+    conn_data = %Plug.Conn{
+      method: method,
+      params: params,
+      req_headers: req_headers,
+      request_path: request_path,
+      query_string: query_string
+    }
 
     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)},
+         {:signature, true} <- {:signature, Signature.validate_signature(conn_data)},
          {:ok, res} <- Federator.perform(:incoming_ap_doc, params) do
       {:ok, res}
     else
@@ -47,11 +60,13 @@ defmodule Pleroma.Workers.ReceiverWorker 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}
+      {:error, {:validate_object, _} = reason} -> {:cancel, reason}
+      {:error, {:error, {:validate, {:error, _changeset} = reason}}} -> {:cancel, reason}
+      {:error, {:reject, _} = reason} -> {:cancel, reason}
       {:signature, false} -> {:cancel, :invalid_signature}
-      {:error, {:error, reason = "Object has been deleted"}} -> {:cancel, reason}
+      {:error, "Object has been deleted"} = reason -> {:cancel, reason}
+      {:error, {:side_effects, {:error, :no_object_actor}} = reason} -> {:cancel, reason}
+      {:error, :not_found} = reason -> {:cancel, reason}
       {:error, _} = e -> e
       e -> {:error, e}
     end

lib/pleroma/workers/remote_fetcher_worker.ex

@@ -13,23 +13,23 @@ defmodule Pleroma.Workers.RemoteFetcherWorker do
       {:ok, _object} ->
         :ok
 
+      {:reject, reason} ->
+        {:cancel, reason}
+
       {:error, :forbidden} ->
-        {:discard, :forbidden}
+        {:cancel, :forbidden}
 
       {:error, :not_found} ->
-        {:discard, :not_found}
+        {:cancel, :not_found}
 
       {:error, :allowed_depth} ->
-        {:discard, :allowed_depth}
+        {:cancel, :allowed_depth}
 
       {:error, _} = e ->
         e
-
-      e ->
-        {:error, e}
     end
   end
 
   @impl Oban.Worker
-  def timeout(_job), do: :timer.seconds(10)
+  def timeout(_job), do: :timer.seconds(15)
 end

lib/pleroma/workers/rich_media_expiration_worker.ex

@@ -1,15 +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.RichMediaExpirationWorker do
-  alias Pleroma.Web.RichMedia.Card
-
-  use Oban.Worker,
-    queue: :background
-
-  @impl Oban.Worker
-  def perform(%Job{args: %{"url" => url} = _args}) do
-    Card.delete(url)
-  end
-end

lib/pleroma/workers/rich_media_worker.ex

@@ -0,0 +1,40 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.RichMediaWorker do
+  alias Pleroma.Config
+  alias Pleroma.Web.RichMedia.Backfill
+  alias Pleroma.Web.RichMedia.Card
+
+  use Oban.Worker, queue: :background, max_attempts: 3, unique: [period: 300]
+
+  @impl Oban.Worker
+  def perform(%Job{args: %{"op" => "expire", "url" => url} = _args}) do
+    Card.delete(url)
+  end
+
+  def perform(%Job{args: %{"op" => "backfill", "url" => _url} = args}) do
+    case Backfill.run(args) do
+      :ok ->
+        :ok
+
+      {:error, type}
+      when type in [:invalid_metadata, :body_too_large, :content_type, :validate, :get, :head] ->
+        {:cancel, type}
+
+      error ->
+        {:error, error}
+    end
+  end
+
+  # There is timeout value enforced by Tesla.Middleware.Timeout
+  # which can be found in the RichMedia.Helpers module to allow us to detect
+  # a slow/infinite data stream and insert a negative cache entry for the URL
+  # We pad it by 2 seconds to be certain a slow connection is detected and we
+  # can inject a negative cache entry for the URL
+  @impl Oban.Worker
+  def timeout(_job) do
+    Config.get!([:rich_media, :timeout]) + :timer.seconds(2)
+  end
+end

lib/pleroma/workers/search_indexing_worker.ex

@@ -20,4 +20,7 @@ defmodule Pleroma.Workers.SearchIndexingWorker do
 
     search_module.remove_from_index(object)
   end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(5)
 end

lib/pleroma/workers/user_refresh_worker.ex

@@ -0,0 +1,17 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Workers.UserRefreshWorker do
+  use Oban.Worker, queue: :background, max_attempts: 1, unique: [period: 300]
+
+  alias Pleroma.User
+
+  @impl true
+  def perform(%Job{args: %{"ap_id" => ap_id}}) do
+    User.fetch_by_ap_id(ap_id)
+  end
+
+  @impl Oban.Worker
+  def timeout(_job), do: :timer.seconds(15)
+end