Merge remote-tracking branch 'origin/develop' into audit-mix-audit-report

Conflict resolution notes:

- mix.lock: kept this branch's versions for cowlib, h2, hackney and hpax.
  Those upgrades are the point of the branch; develop is still on hackney
  1.20.1.

- .woodpecker/docker-armv7.yaml: accepted develop's deletion. It only
  existed here to build mdex_native on armv7, and mdex was already
  replaced with earmark again.

- lib/pleroma/reverse_proxy.ex: took develop's version. It independently
  grew request_with_constraints/5 and the invalid-status close, which
  supersede the equivalent fixes on this branch. Re-applied this branch's
  recv_timeout seeding, which develop does not have.

- lib/pleroma/gun/connection_pool/worker.ex and
  lib/pleroma/reverse_proxy/client/tesla.ex: took develop's versions. The
  gun pool rework (#7934) already guards the demonitor and makes
  release_conn/cancel_stream exit-safe, and it splits the Tesla client's
  graceful release from its abort path, so this branch's changes there are
  obsolete.

That leaves the client leak fix as a Hackney-only concern, which
stream_body/1 now handles the same way the gun client does.

lib/mix/tasks/pleroma/config.ex

@@ -252,40 +252,43 @@ defmodule Mix.Tasks.Pleroma.Config do
     if whitelisted_configs in [nil, false] do
       shell_error("No unwanted settings in ConfigDB. No changes made.")
     else
-      whitelisted_groups =
-        whitelisted_configs
-        |> Enum.filter(fn
-          {_group} -> true
-          _ -> false
-        end)
-        |> Enum.map(fn {group} -> group end)
-
-      whitelisted_keys =
-        whitelisted_configs
-        |> Enum.filter(fn
-          {_group, _key} -> true
-          _ -> false
-        end)
+      {whitelisted_groups, whitelisted_keys} = get_filters(whitelisted_configs)
 
       filtered =
         from(c in ConfigDB)
         |> Repo.all()
         |> Enum.filter(&not_whitelisted?(&1, whitelisted_groups, whitelisted_keys))
 
-      if not Enum.empty?(filtered) do
-        shell_info("The following settings will be removed from ConfigDB:\n")
-        Enum.each(filtered, &dump(&1))
+      do_config_filter(filtered, force)
+    end
+  end
 
-        if force or shell_prompt("Are you sure you want to continue?", "n") in ~w(Yn Y y) do
-          filtered_ids = Enum.map(filtered, fn %{id: id} -> id end)
+  # Removes blacklisted configuration sections
+  def run(["filter_blacklisted" | rest]) do
+    {options, [], []} =
+      OptionParser.parse(
+        rest,
+        strict: [force: :boolean],
+        aliases: [f: :force]
+      )
 
-          Repo.delete_all(from(c in ConfigDB, where: c.id in ^filtered_ids))
-        else
-          shell_error("No changes made.")
-        end
-      else
-        shell_error("No unwanted settings in ConfigDB. No changes made.")
-      end
+    force = Keyword.get(options, :force, false)
+
+    start_pleroma()
+
+    blacklisted_configs = Pleroma.Config.get(:database_config_blacklist)
+
+    if blacklisted_configs in [nil, false] do
+      shell_error("No unwanted settings in ConfigDB. No changes made.")
+    else
+      {blacklisted_groups, blacklisted_keys} = get_filters(blacklisted_configs)
+
+      filtered =
+        from(c in ConfigDB)
+        |> Repo.all()
+        |> Enum.filter(&blacklisted?(&1, blacklisted_groups, blacklisted_keys))
+
+      do_config_filter(filtered, force)
     end
   end
 
@@ -490,8 +493,49 @@ defmodule Mix.Tasks.Pleroma.Config do
     Ecto.Adapters.SQL.query!(Repo, "ALTER SEQUENCE config_id_seq RESTART;")
   end
 
+  defp get_filters(config) do
+    groups =
+      config
+      |> Enum.filter(fn
+        {_group} -> true
+        _ -> false
+      end)
+      |> Enum.map(fn {group} -> group end)
+
+    keys =
+      config
+      |> Enum.filter(fn
+        {_group, _key} -> true
+        _ -> false
+      end)
+
+    {groups, keys}
+  end
+
   defp not_whitelisted?(%{group: group, key: key}, whitelisted_groups, whitelisted_keys) do
     not Enum.member?(whitelisted_groups, group) and
       not Enum.member?(whitelisted_keys, {group, key})
   end
+
+  defp blacklisted?(%{group: group, key: key}, blacklisted_groups, blacklisted_keys) do
+    Enum.member?(blacklisted_groups, group) or
+      Enum.member?(blacklisted_keys, {group, key})
+  end
+
+  defp do_config_filter(filtered, _) when length(filtered) == 0 do
+    shell_error("No unwanted settings in ConfigDB. No changes made.")
+  end
+
+  defp do_config_filter(filtered, force) do
+    shell_info("The following settings will be removed from ConfigDB:\n")
+    Enum.each(filtered, &dump(&1))
+
+    if force or shell_prompt("Are you sure you want to continue?", "n") in ~w(Yn Y y) do
+      filtered_ids = Enum.map(filtered, fn %{id: id} -> id end)
+
+      Repo.delete_all(from(c in ConfigDB, where: c.id in ^filtered_ids))
+    else
+      shell_error("No changes made.")
+    end
+  end
 end

lib/pleroma/application.ex

@@ -43,6 +43,7 @@ defmodule Pleroma.Application do
     # every time the application is restarted, so we disable module
     # conflicts at runtime
     Code.compiler_options(ignore_module_conflict: true)
+    configure_logger()
     Pleroma.Telemetry.Logger.attach()
     Config.Holder.save_default()
     Pleroma.HTML.compile_scrubbers()
@@ -113,6 +114,44 @@ defmodule Pleroma.Application do
     Supervisor.start_link(children, opts)
   end
 
+  def configure_logger do
+    Config.get([:logger, :backends], [])
+    |> Enum.each(fn backend ->
+      case backend_to_logger(backend) do
+        {:ok, logger} ->
+          add_logger(logger)
+
+        {:error, :console_backend} ->
+          Logger.warning(":console is no longer considered a backend and is enabled by default")
+      end
+    end)
+  end
+
+  defp add_logger(backend) do
+    case LoggerBackends.add(backend) do
+      {:ok, _} ->
+        Logger.debug("Successfully added logger backend: #{inspect(backend)}")
+
+      {:error, :already_present} ->
+        :ok
+
+      {:error, reason} ->
+        Logger.error("Failed to add logger backend #{inspect(backend)}: #{inspect(reason)}")
+    end
+  end
+
+  defp backend_to_logger(:console), do: {:error, :console_backend}
+
+  defp backend_to_logger({:ex_syslogger = backend, name}) do
+    Logger.warning(
+      "Configuration {:#{backend}, :#{name}} is incorrect. Use {ExSyslogger, :#{name}} instead!"
+    )
+
+    {:ok, {ExSyslogger, name}}
+  end
+
+  defp backend_to_logger(backend), do: {:ok, backend}
+
   def load_custom_modules do
     dir = Config.get([:modules, :runtime_dir])
 

lib/pleroma/chat.ex

@@ -108,4 +108,6 @@ defmodule Pleroma.Chat do
     |> cast(%{pinned: false}, [:pinned])
     |> Repo.update()
   end
+
+  def enabled?, do: Pleroma.Config.get([__MODULE__, :enabled], true)
 end

lib/pleroma/config/deprecation_warnings.ex

@@ -20,6 +20,8 @@ defmodule Pleroma.Config.DeprecationWarnings do
      "\n* `config :pleroma, :instance, mrf_transparency_exclusions` is now `config :pleroma, :mrf, transparency_exclusions`"}
   ]
 
+  @logger_formatter_config_knobs [:colors, :format, :metadata, :truncate, :utc_log]
+
   def check_exiftool_filter do
     filters = Config.get([Pleroma.Upload]) |> Keyword.get(:filters, [])
 
@@ -218,7 +220,8 @@ defmodule Pleroma.Config.DeprecationWarnings do
       check_quarantined_instances_tuples(),
       check_transparency_exclusions_tuples(),
       check_simple_policy_tuples(),
-      check_exiftool_filter()
+      check_exiftool_filter(),
+      check_deprecated_logger_config()
     ]
     |> Enum.reduce(:ok, fn
       :ok, :ok -> :ok
@@ -415,4 +418,72 @@ defmodule Pleroma.Config.DeprecationWarnings do
       :ok
     end
   end
+
+  defp merge_deprecated_logger_config(config) do
+    formatter_config = Enum.filter(config, fn {k, _} -> k in @logger_formatter_config_knobs end)
+    formatter = Logger.default_formatter(formatter_config)
+    log_level = Keyword.get(config, :level, nil)
+
+    Logger.debug("""
+    Reconfiguring console Logger with deprecated configuration syntax.
+      Handler configuration:
+        log_level: #{inspect(log_level)}
+
+      Formatter:
+        #{inspect(formatter)}
+    """)
+
+    if Config.get(:env) != :test do
+      if log_level, do: :logger.update_handler_config(:default, :level, log_level)
+      :logger.set_handler_config(:default, :formatter, formatter)
+    end
+  end
+
+  @spec check_deprecated_logger_config() :: :ok | :error
+  def check_deprecated_logger_config do
+    backends_config = Application.get_env(:logger, :backends)
+    console_config = Application.get_env(:logger, :console)
+
+    # NOTE: No need to merge the old backends config since it still works.
+    # And new configuration will just add new Logger backends.
+    backend =
+      if backends_config do
+        Logger.warning("""
+        !!!DEPRECATION WARNING!!!
+        Your configuration is using deprecated syntax for configuring backends of Elixir's logger.
+        `config :logger, backends: [...]` is deprecated syntax due to changes in Elixir.
+        Please update your configuration at your earliest convenience to use:
+          `config :pleroma, :logger, backends: [...]`
+
+        Note: `:console` is no longer considered a backend and is used by default, you can disable it using:
+          `config :logger, :default_handler: false`
+        """)
+
+        true
+      else
+        false
+      end
+
+    console =
+      if console_config do
+        Logger.warning("""
+        !!!DEPRECATION WARNING!!!
+        Your configuration is using deprecated syntax for configuring logging to console.
+        `config :logger, :console` is deprecated syntax due to changes in Elixir.
+        Please update your configuration at your earliest convenience to use
+          `config :logger, default_handler` and `config :logger, :default_formatter`.
+
+        Note: `:default_handler` is used only for the `level` setting. All other configurations go under
+        `:default_formatter`. For more info visit: https://hexdocs.pm/logger/Logger.html#module-backends-and-backwards-compatibility
+        """)
+
+        merge_deprecated_logger_config(console_config)
+
+        true
+      else
+        false
+      end
+
+    if backend or console, do: :error, else: :ok
+  end
 end

lib/pleroma/config/transfer_task.ex

@@ -51,7 +51,7 @@ defmodule Pleroma.Config.TransferTask do
 
       started_applications = Application.started_applications()
 
-      reject = [nil, :postgrex]
+      reject = [nil, :postgrex, :logger]
 
       reject =
         if restart_pleroma? do

lib/pleroma/docs/generator.ex

@@ -12,13 +12,27 @@ defmodule Pleroma.Docs.Generator do
 
   @spec list_behaviour_implementations(behaviour :: module()) :: [module()]
   def list_behaviour_implementations(behaviour) do
-    :code.all_loaded()
-    |> Enum.filter(fn {module, _} ->
+    # :code.all_loaded() no longer works on Elixir 1.19 and newer since modules are lazily loaded.
+    # We can get a complete list of modules by interating over Application.loaded_applications/0,
+    # but that includes every module which would get loaded below, which is very slow.
+    # Hardcode the Pleroma application instead, which is good enough for now.
+    application_modules =
+      case :application.get_key(:pleroma, :modules) do
+        {:ok, modules} -> modules
+        :undefined -> []
+      end
+
+    # At compile time or in tests some modules will be outside of the Pleroma application.
+    loaded_modules = Enum.map(:code.all_loaded(), fn {module, _} -> module end)
+
+    (application_modules ++ loaded_modules)
+    |> Enum.uniq()
+    |> Enum.filter(fn module ->
+      Code.ensure_loaded(module)
+
       # This shouldn't be needed as all modules are expected to have module_info/1,
       # but in test environments some transient modules `:elixir_compiler_XX`
       # are loaded for some reason (where XX is a random integer).
-      Code.ensure_loaded(module)
-
       if function_exported?(module, :module_info, 1) do
         module.module_info(:attributes)
         |> Keyword.get_values(:behaviour)
@@ -26,7 +40,6 @@ defmodule Pleroma.Docs.Generator do
         |> Enum.member?(behaviour)
       end
     end)
-    |> Enum.map(fn {module, _} -> module end)
   end
 
   @doc """

lib/pleroma/gun.ex

@@ -6,9 +6,12 @@ defmodule Pleroma.Gun do
   @callback open(charlist(), pos_integer(), map()) :: {:ok, pid()}
   @callback info(pid()) :: map()
   @callback close(pid()) :: :ok
+  @callback cancel(pid(), reference() | [reference()]) :: :ok
   @callback await_up(pid, pos_integer()) :: {:ok, atom()} | {:error, atom()}
   @callback connect(pid(), map()) :: reference()
-  @callback await(pid(), reference()) :: {:response, :fin, 200, []}
+  @callback await(pid(), reference()) :: tuple()
+  @callback await_tunnel_up(pid(), reference() | :undefined, timeout()) ::
+              {:ok, atom()} | {:error, term()}
   @callback set_owner(pid(), pid()) :: :ok
 
   defp api, do: Pleroma.Config.get([Pleroma.Gun], Pleroma.Gun.API)
@@ -19,11 +22,15 @@ defmodule Pleroma.Gun do
 
   def close(pid), do: api().close(pid)
 
+  def cancel(pid, stream), do: api().cancel(pid, stream)
+
   def await_up(pid, timeout \\ 5_000), do: api().await_up(pid, timeout)
 
   def connect(pid, opts), do: api().connect(pid, opts)
 
   def await(pid, ref), do: api().await(pid, ref)
 
+  def await_tunnel_up(pid, ref, timeout), do: api().await_tunnel_up(pid, ref, timeout)
+
   def set_owner(pid, owner), do: api().set_owner(pid, owner)
 end

lib/pleroma/gun/api.ex

@@ -32,6 +32,9 @@ defmodule Pleroma.Gun.API do
   @impl Gun
   defdelegate close(pid), to: :gun
 
+  @impl Gun
+  defdelegate cancel(pid, stream), to: :gun
+
   @impl Gun
   defdelegate await_up(pid, timeout \\ 5_000), to: :gun
 
@@ -41,6 +44,28 @@ defmodule Pleroma.Gun.API do
   @impl Gun
   defdelegate await(pid, ref), to: :gun
 
+  @impl Gun
+  def await_tunnel_up(pid, stream_ref, timeout) do
+    monitor_ref = Process.monitor(pid)
+
+    result =
+      receive do
+        {:gun_tunnel_up, ^pid, ^stream_ref, protocol} ->
+          {:ok, protocol}
+
+        {:gun_down, ^pid, _protocol, reason, _killed_streams} ->
+          {:error, {:down, reason}}
+
+        {:DOWN, ^monitor_ref, :process, ^pid, reason} ->
+          {:error, {:down, reason}}
+      after
+        timeout -> {:error, :timeout}
+      end
+
+    Process.demonitor(monitor_ref, [:flush])
+    result
+  end
+
   @impl Gun
   defdelegate set_owner(pid, owner), to: :gun
 end

lib/pleroma/gun/conn.ex

@@ -42,18 +42,34 @@ defmodule Pleroma.Gun.Conn do
     Map.put(opts, :tls_opts, tls_opts)
   end
 
+  defp do_open(%URI{scheme: "http"} = uri, %{proxy: {proxy_host, proxy_port}} = opts) do
+    with open_opts <- opts |> proxy_open_opts() |> Map.put(:protocols, [:http]),
+         {:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
+         {:ok, protocol} <- await_up(conn, opts[:connect_timeout]) do
+      {:ok, conn, protocol, :forward_proxy}
+    else
+      error ->
+        Logger.warning(
+          "Opening forward proxy connection to #{compose_uri_log(uri)} failed with error #{inspect(error)}"
+        )
+
+        error
+    end
+  end
+
   defp do_open(uri, %{proxy: {proxy_host, proxy_port}} = opts) do
     connect_opts =
       uri
       |> destination_opts()
       |> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
+      |> add_proxy_auth(opts)
 
-    with open_opts <- Map.delete(opts, :tls_opts),
+    with open_opts <- proxy_open_opts(opts),
          {:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
-         {:ok, protocol} <- Gun.await_up(conn, opts[:connect_timeout]),
-         stream <- Gun.connect(conn, connect_opts),
-         {:response, :fin, 200, _} <- Gun.await(conn, stream) do
-      {:ok, conn, protocol}
+         {:ok, _proxy_protocol} <- await_up(conn, opts[:connect_timeout]) do
+      conn
+      |> Gun.connect(connect_opts)
+      |> await_proxy_connect(conn, opts[:connect_timeout])
     else
       error ->
         Logger.warning(
@@ -64,30 +80,29 @@ defmodule Pleroma.Gun.Conn do
     end
   end
 
-  defp do_open(uri, %{proxy: {proxy_type, proxy_host, proxy_port}} = opts) do
-    version =
-      proxy_type
-      |> to_string()
-      |> String.last()
-      |> case do
-        "4" -> 4
-        _ -> 5
-      end
+  defp do_open(_uri, %{proxy: {:socks4, _proxy_host, _proxy_port}}) do
+    {:error, :socks4_unsupported}
+  end
 
+  defp do_open(uri, %{proxy: {proxy_type, proxy_host, proxy_port}} = opts)
+       when proxy_type in [:socks, :socks5] do
     socks_opts =
       uri
       |> destination_opts()
       |> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
-      |> Map.put(:version, version)
+      |> Map.put(:version, 5)
+      |> add_socks_proxy_auth(opts)
 
-    opts =
+    open_opts =
       opts
+      |> proxy_open_opts()
       |> Map.put(:protocols, [:socks])
       |> Map.put(:socks_opts, socks_opts)
 
-    with {:ok, conn} <- Gun.open(proxy_host, proxy_port, opts),
-         {:ok, protocol} <- Gun.await_up(conn, opts[:connect_timeout]) do
-      {:ok, conn, protocol}
+    with {:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
+         {:ok, :socks} <- await_up(conn, opts[:connect_timeout]),
+         {:ok, protocol} <- await_tunnel_up(conn, :undefined, opts[:connect_timeout]) do
+      {:ok, conn, protocol, nil}
     else
       error ->
         Logger.warning(
@@ -100,10 +115,11 @@ defmodule Pleroma.Gun.Conn do
 
   defp do_open(%URI{host: host, port: port} = uri, opts) do
     host = Pleroma.HTTP.AdapterHelper.parse_host(host)
+    opts = Map.put(opts, :transport, transport(uri.scheme))
 
     with {:ok, conn} <- Gun.open(host, port, opts),
-         {:ok, protocol} <- Gun.await_up(conn, opts[:connect_timeout]) do
-      {:ok, conn, protocol}
+         {:ok, protocol} <- await_up(conn, opts[:connect_timeout]) do
+      {:ok, conn, protocol, nil}
     else
       error ->
         Logger.warning(
@@ -120,11 +136,90 @@ defmodule Pleroma.Gun.Conn do
   end
 
   defp add_http2_opts(opts, "https", tls_opts) do
-    Map.merge(opts, %{protocols: [:http2], transport: :tls, tls_opts: tls_opts})
+    Map.merge(opts, %{protocols: [:http2, :http], transport: :tls, tls_opts: tls_opts})
   end
 
   defp add_http2_opts(opts, _, _), do: opts
 
+  defp add_proxy_auth(connect_opts, %{proxy_auth: {username, password}})
+       when is_binary(username) and is_binary(password) do
+    Map.merge(connect_opts, %{username: username, password: password})
+  end
+
+  defp add_proxy_auth(connect_opts, _), do: connect_opts
+
+  defp add_socks_proxy_auth(socks_opts, %{proxy_auth: {username, password}})
+       when is_binary(username) and is_binary(password) do
+    Map.put(socks_opts, :auth, [{:username_password, username, password}])
+  end
+
+  defp add_socks_proxy_auth(socks_opts, _), do: socks_opts
+
+  defp proxy_open_opts(opts) do
+    proxy_tls_opts = Map.get(opts, :proxy_tls_opts, [])
+
+    opts =
+      opts
+      |> Map.drop([:proxy, :proxy_auth, :proxy_tls_opts, :tls_opts])
+      |> Map.put_new(:transport, :tcp)
+
+    if opts.transport == :tls do
+      Map.put(opts, :tls_opts, proxy_tls_opts)
+    else
+      opts
+    end
+  end
+
+  defp await_up(conn, timeout) do
+    case Gun.await_up(conn, timeout) do
+      {:ok, protocol} ->
+        {:ok, protocol}
+
+      error ->
+        Gun.close(conn)
+        error
+    end
+  end
+
+  defp await_proxy_connect(stream, conn, timeout) do
+    case Gun.await(conn, stream) do
+      {:response, :fin, 200, _headers} ->
+        case await_tunnel_up(conn, stream, timeout) do
+          {:ok, protocol} -> {:ok, conn, protocol, stream}
+          error -> error
+        end
+
+      {:response, _fin, 407, _headers} ->
+        close_with_error(conn, :proxy_auth_failed)
+
+      {:response, _fin, status, _headers} when is_integer(status) ->
+        close_with_error(conn, {:proxy_connect_failed, status})
+
+      error ->
+        Gun.close(conn)
+        error
+    end
+  end
+
+  defp close_with_error(conn, reason) do
+    Gun.close(conn)
+    {:error, reason}
+  end
+
+  defp transport("https"), do: :tls
+  defp transport(_), do: :tcp
+
+  defp await_tunnel_up(conn, stream_ref, timeout) do
+    case Gun.await_tunnel_up(conn, stream_ref, timeout) do
+      {:ok, protocol} ->
+        {:ok, protocol}
+
+      error ->
+        Gun.close(conn)
+        error
+    end
+  end
+
   def compose_uri_log(%URI{scheme: scheme, host: host, path: path}) do
     "#{scheme}://#{host}#{path}"
   end

lib/pleroma/gun/connection_pool.ex

@@ -4,53 +4,132 @@
 
 defmodule Pleroma.Gun.ConnectionPool do
   @registry __MODULE__
+  @locks Module.concat(__MODULE__, Locks)
 
   alias Pleroma.Gun.ConnectionPool.WorkerSupervisor
 
+  @connection_option_keys [
+    :http_opts,
+    :http2_opts,
+    :protocols,
+    :proxy,
+    :proxy_auth,
+    :proxy_tls_opts,
+    :retry,
+    :retry_timeout,
+    :socks_opts,
+    :tcp_opts,
+    :tls_opts,
+    :transport,
+    :ws_opts
+  ]
+
   def children do
     [
-      {Registry, keys: :unique, name: @registry},
+      {Registry, keys: :duplicate, name: @registry},
+      {Registry, keys: :unique, name: @locks},
       Pleroma.Gun.ConnectionPool.WorkerSupervisor
     ]
   end
 
   @spec get_conn(URI.t(), keyword()) :: {:ok, pid()} | {:error, term()}
   def get_conn(uri, opts) do
-    key = "#{uri.scheme}:#{uri.host}:#{uri.port}"
+    key = connection_key(uri, opts)
 
-    case Registry.lookup(@registry, key) do
-      # The key has already been registered, but connection is not up yet
-      [{worker_pid, nil}] ->
-        get_gun_pid_from_worker(worker_pid, true)
-
-      [{worker_pid, {gun_pid, _used_by, _crf, _last_reference}}] ->
-        GenServer.call(worker_pid, :add_client)
+    case checkout_existing(Registry.lookup(@registry, key)) do
+      {:ok, gun_pid} ->
         {:ok, gun_pid}
 
-      [] ->
-        # :gun.set_owner fails in :connected state for whatevever reason,
-        # so we open the connection in the process directly and send it's pid back
-        # We trust gun to handle timeouts by itself
-        case WorkerSupervisor.start_worker([key, uri, opts, self()]) do
-          {:ok, worker_pid} ->
-            get_gun_pid_from_worker(worker_pid, false)
+      :not_available ->
+        provision_or_wait(key, uri, opts)
+    end
+  end
 
-          {:error, {:already_started, worker_pid}} ->
-            get_gun_pid_from_worker(worker_pid, true)
+  defp checkout_existing([]), do: :not_available
 
-          err ->
-            err
+  defp checkout_existing([{_worker_pid, nil} | workers]), do: checkout_existing(workers)
+
+  defp checkout_existing([{worker_pid, _value} | workers]) do
+    case checkout_worker(worker_pid) do
+      {:ok, gun_pid} -> {:ok, gun_pid}
+      _ -> checkout_existing(workers)
+    end
+  end
+
+  defp provision_or_wait(key, uri, opts) do
+    result = with_route_lock(key, fn -> provision(key, uri, opts) end)
+
+    case result do
+      {:ok, gun_pid} ->
+        {:ok, gun_pid}
+
+      {:new, worker_pid} ->
+        get_gun_pid_from_worker(worker_pid)
+
+      {:wait, worker_pid} ->
+        case checkout_worker(worker_pid, :infinity) do
+          {:ok, gun_pid} -> {:ok, gun_pid}
+          :busy -> get_conn(uri, opts)
+          error -> error
+        end
+
+      error ->
+        error
+    end
+  end
+
+  defp provision(key, uri, opts) do
+    workers = Registry.lookup(@registry, key)
+
+    case checkout_existing(workers) do
+      {:ok, gun_pid} ->
+        {:ok, gun_pid}
+
+      :not_available ->
+        case Enum.find(workers, fn {_worker_pid, value} -> is_nil(value) end) do
+          {worker_pid, nil} -> {:wait, worker_pid}
+          nil -> start_worker(key, uri, opts)
         end
     end
   end
 
-  defp get_gun_pid_from_worker(worker_pid, register) do
+  defp with_route_lock(key, function) do
+    case Registry.register(@locks, key, nil) do
+      {:ok, _} ->
+        try do
+          function.()
+        after
+          Registry.unregister(@locks, key)
+        end
+
+      {:error, {:already_registered, _owner}} ->
+        Process.sleep(1)
+        with_route_lock(key, function)
+    end
+  end
+
+  defp start_worker(key, uri, opts) do
+    # :gun.set_owner fails in :connected state, so the worker opens the connection itself.
+    case WorkerSupervisor.start_worker([key, uri, opts, self()]) do
+      {:ok, worker_pid} -> {:new, worker_pid}
+      error -> error
+    end
+  end
+
+  defp checkout_worker(worker_pid, timeout \\ 5_000) do
+    try do
+      GenServer.call(worker_pid, :checkout, timeout)
+    catch
+      :exit, reason -> {:error, {:connection_worker_exit, reason}}
+    end
+  end
+
+  defp get_gun_pid_from_worker(worker_pid) do
     # GenServer.call will block the process for timeout length if
     # the server crashes on startup (which will happen if gun fails to connect)
     # so instead we use cast + monitor
 
     ref = Process.monitor(worker_pid)
-    if register, do: GenServer.cast(worker_pid, {:add_client, self()})
 
     receive do
       {:conn_pid, pid} ->
@@ -68,19 +147,115 @@ defmodule Pleroma.Gun.ConnectionPool do
 
   @spec release_conn(pid()) :: :ok
   def release_conn(conn_pid) do
-    # :ets.fun2ms(fn {_, {worker_pid, {gun_pid, _, _, _}}} when gun_pid == conn_pid ->
+    # :ets.fun2ms(fn {_, {worker_pid, {gun_pid, _tunnel}}} when gun_pid == conn_pid ->
     #    worker_pid end)
     query_result =
       Registry.select(@registry, [
-        {{:_, :"$1", {:"$2", :_, :_, :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
+        {{:_, :"$1", {:"$2", :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
       ])
 
     case query_result do
       [worker_pid] ->
-        GenServer.call(worker_pid, :remove_client)
+        try do
+          GenServer.call(worker_pid, :remove_client)
+        catch
+          :exit, _ -> :ok
+        end
 
       [] ->
         :ok
     end
   end
+
+  @spec discard_conn(pid()) :: :ok
+  def discard_conn(conn_pid) do
+    case worker_for_conn(conn_pid) do
+      nil ->
+        :ok
+
+      worker_pid ->
+        DynamicSupervisor.terminate_child(WorkerSupervisor, worker_pid)
+    end
+  end
+
+  @spec cancel_stream(pid(), reference() | [reference()]) :: :ok
+  def cancel_stream(conn_pid, stream) do
+    case worker_for_conn(conn_pid) do
+      nil ->
+        Pleroma.Gun.cancel(conn_pid, stream)
+
+      worker_pid ->
+        try do
+          GenServer.call(worker_pid, {:cancel_stream, stream})
+        catch
+          :exit, _ -> :ok
+        end
+    end
+  end
+
+  @spec register_stream(pid(), reference() | [reference()]) :: :ok
+  def register_stream(conn_pid, stream) do
+    call_worker(conn_pid, {:register_stream, stream})
+  end
+
+  @spec finish_stream(pid(), reference() | [reference()]) :: :ok
+  def finish_stream(conn_pid, stream) do
+    call_worker(conn_pid, {:finish_stream, stream})
+  end
+
+  @spec release_stream(pid(), reference() | [reference()]) :: :ok
+  def release_stream(conn_pid, stream) do
+    call_worker(conn_pid, {:release_stream, stream})
+  end
+
+  @spec tunnel_ref(pid()) :: reference() | :forward_proxy | nil
+  def tunnel_ref(conn_pid) do
+    case Registry.select(@registry, [
+           {{:_, :_, {:"$1", :"$2"}}, [{:==, :"$1", conn_pid}], [:"$2"]}
+         ]) do
+      [tunnel] -> tunnel
+      [] -> nil
+    end
+  end
+
+  defp worker_for_conn(conn_pid) do
+    case Registry.select(@registry, [
+           {{:_, :"$1", {:"$2", :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
+         ]) do
+      [worker_pid] -> worker_pid
+      [] -> nil
+    end
+  end
+
+  defp call_worker(conn_pid, message) do
+    case worker_for_conn(conn_pid) do
+      nil ->
+        :ok
+
+      worker_pid ->
+        try do
+          GenServer.call(worker_pid, message)
+        catch
+          :exit, _ -> :ok
+        end
+    end
+  end
+
+  defp connection_key(uri, opts) do
+    origin = {
+      String.downcase(uri.scheme),
+      String.downcase(uri.host),
+      uri.port
+    }
+
+    profile =
+      opts
+      |> Map.new()
+      |> Map.take(@connection_option_keys)
+      |> :erlang.term_to_binary()
+      |> then(&:crypto.hash(:sha256, &1))
+
+    fingerprint = Base.encode16(profile, case: :lower)
+    Enum.join(Tuple.to_list(origin) ++ [fingerprint], ":")
+  end
 end

lib/pleroma/gun/connection_pool/reclaimer.ex

@@ -9,11 +9,11 @@ defmodule Pleroma.Gun.ConnectionPool.Reclaimer do
 
   def start_monitor do
     pid =
-      case GenServer.start(__MODULE__, [], name: {:via, Registry, {registry(), "reclaimer"}}) do
+      case GenServer.start(__MODULE__, [], name: __MODULE__) do
         {:ok, pid} ->
           pid
 
-        {:error, {:already_registered, pid}} ->
+        {:error, {:already_started, pid}} ->
           pid
       end
 
@@ -41,17 +41,16 @@ defmodule Pleroma.Gun.ConnectionPool.Reclaimer do
       reclaim_max: reclaim_max
     })
 
-    # :ets.fun2ms(
-    # fn {_, {worker_pid, {_, used_by, crf, last_reference}}} when used_by == [] ->
-    #   {worker_pid, crf, last_reference} end)
-    unused_conns =
+    workers =
       Registry.select(
         registry(),
         [
-          {{:_, :"$1", {:_, :"$2", :"$3", :"$4"}}, [{:==, :"$2", []}], [{{:"$1", :"$3", :"$4"}}]}
+          {{:_, :"$1", {:_, :_}}, [], [:"$1"]}
         ]
       )
 
+    unused_conns = Enum.flat_map(workers, &reclaim_info/1)
+
     case unused_conns do
       [] ->
         :telemetry.execute(
@@ -65,25 +64,43 @@ defmodule Pleroma.Gun.ConnectionPool.Reclaimer do
         {:stop, :no_unused_conns, nil}
 
       unused_conns ->
-        reclaimed =
+        reclaimed_count =
           unused_conns
-          |> Enum.sort(fn {_pid1, crf1, last_reference1}, {_pid2, crf2, last_reference2} ->
-            crf1 <= crf2 and last_reference1 <= last_reference2
-          end)
-          |> Enum.take(reclaim_max)
+          |> Enum.sort_by(fn {_pid, crf, last_reference} -> {crf, last_reference} end)
+          |> Enum.reduce_while(0, fn
+            _worker, count when count == reclaim_max ->
+              {:halt, count}
 
-        reclaimed
-        |> Enum.each(fn {pid, _, _} ->
-          DynamicSupervisor.terminate_child(Pleroma.Gun.ConnectionPool.WorkerSupervisor, pid)
-        end)
+            {worker_pid, _crf, _last_reference}, count ->
+              if reclaim(worker_pid), do: {:cont, count + 1}, else: {:cont, count}
+          end)
 
         :telemetry.execute(
           [:pleroma, :connection_pool, :reclaim, :stop],
-          %{reclaimed_count: Enum.count(reclaimed)},
+          %{reclaimed_count: reclaimed_count},
           %{max_connections: max_connections}
         )
 
         {:stop, :normal, nil}
     end
   end
+
+  defp reclaim_info(worker_pid) do
+    try do
+      case GenServer.call(worker_pid, :reclaim_info) do
+        {:idle, crf, last_reference} -> [{worker_pid, crf, last_reference}]
+        :in_use -> []
+      end
+    catch
+      :exit, _ -> []
+    end
+  end
+
+  defp reclaim(worker_pid) do
+    try do
+      GenServer.call(worker_pid, :reclaim) == :reclaimed
+    catch
+      :exit, _ -> false
+    end
+  end
 end

lib/pleroma/gun/connection_pool/worker.ex

@@ -8,25 +8,22 @@ defmodule Pleroma.Gun.ConnectionPool.Worker do
 
   defp registry, do: Pleroma.Gun.ConnectionPool
 
-  def start_link([key | _] = opts) do
-    GenServer.start_link(__MODULE__, opts, name: {:via, Registry, {registry(), key}})
-  end
+  def start_link(opts), do: GenServer.start_link(__MODULE__, opts)
 
   @impl true
-  def init([_key, _uri, _opts, _client_pid] = opts) do
+  def init([key, _uri, _opts, _client_pid] = opts) do
+    {:ok, _} = Registry.register(registry(), key, nil)
     {:ok, nil, {:continue, {:connect, opts}}}
   end
 
   @impl true
   def handle_continue({:connect, [key, uri, opts, client_pid]}, _) do
-    with {:ok, conn_pid, protocol} <- Gun.Conn.open(uri, opts),
+    with {:ok, conn_pid, protocol, tunnel} <- Gun.Conn.open(uri, opts),
          Process.link(conn_pid) do
       time = :erlang.monotonic_time(:millisecond)
 
-      {_, _} =
-        Registry.update_value(registry(), key, fn _ ->
-          {conn_pid, [client_pid], 1, time}
-        end)
+      Registry.unregister(registry(), key)
+      {:ok, _} = Registry.register(registry(), key, {conn_pid, tunnel})
 
       send(client_pid, {:conn_pid, conn_pid})
 
@@ -34,8 +31,14 @@ defmodule Pleroma.Gun.ConnectionPool.Worker do
        %{
          key: key,
          timer: nil,
-         client_monitors: %{client_pid => Process.monitor(client_pid)},
-         protocol: protocol
+         clients: %{
+           client_pid => %{count: 1, monitor: Process.monitor(client_pid), streams: MapSet.new()}
+         },
+         conn_pid: conn_pid,
+         protocol: protocol,
+         proxied: Keyword.has_key?(opts, :proxy),
+         crf: 1,
+         last_reference: time
        }, :hibernate}
     else
       err ->
@@ -44,108 +47,225 @@ defmodule Pleroma.Gun.ConnectionPool.Worker do
   end
 
   @impl true
-  def handle_cast({:add_client, client_pid}, state) do
-    case handle_call(:add_client, {client_pid, nil}, state) do
-      {:reply, conn_pid, state, :hibernate} ->
-        send(client_pid, {:conn_pid, conn_pid})
-        {:noreply, state, :hibernate}
-    end
+  def handle_call(:checkout, _from, %{protocol: protocol, clients: clients} = state)
+      when protocol != :http2 and map_size(clients) > 0 do
+    {:reply, :busy, state, :hibernate}
   end
 
-  @impl true
-  def handle_cast({:remove_client, client_pid}, state) do
-    case handle_call(:remove_client, {client_pid, nil}, state) do
-      {:reply, _, state, :hibernate} ->
-        {:noreply, state, :hibernate}
-    end
-  end
-
-  @impl true
-  def handle_call(:add_client, {client_pid, _}, %{key: key, protocol: protocol} = state) do
+  def handle_call(:checkout, {client_pid, _}, %{key: key, protocol: protocol} = state) do
     time = :erlang.monotonic_time(:millisecond)
 
-    {{conn_pid, used_by, _, _}, _} =
-      Registry.update_value(registry(), key, fn {conn_pid, used_by, crf, last_reference} ->
-        {conn_pid, [client_pid | used_by], crf(time - last_reference, crf), time}
-      end)
+    [{_, {conn_pid, _tunnel}}] =
+      Registry.lookup(registry(), key) |> Enum.filter(&(elem(&1, 0) == self()))
+
+    state =
+      state
+      |> cancel_idle_timer()
+      |> add_client(client_pid)
+      |> Map.put(:crf, crf(time - state.last_reference, state.crf))
+      |> Map.put(:last_reference, time)
 
     :telemetry.execute(
       [:pleroma, :connection_pool, :client, :add],
-      %{client_pid: client_pid, clients: used_by},
+      %{client_pid: client_pid, clients: Map.keys(state.clients)},
       %{key: state.key, protocol: protocol}
     )
 
-    state =
-      if state.timer != nil do
-        Process.cancel_timer(state[:timer])
-        %{state | timer: nil}
-      else
-        state
-      end
-
-    ref = Process.monitor(client_pid)
-
-    state = put_in(state.client_monitors[client_pid], ref)
-    {:reply, conn_pid, state, :hibernate}
+    {:reply, {:ok, conn_pid}, state, :hibernate}
   end
 
   @impl true
-  def handle_call(:remove_client, {client_pid, _}, %{key: key} = state) do
-    {{_conn_pid, used_by, _crf, _last_reference}, _} =
-      Registry.update_value(registry(), key, fn {conn_pid, used_by, crf, last_reference} ->
-        {conn_pid, List.delete(used_by, client_pid), crf, last_reference}
-      end)
+  def handle_call(:remove_client, {client_pid, _}, state) do
+    case active_streams(state, client_pid) do
+      [] ->
+        {:reply, :ok, remove_client(state, client_pid, true), :hibernate}
 
-    {ref, state} = pop_in(state.client_monitors[client_pid])
+      streams when state.protocol == :http2 ->
+        Enum.each(streams, &(:ok = Gun.cancel(state.conn_pid, &1)))
+        {:reply, :ok, remove_client(state, client_pid, true), :hibernate}
 
-    # A client that is already gone has no monitor left. Tolerate it rather than
-    # crashing the worker, which would take the shared connection down with it.
-    if ref, do: Process.demonitor(ref, [:flush])
+      _streams ->
+        {:stop, :normal, :ok, state}
+    end
+  end
 
-    timer =
-      if used_by == [] do
-        max_idle = Pleroma.Config.get([:connections_pool, :max_idle_time], 30_000)
-        Process.send_after(self(), :idle_close, max_idle)
-      else
-        nil
-      end
+  def handle_call({:register_stream, stream}, {client_pid, _}, state) do
+    {:reply, :ok, update_stream(state, client_pid, &MapSet.put(&1, stream)), :hibernate}
+  end
 
-    {:reply, :ok, %{state | timer: timer}, :hibernate}
+  def handle_call({:finish_stream, stream}, {client_pid, _}, state) do
+    {:reply, :ok, update_stream(state, client_pid, &MapSet.delete(&1, stream)), :hibernate}
+  end
+
+  def handle_call({:release_stream, stream}, {client_pid, _}, state) do
+    state = update_stream(state, client_pid, &MapSet.delete(&1, stream))
+    {:reply, :ok, remove_client(state, client_pid, true), :hibernate}
+  end
+
+  def handle_call({:cancel_stream, stream}, {client_pid, _}, %{protocol: :http2} = state) do
+    :ok = Gun.cancel(state.conn_pid, stream)
+    state = update_stream(state, client_pid, &MapSet.delete(&1, stream))
+    {:reply, :ok, remove_client(state, client_pid, true), :hibernate}
+  end
+
+  def handle_call({:cancel_stream, _stream}, _from, state) do
+    {:stop, :normal, :ok, state}
+  end
+
+  def handle_call(:reclaim_info, _from, %{clients: clients} = state)
+      when map_size(clients) == 0 do
+    {:reply, {:idle, state.crf, state.last_reference}, state, :hibernate}
+  end
+
+  def handle_call(:reclaim_info, _from, state) do
+    {:reply, :in_use, state, :hibernate}
+  end
+
+  def handle_call(:reclaim, _from, %{clients: clients} = state) when map_size(clients) == 0 do
+    {:stop, :normal, :reclaimed, state}
+  end
+
+  def handle_call(:reclaim, _from, state) do
+    {:reply, :in_use, state, :hibernate}
   end
 
   @impl true
-  def handle_info(:idle_close, state) do
-    # Gun monitors the owner process, and will close the connection automatically
-    # when it's terminated
+  def handle_info({:idle_close, token}, %{timer: {_timer_ref, token}, clients: clients} = state)
+      when map_size(clients) == 0 do
+    # Gun monitors the owner process, and will close the connection automatically.
     {:stop, :normal, state}
   end
 
+  def handle_info({:idle_close, _token}, state), do: {:noreply, state, :hibernate}
+
   @impl true
-  def handle_info({:gun_up, _pid, _protocol}, state) do
-    {:noreply, state, :hibernate}
+  def handle_info({:gun_up, _pid, protocol}, state) do
+    {:noreply, %{state | protocol: protocol}, :hibernate}
   end
 
-  # Gracefully shutdown if the connection got closed without any streams left
+  @impl true
+  def handle_info({:gun_error, conn_pid, stream, _reason}, %{conn_pid: conn_pid} = state) do
+    {:noreply, delete_stream(state, stream), :hibernate}
+  end
+
+  @impl true
+  def handle_info({:gun_down, _pid, _protocol, _reason, _streams}, %{proxied: true} = state) do
+    {:stop, :normal, state}
+  end
+
+  # Gracefully shutdown if the connection got closed without any streams left.
   @impl true
   def handle_info({:gun_down, _pid, _protocol, _reason, []}, state) do
     {:stop, :normal, state}
   end
 
-  # Otherwise, wait for retry
+  # Otherwise, wait for retry.
   @impl true
   def handle_info({:gun_down, _pid, _protocol, _reason, _killed_streams}, state) do
     {:noreply, state, :hibernate}
   end
 
   @impl true
-  def handle_info({:DOWN, _ref, :process, pid, reason}, state) do
-    :telemetry.execute(
-      [:pleroma, :connection_pool, :client, :dead],
-      %{client_pid: pid, reason: reason},
-      %{key: state.key}
-    )
+  def handle_info({:DOWN, ref, :process, pid, reason}, state) do
+    case state.clients[pid] do
+      %{monitor: ^ref} ->
+        :telemetry.execute(
+          [:pleroma, :connection_pool, :client, :dead],
+          %{client_pid: pid, reason: reason},
+          %{key: state.key}
+        )
 
-    handle_cast({:remove_client, pid}, state)
+        handle_dead_client(state, pid)
+
+      _ ->
+        {:noreply, state, :hibernate}
+    end
+  end
+
+  defp add_client(state, client_pid) do
+    update_in(state.clients, fn clients ->
+      case clients[client_pid] do
+        nil ->
+          Map.put(clients, client_pid, %{
+            count: 1,
+            monitor: Process.monitor(client_pid),
+            streams: MapSet.new()
+          })
+
+        client ->
+          Map.put(clients, client_pid, %{client | count: client.count + 1})
+      end
+    end)
+  end
+
+  defp handle_dead_client(state, client_pid) do
+    streams = active_streams(state, client_pid)
+
+    cond do
+      streams == [] ->
+        {:noreply, remove_client(state, client_pid, false, true), :hibernate}
+
+      state.protocol == :http2 ->
+        Enum.each(streams, &(:ok = Gun.cancel(state.conn_pid, &1)))
+        {:noreply, remove_client(state, client_pid, false, true), :hibernate}
+
+      true ->
+        {:stop, :normal, state}
+    end
+  end
+
+  defp active_streams(state, client_pid) do
+    case state.clients[client_pid] do
+      %{streams: streams} -> MapSet.to_list(streams)
+      nil -> []
+    end
+  end
+
+  defp update_stream(state, client_pid, update) do
+    update_in(state.clients, fn clients ->
+      case clients[client_pid] do
+        nil -> clients
+        client -> Map.put(clients, client_pid, %{client | streams: update.(client.streams)})
+      end
+    end)
+  end
+
+  defp delete_stream(state, stream) do
+    update_in(state.clients, fn clients ->
+      Map.new(clients, fn {pid, client} ->
+        {pid, %{client | streams: MapSet.delete(client.streams, stream)}}
+      end)
+    end)
+  end
+
+  defp remove_client(state, client_pid, demonitor, all \\ false) do
+    case state.clients[client_pid] do
+      nil ->
+        state
+
+      %{count: count} = client when count > 1 and not all ->
+        put_in(state.clients[client_pid], %{client | count: count - 1})
+
+      client ->
+        if demonitor, do: Process.demonitor(client.monitor, [:flush])
+        state = update_in(state.clients, &Map.delete(&1, client_pid))
+
+        if map_size(state.clients) == 0, do: schedule_idle_close(state), else: state
+    end
+  end
+
+  defp cancel_idle_timer(%{timer: nil} = state), do: state
+
+  defp cancel_idle_timer(%{timer: {timer_ref, _token}} = state) do
+    Process.cancel_timer(timer_ref)
+    %{state | timer: nil}
+  end
+
+  defp schedule_idle_close(state) do
+    max_idle = Pleroma.Config.get([:connections_pool, :max_idle_time], 30_000)
+    token = make_ref()
+    timer_ref = Process.send_after(self(), {:idle_close, token}, max_idle)
+    %{state | timer: {timer_ref, token}}
   end
 
   # LRFU policy: https://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.55.1478

lib/pleroma/gun/connection_pool/worker_supervisor.ex

@@ -26,7 +26,14 @@ defmodule Pleroma.Gun.ConnectionPool.WorkerSupervisor do
   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})
+        [key | _] = opts
+
+        :telemetry.execute(
+          [:pleroma, :connection_pool, :provision_failure],
+          %{},
+          %{key: key}
+        )
+
         {:error, :pool_full}
 
       res ->

lib/pleroma/http/adapter/gun.ex

@@ -0,0 +1,253 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.HTTP.Adapter.Gun do
+  @behaviour Tesla.Adapter
+
+  alias Pleroma.Gun.ConnectionPool
+  alias Tesla.Multipart
+
+  @default_timeout 1_000
+
+  @impl Tesla.Adapter
+  def call(env, opts) do
+    adapter_opts =
+      Tesla.Adapter.opts(
+        [close_conn: true, body_as: :plain, send_body: :at_once, receive: true],
+        env,
+        opts
+      )
+      |> Map.new()
+
+    with {:ok, status, headers, body} <- request(env, adapter_opts) do
+      env = %{env | status: status, headers: format_headers(headers), body: body}
+      {:ok, put_stream_ref(env, body)}
+    end
+  end
+
+  defp put_stream_ref(env, %{stream: stream}) do
+    put_in(env.opts[:adapter][:stream], stream)
+  end
+
+  defp put_stream_ref(env, _body), do: env
+
+  defp request(env, %{conn: conn, tunnel: tunnel} = opts) do
+    uri = URI.parse(Tesla.build_url(env))
+    path = request_path(uri, tunnel)
+    method = Tesla.Adapter.Shared.format_method(env.method)
+    {headers, body, send_body} = prepare_body(env.headers, env.body, opts[:send_body])
+    headers = maybe_add_forward_proxy_headers(headers, uri, opts, tunnel)
+    request_opts = request_opts(opts[:reply_to], tunnel)
+
+    stream =
+      try do
+        open_stream(conn, method, path, headers, body, request_opts, send_body)
+      catch
+        kind, reason ->
+          ConnectionPool.release_conn(conn)
+          :erlang.raise(kind, reason, __STACKTRACE__)
+      end
+
+    :ok = ConnectionPool.register_stream(conn, stream)
+    :ok = send_stream_body(conn, stream, body, send_body)
+    response = read_response(conn, stream, opts)
+
+    case response do
+      {:error, reason} ->
+        ConnectionPool.cancel_stream(conn, stream)
+        {:error, {:gun_stream_error, reason}}
+
+      response ->
+        unless opts[:body_as] in [:chunks, :stream] and stream_body?(response) do
+          :ok = ConnectionPool.finish_stream(conn, stream)
+        end
+
+        if opts[:close_conn] and opts[:body_as] not in [:stream, :chunks] do
+          :gun.close(conn)
+        end
+
+        response
+    end
+  end
+
+  defp stream_body?({:ok, _status, _headers, %{stream: _stream}}), do: true
+  defp stream_body?({:ok, _status, _headers, %Stream{}}), do: true
+  defp stream_body?(_response), do: false
+
+  defp request_opts(reply_to, nil), do: %{reply_to: reply_to || self()}
+
+  defp request_opts(reply_to, :forward_proxy), do: %{reply_to: reply_to || self()}
+
+  defp request_opts(reply_to, tunnel) do
+    %{reply_to: reply_to || self(), tunnel: tunnel}
+  end
+
+  defp request_path(uri, :forward_proxy) do
+    path = Tesla.Adapter.Shared.prepare_path(uri.path, uri.query)
+    "#{uri.scheme}://#{authority(uri)}#{path}"
+  end
+
+  defp request_path(uri, _tunnel) do
+    Tesla.Adapter.Shared.prepare_path(uri.path, uri.query)
+  end
+
+  defp maybe_add_forward_proxy_headers(headers, uri, opts, :forward_proxy) do
+    headers
+    |> put_header_new("host", authority(uri))
+    |> maybe_put_proxy_authorization(opts)
+  end
+
+  defp maybe_add_forward_proxy_headers(headers, _uri, _opts, _tunnel), do: headers
+
+  defp maybe_put_proxy_authorization(headers, %{proxy_auth: {username, password}})
+       when is_binary(username) and is_binary(password) do
+    value = "Basic " <> Base.encode64(username <> ":" <> password)
+    List.keystore(headers, "proxy-authorization", 0, {"proxy-authorization", value})
+  end
+
+  defp maybe_put_proxy_authorization(headers, _opts), do: headers
+
+  defp put_header_new(headers, name, value) do
+    if List.keymember?(headers, name, 0), do: headers, else: [{name, value} | headers]
+  end
+
+  defp authority(%URI{scheme: scheme, host: host, port: port}) do
+    host = if String.contains?(host, ":"), do: "[#{host}]", else: host
+    if port == URI.default_port(scheme), do: host, else: "#{host}:#{port}"
+  end
+
+  defp prepare_body(headers, %Multipart{} = multipart, _send_body) do
+    {format_headers(headers ++ Multipart.headers(multipart)), Multipart.body(multipart), :stream}
+  end
+
+  defp prepare_body(headers, body, _send_body)
+       when is_struct(body, Stream) or is_function(body) do
+    {format_headers(headers), body, :stream}
+  end
+
+  defp prepare_body(headers, body, send_body) do
+    {format_headers(headers), body || "", send_body}
+  end
+
+  defp open_stream(conn, method, path, headers, _body, request_opts, :stream) do
+    :gun.headers(conn, method, path, headers, request_opts)
+  end
+
+  defp open_stream(conn, method, path, headers, body, request_opts, :at_once) do
+    :gun.request(conn, method, path, headers, body, request_opts)
+  end
+
+  defp send_stream_body(conn, stream, body, :stream) do
+    try do
+      for data <- body, do: :ok = :gun.data(conn, stream, :nofin, data)
+      :gun.data(conn, stream, :fin, "")
+    catch
+      kind, reason ->
+        ConnectionPool.cancel_stream(conn, stream)
+        :erlang.raise(kind, reason, __STACKTRACE__)
+    end
+  end
+
+  defp send_stream_body(_conn, _stream, _body, :at_once), do: :ok
+
+  defp read_response(conn, stream, opts) do
+    receive? = opts[:receive]
+
+    receive do
+      {:gun_response, ^conn, ^stream, :fin, status, headers} ->
+        {:ok, status, headers, ""}
+
+      {:gun_response, ^conn, ^stream, :nofin, status, headers} ->
+        format_response(conn, stream, opts, status, headers, opts[:body_as])
+
+      {:gun_up, ^conn, _protocol} when receive? ->
+        read_response(conn, stream, opts)
+
+      {:gun_error, ^conn, reason} ->
+        {:error, reason}
+
+      {:gun_error, ^conn, ^stream, reason} ->
+        {:error, reason}
+
+      {:gun_down, ^conn, _protocol, _reason, _killed_streams} when receive? ->
+        read_response(conn, stream, opts)
+
+      {:DOWN, _ref, :process, ^conn, reason} ->
+        {:error, reason}
+    after
+      opts[:timeout] || @default_timeout -> {:error, :recv_response_timeout}
+    end
+  end
+
+  defp format_response(conn, stream, opts, status, headers, :plain) do
+    case read_body(conn, stream, opts) do
+      {:ok, body} ->
+        {:ok, status, headers, body}
+
+      {:error, error} ->
+        :ok = :gun.flush(stream)
+        {:error, error}
+    end
+  end
+
+  defp format_response(conn, stream, opts, status, headers, :stream) do
+    body =
+      Stream.resource(
+        fn -> :reading end,
+        fn
+          :reading ->
+            case Tesla.Adapter.Gun.read_chunk(conn, stream, opts) do
+              {:nofin, part} -> {[part], :reading}
+              {:fin, part} -> {[part], :done}
+              {:error, reason} -> raise "Gun stream failed: #{inspect(reason)}"
+            end
+
+          :done ->
+            {:halt, :done}
+        end,
+        fn _state ->
+          if opts[:close_conn], do: :gun.close(conn)
+        end
+      )
+
+    {:ok, status, headers, body}
+  end
+
+  defp format_response(conn, stream, opts, status, headers, :chunks) do
+    {:ok, status, headers, %{pid: conn, stream: stream, opts: Map.to_list(opts)}}
+  end
+
+  defp read_body(conn, stream, opts, acc \\ "") do
+    receive do
+      {:gun_data, ^conn, ^stream, :fin, body} ->
+        append_body(acc, body, opts[:max_body])
+
+      {:gun_data, ^conn, ^stream, :nofin, body} ->
+        with {:ok, acc} <- append_body(acc, body, opts[:max_body]) do
+          read_body(conn, stream, opts, acc)
+        end
+
+      {:gun_error, ^conn, ^stream, reason} ->
+        {:error, reason}
+
+      {:DOWN, _ref, :process, ^conn, reason} ->
+        {:error, reason}
+    after
+      opts[:timeout] || @default_timeout -> {:error, :recv_body_timeout}
+    end
+  end
+
+  defp append_body(acc, part, nil), do: {:ok, acc <> part}
+
+  defp append_body(acc, part, limit) do
+    body = acc <> part
+    if byte_size(body) <= limit, do: {:ok, body}, else: {:error, :body_too_large}
+  end
+
+  defp format_headers(headers) do
+    Enum.map(headers, fn {key, value} ->
+      {String.downcase(to_string(key)), to_string(value)}
+    end)
+  end
+end

lib/pleroma/http/adapter_helper/gun.ex

@@ -48,10 +48,10 @@ defmodule Pleroma.HTTP.AdapterHelper.Gun do
     Keyword.put(opts, :timeout, recv_timeout)
   end
 
-  # Gun uses [body_as: :stream]
+  # Keep the pool lease until the caller consumes or cancels the body.
   defp maybe_stream(opts) do
     case Keyword.pop(opts, :stream, nil) do
-      {true, opts} -> Keyword.put(opts, :body_as, :stream)
+      {true, opts} -> Keyword.put(opts, :body_as, :chunks)
       {_, opts} -> opts
     end
   end

lib/pleroma/reverse_proxy.ex

@@ -17,6 +17,7 @@ defmodule Pleroma.ReverseProxy do
   @max_read_duration :timer.seconds(30)
   @max_body_length :infinity
   @failed_request_ttl :timer.seconds(60)
+  @sniff_bytes 8 * 1024
   @methods ~w(GET HEAD)
 
   @allowed_mime_types Pleroma.Config.get([Pleroma.Upload, :allowed_mime_types], [])
@@ -59,7 +60,12 @@ defmodule Pleroma.ReverseProxy do
   * `inline_content_types`:
     * `true` will not alter `content-disposition` (up to the upstream),
     * `false` will add `content-disposition: attachment` to any request,
-    * a list of whitelisted content types
+    * a list of whitelisted content types for which `content-disposition: inline`
+    is always set (overriding any upstream header) so the media can be embedded
+    in pages; the filename is derived from the content type
+
+  * `sniff_content_type` (default `false`): detects an image MIME type from the first
+  response chunk when the upstream type is missing or `application/octet-stream`.
 
   * `req_headers`, `resp_headers` additional headers.
 
@@ -69,11 +75,15 @@ defmodule Pleroma.ReverseProxy do
   @default_options [pool: :media]
 
   @inline_content_types [
+    "image/apng",
+    "image/avif",
+    "image/bmp",
     "image/gif",
     "image/jpeg",
     "image/jpg",
     "image/png",
     "image/svg+xml",
+    "image/webp",
     "audio/mpeg",
     "audio/mp3",
     "video/webm",
@@ -92,6 +102,7 @@ defmodule Pleroma.ReverseProxy do
           | {:req_headers, [{String.t(), String.t()}]}
           | {:resp_headers, [{String.t(), String.t()}]}
           | {:inline_content_types, boolean() | list(String.t())}
+          | {:sniff_content_type, boolean()}
           | {:redirect_on_failure, boolean()}
 
   @spec call(Plug.Conn.t(), String.t(), list(option())) :: Plug.Conn.t()
@@ -116,13 +127,8 @@ defmodule Pleroma.ReverseProxy do
       end
 
     with {:ok, nil} <- @cachex.get(:failed_proxy_url_cache, url),
-         {:ok, code, headers, client} <- request(method, url, req_headers, client_opts),
-         :ok <-
-           enforce_header_length(
-             client,
-             headers,
-             Keyword.get(opts, :max_body_length, @max_body_length)
-           ) do
+         {:ok, code, headers, client} <-
+           request_with_constraints(method, url, req_headers, client_opts, opts) do
       response(conn, client, url, code, headers, opts)
     else
       {:ok, true} ->
@@ -189,26 +195,44 @@ defmodule Pleroma.ReverseProxy do
     end
   end
 
+  defp request_with_constraints(method, url, headers, client_opts, opts) do
+    case request(method, url, headers, client_opts) do
+      {:ok, _code, headers, client} = response ->
+        case header_length_constraint(
+               headers,
+               Keyword.get(opts, :max_body_length, @max_body_length)
+             ) do
+          :ok ->
+            response
+
+          error ->
+            client().close(client)
+            error
+        end
+
+      response ->
+        response
+    end
+  end
+
   defp response(conn, client, url, status, headers, opts) do
     Logger.debug("#{__MODULE__} #{status} #{url} #{inspect(headers)}")
 
+    {headers, client, prefetched} = maybe_prefetch_for_content_type(headers, client, opts)
+
     result =
       conn
       |> put_resp_headers(build_resp_headers(headers, opts))
       |> streaming_compat
       |> send_chunked(status)
-      |> chunk_reply(client, opts)
-
-    # Release the ref on every outcome, including a fully streamed body. The
-    # connection process behind it is not tied to this request and keeps the
-    # response body's binaries alive until it is explicitly closed.
-    client().close(client)
+      |> chunk_reply(client, opts, prefetched)
 
     case result do
       {:ok, conn} ->
         halt(conn)
 
       {:error, :closed, conn} ->
+        client().close(client)
         halt(conn)
 
       {:error, error, conn} ->
@@ -216,22 +240,33 @@ defmodule Pleroma.ReverseProxy do
           "#{__MODULE__} request to #{url} failed while reading/chunking: #{inspect(error)}"
         )
 
+        client().close(client)
         halt(conn)
     end
   end
 
-  defp chunk_reply(conn, client, opts) do
+  defp chunk_reply(conn, client, opts, :none) do
     chunk_reply(conn, client, opts, 0, 0)
   end
 
-  defp chunk_reply(conn, client, opts, sent_so_far, duration) do
-    with {:ok, duration} <-
-           check_read_duration(
-             duration,
-             Keyword.get(opts, :max_read_duration, @max_read_duration)
+  defp chunk_reply(conn, _client, _opts, :done), do: {:ok, conn}
+  defp chunk_reply(conn, _client, _opts, {:error, error}), do: {:error, error, conn}
+
+  defp chunk_reply(conn, client, opts, {:ok, data, duration}) do
+    with :ok <-
+           body_size_constraint(
+             byte_size(data),
+             Keyword.get(opts, :max_body_length, @max_body_length)
            ),
-         {:ok, data, client} <- client().stream_body(client),
-         {:ok, duration} <- increase_read_duration(duration),
+         {:ok, conn} <- chunk(conn, data) do
+      chunk_reply(conn, client, opts, byte_size(data), duration)
+    else
+      {:error, error} -> {:error, error, conn}
+    end
+  end
+
+  defp chunk_reply(conn, client, opts, sent_so_far, duration) do
+    with {:ok, data, client, duration} <- read_chunk(client, duration, opts),
          sent_so_far = sent_so_far + byte_size(data),
          :ok <-
            body_size_constraint(
@@ -246,6 +281,66 @@ defmodule Pleroma.ReverseProxy do
     end
   end
 
+  defp read_chunk(client, duration, opts) do
+    with {:ok, timer} <-
+           check_read_duration(
+             duration,
+             Keyword.get(opts, :max_read_duration, @max_read_duration)
+           ),
+         {:ok, data, client} <- client().stream_body(client),
+         {:ok, duration} <- increase_read_duration(timer) do
+      {:ok, data, client, duration}
+    end
+  end
+
+  defp maybe_prefetch_for_content_type(headers, client, opts) do
+    if Keyword.get(opts, :sniff_content_type, false) and generic_content_type?(headers) do
+      case read_chunk(client, 0, opts) do
+        {:ok, data, client, duration} ->
+          {maybe_put_image_content_type(headers, data), client, {:ok, data, duration}}
+
+        :done ->
+          {headers, client, :done}
+
+        {:error, error} ->
+          {headers, client, {:error, error}}
+      end
+    else
+      {headers, client, :none}
+    end
+  end
+
+  defp generic_content_type?(headers) do
+    headers
+    |> get_content_type()
+    |> String.trim()
+    |> String.downcase()
+    |> then(&(&1 in ["", "application/octet-stream"]))
+  end
+
+  defp maybe_put_image_content_type(headers, data) do
+    with false <- data == "",
+         {:ok, %{mime_type: "image/" <> _ = content_type}} <-
+           Majic.perform({:bytes, binary_part(data, 0, min(byte_size(data), @sniff_bytes))},
+             pool: Pleroma.MajicPool
+           ) do
+      [
+        {"content-type", content_type}
+        | Enum.reject(headers, fn {key, _} -> key == "content-type" end)
+      ]
+    else
+      _ -> headers
+    end
+  rescue
+    error ->
+      Logger.debug("#{__MODULE__}: content-type sniffing failed: #{Exception.message(error)}")
+      headers
+  catch
+    kind, reason ->
+      Logger.debug("#{__MODULE__}: content-type sniffing failed: #{kind}: #{inspect(reason)}")
+      headers
+  end
+
   defp head_response(conn, url, code, headers, opts) do
     Logger.debug("#{__MODULE__} #{code} #{url} #{inspect(headers)}")
 
@@ -292,17 +387,34 @@ defmodule Pleroma.ReverseProxy do
     |> Enum.filter(fn {k, _} -> k in @keep_req_headers end)
     |> build_req_range_or_encoding_header(opts)
     |> build_req_user_agent_header(opts)
-    |> Keyword.merge(Keyword.get(opts, :req_headers, []))
+    |> merge_headers(Keyword.get(opts, :req_headers, []))
+    |> maybe_force_identity_encoding(opts)
+  end
+
+  defp merge_headers(headers, extra_headers) do
+    Enum.reduce(extra_headers, headers, fn {key, value}, headers ->
+      List.keystore(headers, String.downcase(key), 0, {String.downcase(key), value})
+    end)
+  end
+
+  defp maybe_force_identity_encoding(headers, opts) do
+    if Keyword.get(opts, :sniff_content_type, false) do
+      List.keystore(headers, "accept-encoding", 0, {"accept-encoding", "identity"})
+    else
+      headers
+    end
   end
 
   # Disable content-encoding if any @range_headers are requested (see #1823).
   defp build_req_range_or_encoding_header(headers, _opts) do
     range? = Enum.any?(headers, fn {header, _} -> Enum.member?(@range_headers, header) end)
 
-    if range? && List.keymember?(headers, "accept-encoding", 0) do
-      List.keydelete(headers, "accept-encoding", 0)
-    else
-      headers
+    cond do
+      range? && List.keymember?(headers, "accept-encoding", 0) ->
+        List.keydelete(headers, "accept-encoding", 0)
+
+      true ->
+        headers
     end
   end
 
@@ -395,22 +507,33 @@ defmodule Pleroma.ReverseProxy do
 
       disposition = "attachment; filename=\"#{name}\""
 
-      List.keystore(headers, "content-disposition", 0, {"content-disposition", disposition})
+      replace_header(headers, "content-disposition", disposition)
     else
-      headers
+      if opt == true do
+        headers
+      else
+        name = inline_filename(content_type)
+
+        disposition =
+          if name do
+            "inline; filename=\"#{name}\""
+          else
+            "inline"
+          end
+
+        replace_header(headers, "content-disposition", disposition)
+      end
     end
   end
 
-  # Rejecting on content-length happens after the upstream response was already
-  # accepted, so the ref is live and has to be released before we bail out.
-  defp enforce_header_length(client, headers, limit) do
-    case header_length_constraint(headers, limit) do
-      :ok ->
-        :ok
+  defp replace_header(headers, key, value) do
+    [{key, value} | Enum.reject(headers, fn {header, _} -> header == key end)]
+  end
 
-      {:error, _} = error ->
-        client().close(client)
-        error
+  defp inline_filename(content_type) do
+    case MIME.extensions(content_type) do
+      [ext | _] when ext != "" -> "inline.#{ext}"
+      _ -> nil
     end
   end
 
@@ -430,7 +553,7 @@ defmodule Pleroma.ReverseProxy do
 
   defp header_length_constraint(_, _), do: :ok
 
-  defp body_size_constraint(size, limit) when is_integer(limit) and limit > 0 and size >= limit do
+  defp body_size_constraint(size, limit) when is_integer(limit) and limit > 0 and size > limit do
     {:error, :body_too_large}
   end
 
@@ -445,7 +568,7 @@ defmodule Pleroma.ReverseProxy do
     end
   end
 
-  defp check_read_duration(_, _), do: {:ok, :no_duration_limit, :no_duration_limit}
+  defp check_read_duration(_, _), do: {:ok, :no_duration_limit}
 
   defp receive_timeout(opts) do
     case Keyword.get(opts, :max_read_duration, @max_read_duration) do
@@ -454,6 +577,8 @@ defmodule Pleroma.ReverseProxy do
     end
   end
 
+  defp increase_read_duration(:no_duration_limit), do: {:ok, :no_duration_limit}
+
   defp increase_read_duration({previous_duration, started})
        when is_integer(previous_duration) and is_integer(started) do
     duration = :erlang.system_time(:millisecond) - started

lib/pleroma/reverse_proxy/client/hackney.ex

@@ -60,9 +60,20 @@ defmodule Pleroma.ReverseProxy.Client.Hackney do
     _ = :hackney.stream_next(ref)
 
     case receive_response(ref, recv_timeout) do
-      {:ok, data} when is_binary(data) -> {:ok, data, client}
-      {:ok, :done} -> :done
-      {:error, error} -> {:error, error}
+      {:ok, data} when is_binary(data) ->
+        {:ok, data, client}
+
+      # Reaching the end of the body does not free the connection: Hackney parks
+      # a finished conn instead of stopping it, and an unpooled one has neither
+      # an idle timeout nor a live owner monitor to reap it, so it would sit
+      # there forever pinning every binary it read. Release it here, mirroring
+      # what the gun client does, so callers only close what they abandon.
+      {:ok, :done} ->
+        close(client)
+        :done
+
+      {:error, error} ->
+        {:error, error}
     end
   end
 

lib/pleroma/reverse_proxy/client/tesla.ex

@@ -6,6 +6,7 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
   @behaviour Pleroma.ReverseProxy.Client
 
   alias Pleroma.Gun.ConnectionPool
+  alias Pleroma.HTTP.AdapterHelper
 
   @type headers() :: [{String.t(), String.t()}]
   @type status() :: pos_integer()
@@ -20,6 +21,13 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
   def request(method, url, headers, body, opts \\ []) do
     check_adapter()
 
+    # MediaProxyController inserts MediaProxy proxy configuration into the ReverseProxy call.
+    # This config can be unformatted and in String representation, while this works for Hackney,
+    # it does not work for Tesla and the proxy needs to be reformatted.
+    # Gun AdapterHelper doesn't do it, since it does not overwrite proxy config.
+    proxy = AdapterHelper.format_proxy(opts[:proxy])
+    opts = if proxy, do: Keyword.put(opts, :proxy, proxy), else: opts
+
     opts = Keyword.put(opts, :body_as, :chunks)
 
     with {:ok, response} <-
@@ -42,14 +50,13 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
     end
   end
 
-  # Reaching the end of the body does not release the ref: callers must always
-  # call close/1, exactly once, whatever the outcome. Releasing here as well
-  # would remove the client twice and crash the pool worker on the missing
-  # monitor.
   @impl true
   @spec stream_body(map()) ::
           {:ok, binary(), map()} | {:error, atom() | String.t()} | :done | no_return()
-  def stream_body(%{fin: true}), do: :done
+  def stream_body(%{pid: pid, stream: stream, fin: true}) do
+    ConnectionPool.release_stream(pid, stream)
+    :done
+  end
 
   def stream_body(client) do
     case read_chunk!(client) do
@@ -71,6 +78,10 @@ defmodule Pleroma.ReverseProxy.Client.Tesla do
 
   @impl true
   @spec close(map) :: :ok | no_return()
+  def close(%{pid: pid, stream: stream}) do
+    ConnectionPool.cancel_stream(pid, stream)
+  end
+
   def close(%{pid: pid}) do
     ConnectionPool.release_conn(pid)
   end

lib/pleroma/telemetry/logger.ex

@@ -55,8 +55,8 @@ defmodule Pleroma.Telemetry.Logger do
 
   def handle_event(
         [:pleroma, :connection_pool, :provision_failure],
-        %{opts: [key | _]},
         _,
+        %{key: key},
         _
       ) do
     Logger.debug(fn ->

lib/pleroma/tesla/middleware/connection_pool.ex

@@ -12,25 +12,38 @@ defmodule Pleroma.Tesla.Middleware.ConnectionPool do
   alias Pleroma.Gun.ConnectionPool
 
   @impl Tesla.Middleware
-  def call(%Tesla.Env{url: url, opts: opts} = env, next, _) do
+  def call(%Tesla.Env{opts: opts} = env, next, middleware_opts) do
+    if opts[:adapter][:body_as] == :stream do
+      {:error, :pooled_streaming_requires_chunks}
+    else
+      do_call(env, next, middleware_opts)
+    end
+  end
+
+  defp do_call(%Tesla.Env{url: url, opts: opts} = env, next, _) do
     uri = URI.parse(url)
 
     # Avoid leaking connections when the middleware is called twice
     # with body_as: :chunks. We assume only the middleware can set
     # opts[:adapter][:conn]
-    if opts[:adapter][:conn] do
-      ConnectionPool.release_conn(opts[:adapter][:conn])
-    end
+    cleanup_previous_request(env)
 
     case ConnectionPool.get_conn(uri, opts[:adapter]) do
       {:ok, conn_pid} ->
-        adapter_opts = Keyword.merge(opts[:adapter], conn: conn_pid, close_conn: false)
+        tunnel = ConnectionPool.tunnel_ref(conn_pid)
+
+        adapter_opts =
+          opts[:adapter]
+          |> Keyword.drop([:conn, :stream, :tunnel])
+          |> Keyword.merge(conn: conn_pid, close_conn: false, tunnel: tunnel)
+
         opts = Keyword.put(opts, :adapter, adapter_opts)
         env = %{env | opts: opts}
+        next = use_pool_adapter(next)
 
         case Tesla.run(env, next) do
           {:ok, env} ->
-            unless opts[:adapter][:body_as] == :chunks do
+            unless opts[:adapter][:body_as] == :chunks and chunk_client?(env.body) do
               ConnectionPool.release_conn(conn_pid)
               {_, res} = pop_in(env.opts[:adapter][:conn])
               {:ok, res}
@@ -38,8 +51,11 @@ defmodule Pleroma.Tesla.Middleware.ConnectionPool do
               {:ok, env}
             end
 
+          {:error, {:gun_stream_error, reason}} ->
+            {:error, reason}
+
           err ->
-            ConnectionPool.release_conn(conn_pid)
+            ConnectionPool.discard_conn(conn_pid)
             err
         end
 
@@ -47,4 +63,34 @@ defmodule Pleroma.Tesla.Middleware.ConnectionPool do
         err
     end
   end
+
+  defp cleanup_previous_request(%Tesla.Env{
+         body: %{pid: pid, stream: stream},
+         opts: opts
+       }) do
+    if opts[:adapter][:conn], do: ConnectionPool.cancel_stream(pid, stream)
+  end
+
+  defp cleanup_previous_request(%Tesla.Env{opts: opts}) do
+    conn = opts[:adapter][:conn]
+    stream = opts[:adapter][:stream]
+
+    cond do
+      conn && stream -> ConnectionPool.cancel_stream(conn, stream)
+      conn -> ConnectionPool.release_conn(conn)
+      true -> :ok
+    end
+  end
+
+  defp chunk_client?(%{pid: pid, stream: stream}) when is_pid(pid) and not is_nil(stream),
+    do: true
+
+  defp chunk_client?(_body), do: false
+
+  defp use_pool_adapter(next) do
+    List.update_at(next, -1, fn
+      {Tesla.Adapter.Gun, function, args} -> {Pleroma.HTTP.Adapter.Gun, function, args}
+      adapter -> adapter
+    end)
+  end
 end

lib/pleroma/user.ex

@@ -272,13 +272,13 @@ defmodule Pleroma.User do
   end
 
   def cached_blocked_users_ap_ids(user) do
-    @cachex.fetch!(:user_cache, "blocked_users_ap_ids:#{user.ap_id}", fn _ ->
+    @cachex.fetch!(:user_cache, cache_key(:blocked_users_ap_ids, user), fn _ ->
       blocked_users_ap_ids(user)
     end)
   end
 
   def cached_muted_users_ap_ids(user) do
-    @cachex.fetch!(:user_cache, "muted_users_ap_ids:#{user.ap_id}", fn _ ->
+    @cachex.fetch!(:user_cache, cache_key(:muted_users_ap_ids, user), fn _ ->
       muted_users_ap_ids(user)
     end)
   end
@@ -1253,13 +1253,18 @@ defmodule Pleroma.User do
     get_cached_by_nickname(nickname)
   end
 
+  defp cache_key(:id, %User{id: id}), do: cache_key(:id, id)
+  defp cache_key(:nickname, %User{nickname: nickname}), do: cache_key(:nickname, nickname)
+  defp cache_key(namespace, %User{ap_id: ap_id}), do: cache_key(namespace, ap_id)
+  defp cache_key(namespace, value), do: "#{namespace}:#{value}"
+
   def set_cache({:ok, user}), do: set_cache(user)
   def set_cache({:error, err}), do: {:error, err}
 
   def set_cache(%User{} = user) do
-    @cachex.put(:user_cache, "ap_id:#{user.ap_id}", user)
-    @cachex.put(:user_cache, "nickname:#{user.nickname}", user)
-    @cachex.put(:user_cache, "friends_ap_ids:#{user.nickname}", get_user_friends_ap_ids(user))
+    @cachex.put(:user_cache, cache_key(:ap_id, user), user)
+    @cachex.put(:user_cache, cache_key(:nickname, user), user)
+    @cachex.put(:user_cache, cache_key(:friends_ap_ids, user), get_user_friends_ap_ids(user))
     {:ok, user}
   end
 
@@ -1287,22 +1292,23 @@ defmodule Pleroma.User do
 
   @spec get_cached_user_friends_ap_ids(User.t()) :: [String.t()]
   def get_cached_user_friends_ap_ids(user) do
-    @cachex.fetch!(:user_cache, "friends_ap_ids:#{user.ap_id}", fn _ ->
+    @cachex.fetch!(:user_cache, cache_key(:friends_ap_ids, user), fn _ ->
       get_user_friends_ap_ids(user)
     end)
   end
 
   def invalidate_cache(user) do
-    @cachex.del(:user_cache, "ap_id:#{user.ap_id}")
-    @cachex.del(:user_cache, "nickname:#{user.nickname}")
-    @cachex.del(:user_cache, "friends_ap_ids:#{user.ap_id}")
-    @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
-    @cachex.del(:user_cache, "muted_users_ap_ids:#{user.ap_id}")
+    @cachex.del(:user_cache, cache_key(:ap_id, user))
+    @cachex.del(:user_cache, cache_key(:nickname, user))
+    @cachex.del(:user_cache, cache_key(:id, user))
+    @cachex.del(:user_cache, cache_key(:friends_ap_ids, user))
+    @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
+    @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, user))
   end
 
   @spec get_cached_by_ap_id(String.t()) :: User.t() | nil
   def get_cached_by_ap_id(ap_id) do
-    key = "ap_id:#{ap_id}"
+    key = cache_key(:ap_id, ap_id)
 
     with {:ok, nil} <- @cachex.get(:user_cache, key),
          user when not is_nil(user) <- get_by_ap_id(ap_id),
@@ -1315,14 +1321,14 @@ defmodule Pleroma.User do
   end
 
   def get_cached_by_id(id) do
-    key = "id:#{id}"
+    key = cache_key(:id, id)
 
     ap_id =
       @cachex.fetch!(:user_cache, key, fn _ ->
         user = get_by_id(id)
 
         if user do
-          @cachex.put(:user_cache, "ap_id:#{user.ap_id}", user)
+          @cachex.put(:user_cache, cache_key(:ap_id, user), user)
           {:commit, user.ap_id}
         else
           {:ignore, ""}
@@ -1333,7 +1339,7 @@ defmodule Pleroma.User do
   end
 
   def get_cached_by_nickname(nickname) do
-    key = "nickname:#{nickname}"
+    key = cache_key(:nickname, nickname)
 
     @cachex.fetch!(:user_cache, key, fn _ ->
       case get_or_fetch_by_nickname(nickname) do
@@ -1651,7 +1657,7 @@ defmodule Pleroma.User do
         |> Oban.insert()
       end
 
-      @cachex.del(:user_cache, "muted_users_ap_ids:#{muter.ap_id}")
+      @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, muter))
 
       {:ok, Enum.filter([user_mute, user_notification_mute], & &1)}
     end
@@ -1661,7 +1667,7 @@ defmodule Pleroma.User do
     with {:ok, user_mute} <- UserRelationship.delete_mute(muter, mutee),
          {:ok, user_notification_mute} <-
            UserRelationship.delete_notification_mute(muter, mutee) do
-      @cachex.del(:user_cache, "muted_users_ap_ids:#{muter.ap_id}")
+      @cachex.del(:user_cache, cache_key(:muted_users_ap_ids, muter))
       {:ok, [user_mute, user_notification_mute]}
     end
   end
@@ -2815,7 +2821,7 @@ defmodule Pleroma.User do
           {:ok, UserRelationship.t()} | {:error, Ecto.Changeset.t()}
   defp add_to_block(%User{} = user, %User{} = blocked, expires_at) do
     with {:ok, relationship} <- UserRelationship.create_block(user, blocked, expires_at) do
-      @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
+      @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
       {:ok, relationship}
     end
   end
@@ -2824,7 +2830,7 @@ defmodule Pleroma.User do
           {:ok, UserRelationship.t()} | {:ok, nil} | {:error, Ecto.Changeset.t()}
   defp remove_from_block(%User{} = user, %User{} = blocked) do
     with {:ok, relationship} <- UserRelationship.delete_block(user, blocked) do
-      @cachex.del(:user_cache, "blocked_users_ap_ids:#{user.ap_id}")
+      @cachex.del(:user_cache, cache_key(:blocked_users_ap_ids, user))
       {:ok, relationship}
     end
   end

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -537,9 +537,12 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     #   and extra sorting on "activities.id DESC NULLS LAST" would worse the query plan
     opts = Map.put(opts, :skip_extra_order, true)
 
-    Pagination.fetch_paginated(query, opts, pagination)
+    Pagination.fetch_paginated(query, opts, pagination, pagination_binding(opts))
   end
 
+  defp pagination_binding(%{favorited_by: _}), do: :favorited_activity
+  defp pagination_binding(_), do: nil
+
   def fetch_activities(recipients, opts \\ %{}, pagination \\ :keyset) do
     list_memberships = Pleroma.List.memberships(opts[:user])
 
@@ -786,6 +789,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
 
   defp restrict_since(query, %{since_id: ""}), do: query
 
+  defp restrict_since(query, %{favorited_by: _, since_id: _}), do: query
+
   defp restrict_since(query, %{since_id: since_id}) do
     from(activity in query, where: activity.id > ^since_id)
   end
@@ -893,14 +898,22 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
       |> Repo.all()
 
     # Note: NO extra ordering should be done on "activities.id desc nulls last" for optimal plan
-    from(
-      [_activity, object] in query,
-      join: hto in "hashtags_objects",
-      on: hto.object_id == object.id,
-      where: hto.hashtag_id in ^hashtag_ids,
-      distinct: [desc: object.id],
-      order_by: [desc: object.id]
-    )
+    query =
+      from(
+        [_activity, object] in query,
+        join: hto in "hashtags_objects",
+        on: hto.object_id == object.id,
+        where: hto.hashtag_id in ^hashtag_ids
+      )
+
+    if is_nil(query.distinct) do
+      from([_activity, object] in query,
+        distinct: [desc: object.id],
+        order_by: [desc: object.id]
+      )
+    else
+      query
+    end
   end
 
   defp restrict_hashtag_any(query, %{tag: tag}) when is_binary(tag) do
@@ -1012,9 +1025,31 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   defp restrict_assigned_account(query, _), do: query
 
   defp restrict_favorited_by(query, %{favorited_by: ap_id}) do
+    newer_favorite =
+      from(newer_favorite in Activity,
+        where: newer_favorite.actor == ^ap_id,
+        where: newer_favorite.id > parent_as(:favorited_activity).id,
+        where: fragment("?->>'type' = ?", newer_favorite.data, "Like"),
+        where:
+          fragment(
+            "associated_object_id(?) = associated_object_id(?)",
+            newer_favorite.data,
+            parent_as(:favorited_activity).data
+          ),
+        select: 1
+      )
+
     from(
-      [_activity, object] in query,
-      where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id)
+      [activity, object: object] in query,
+      join: favorited_activity in Activity,
+      as: :favorited_activity,
+      on:
+        favorited_activity.actor == ^ap_id and
+          fragment("?->>'type' = ?", favorited_activity.data, "Like") and
+          fragment("associated_object_id(?) = (?)->>'id'", favorited_activity.data, object.data),
+      where: fragment("(?)->'likes' \\? (?)", object.data, ^ap_id),
+      where: not exists(subquery(newer_favorite)),
+      select: %Activity{activity | pagination_id: favorited_activity.id}
     )
   end
 

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

@@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.ActivityExpirationPolicy do
   @moduledoc "Adds expiration to all local Create activities"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @impl true
   def filter(activity) do
     activity =

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

@@ -3,11 +3,11 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.AntiFollowbotPolicy do
+  @moduledoc "Prevent followbots from following with a bit of heuristic"
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
   alias Pleroma.User
 
-  @moduledoc "Prevent followbots from following with a bit of heuristic"
-
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   # XXX: this should become User.normalize_by_ap_id() or similar, really.
   defp normalize_by_ap_id(%{"id" => id}), do: User.get_cached_by_ap_id(id)

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

@@ -3,9 +3,11 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.AntiLinkSpamPolicy do
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.User
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   require Logger
 

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

@@ -3,11 +3,13 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.AntiMentionSpamPolicy do
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Config
   alias Pleroma.User
   require Pleroma.Constants
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   defp user_has_posted?(%User{} = u), do: u.note_count > 0
 

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

@@ -3,10 +3,12 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.DropPolicy do
-  require Logger
   @moduledoc "Drop and log everything received"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  require Logger
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @impl true
   def filter(activity) do
     Logger.debug("REJECTING #{inspect(activity)}")

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

@@ -3,6 +3,8 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.EmojiPolicy do
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   require Pleroma.Constants
 
   alias Pleroma.Object.Updater
@@ -10,7 +12,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.EmojiPolicy do
 
   @moduledoc "Reject or force-unlisted emojis with certain URLs or names"
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   defp config_remove_url do
     Pleroma.Config.get([:mrf_emoji, :remove_url], [])

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

@@ -3,11 +3,13 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.EnsureRePrepended do
-  alias Pleroma.Object
-
   @moduledoc "Ensure a re: is prepended on replies to a post with a Subject"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  alias Pleroma.Object
+
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @reply_prefix Regex.compile!("^re:[[:space:]]*", [:caseless])
 
   def history_awareness, do: :auto

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

@@ -4,12 +4,15 @@
 
 defmodule Pleroma.Web.ActivityPub.MRF.FollowBotPolicy do
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Config
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
 
   require Logger
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @impl true
   def filter(activity) do
     with follower_nickname <- Config.get([:mrf_follow_bot, :follower_nickname]),

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

@@ -3,9 +3,12 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.ForceBotUnlistedPolicy do
-  alias Pleroma.User
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
   @moduledoc "Remove bot posts from federated timeline"
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
+  alias Pleroma.User
+
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   require Pleroma.Constants
 

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

@@ -3,13 +3,14 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.ForceMention do
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
   require Pleroma.Constants
 
   alias Pleroma.Config
   alias Pleroma.Object
   alias Pleroma.User
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   defp get_author(url) do
     with %Object{data: %{"actor" => actor}} <- Object.normalize(url, fetch: false),

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

@@ -3,13 +3,15 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.ForceMentionsInContent do
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   require Pleroma.Constants
 
   alias Pleroma.Formatter
   alias Pleroma.Object
   alias Pleroma.User
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   @impl true
   def history_awareness, do: :auto

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

@@ -3,6 +3,8 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.HashtagPolicy do
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   require Pleroma.Constants
 
   alias Pleroma.Config
@@ -14,7 +16,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.HashtagPolicy do
   Note: This MRF Policy is always enabled, if you want to disable it you have to set empty lists.
   """
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   @impl true
   def history_awareness, do: :manual

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

@@ -3,13 +3,14 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.HellthreadPolicy do
+  @moduledoc "Block activities with too much mentions (configurable)"
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.User
 
   require Pleroma.Constants
 
-  @moduledoc "Block activities with too much mentions (configurable)"
-
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   defp delist_activity(activity, threshold) when threshold > 0 do
     follower_collection = User.get_cached_by_ap_id(activity["actor"]).follower_address

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

@@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.InlineQuotePolicy do
   @moduledoc "Force a quote line into the message content."
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   defp build_inline_quote(template, url) do
     quote_line = String.replace(template, "{url}", "<a href=\"#{url}\">#{url}</a>")
 

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

@@ -3,13 +3,14 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.KeywordPolicy do
+  @moduledoc "Reject or Word-Replace activities with a keyword or regex"
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   require Pleroma.Constants
 
   alias Pleroma.Web.ActivityPub.MRF.Utils
 
-  @moduledoc "Reject or Word-Replace activities with a keyword or regex"
-
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   defp string_matches?(string, pattern) when is_binary(pattern) do
     String.contains?(string, pattern)

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

@@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy do
   @moduledoc "Preloads any attachments in the MediaProxy cache by prefetching them"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.HTTP
   alias Pleroma.Web.MediaProxy
 

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

@@ -4,9 +4,10 @@
 
 defmodule Pleroma.Web.ActivityPub.MRF.MentionPolicy do
   @moduledoc "Block activities which mention a user"
-
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @impl true
   def filter(%{"type" => "Create"} = activity) do
     reject_actors = Pleroma.Config.get([:mrf_mention, :actors], [])

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

@@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.NoEmptyPolicy do
   @moduledoc "Filter local activities which have no content"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Web.Endpoint
 
   @impl true

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

@@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.NoOpPolicy do
   @moduledoc "Does nothing (lets the messages go through unmodified)"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @impl true
   def filter(activity) do
     {:ok, activity}

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

@@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.NoPlaceholderTextPolicy do
   @moduledoc "Ensure no content placeholder is present (such as the dot from mastodon)"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @impl true
   def history_awareness, do: :auto
 

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

@@ -4,9 +4,11 @@
 
 defmodule Pleroma.Web.ActivityPub.MRF.NormalizeMarkup do
   @moduledoc "Scrub configured hypertext markup"
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.HTML
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   @impl true
   def history_awareness, do: :auto

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

@@ -38,6 +38,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.NsfwApiPolicy do
   - `unlist`: Unlist all detected NSFW content? Default: `false`
   - `reject`: Reject all detected NSFW content (takes precedence)? Default: `false`
   """
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Config
   alias Pleroma.Constants
   alias Pleroma.HTTP
@@ -46,7 +48,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.NsfwApiPolicy do
   require Logger
   require Pleroma.Constants
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @policy :mrf_nsfw_api
 
   def build_request_url(url) do

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

@@ -3,13 +3,15 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.ObjectAgePolicy do
+  @moduledoc "Filter activities depending on their age"
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Config
   alias Pleroma.User
 
   require Pleroma.Constants
 
-  @moduledoc "Filter activities depending on their age"
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   defp check_date(%{"object" => %{"published" => published}} = activity) do
     with %DateTime{} = now <- DateTime.utc_now(),

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

@@ -15,4 +15,12 @@ defmodule Pleroma.Web.ActivityPub.MRF.Policy do
             }
   @callback history_awareness() :: :auto | :manual
   @optional_callbacks config_description: 0, history_awareness: 0, id_filter: 1
+
+  defmacro __using__(_) do
+    quote do
+      # Otherwise Pleroma.Docs.Generator.list_behaviour_implementations fails
+      # since the Elixir compiler doesn't automatically load the module since v1.19.
+      @compile {:autoload, true}
+    end
+  end
 end

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

@@ -6,11 +6,13 @@ defmodule Pleroma.Web.ActivityPub.MRF.QuietReply do
   @moduledoc """
   QuietReply alters the scope of activities from local users when replying by enforcing them to be "Unlisted" or "Quiet Public". This delivers the activity to all the expected recipients and instances, but it will not be published in the Federated / The Whole Known Network timelines. It will still be published to the Home timelines of the user's followers and visible to anyone who opens the thread.
   """
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   require Pleroma.Constants
 
   alias Pleroma.User
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   @impl true
   def history_awareness, do: :auto

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

@@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.QuoteToLinkTagPolicy do
   @moduledoc "Force a Link tag for posts quoting another post. (may break outgoing federation of quote posts with older Pleroma versions)"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Web.ActivityPub.ObjectValidators.CommonFixes
 
   require Pleroma.Constants

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

@@ -4,11 +4,12 @@
 
 defmodule Pleroma.Web.ActivityPub.MRF.RejectNonPublic do
   @moduledoc "Rejects non-public (followers-only, direct) activities"
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
   alias Pleroma.Config
   alias Pleroma.User
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   require Pleroma.Constants
 

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

@@ -2,6 +2,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.RemoteReportPolicy do
   @moduledoc "Drop remote reports if they don't contain enough information."
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Config
 
   @impl true

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

@@ -6,6 +6,8 @@ defmodule Pleroma.Web.ActivityPub.MRF.SimplePolicy do
   @moduledoc "Filter activities depending on their origin instance"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Config
   alias Pleroma.FollowingRelationship
   alias Pleroma.User

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

@@ -3,12 +3,14 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.StealEmojiPolicy do
+  @moduledoc "Detect new emojis by their shortcode and steals them"
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   require Logger
 
   alias Pleroma.Config
 
-  @moduledoc "Detect new emojis by their shortcode and steals them"
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   defp accept_host?(host), do: host in Config.get([:mrf_steal_emoji, :hosts], [])
 

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

@@ -3,12 +3,14 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.SubchainPolicy do
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
   alias Pleroma.Config
   alias Pleroma.Web.ActivityPub.MRF
 
   require Logger
 
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   defp lookup_subchain(actor) do
     with matches <- Config.get([:mrf_subchain, :match_actor]),

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

@@ -3,8 +3,6 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.TagPolicy do
-  alias Pleroma.User
-  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
   @moduledoc """
      Apply policies based on user tags
 
@@ -18,6 +16,11 @@ defmodule Pleroma.Web.ActivityPub.MRF.TagPolicy do
      - `mrf_tag:disable-remote-subscription`: Reject non-local follow requests
      - `mrf_tag:disable-any-subscription`: Reject any follow requests
   """
+  @behaviour Pleroma.Web.ActivityPub.MRF.Policy
+
+  alias Pleroma.User
+
+  use Pleroma.Web.ActivityPub.MRF.Policy
 
   require Pleroma.Constants
 

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

@@ -3,11 +3,13 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.MRF.UserAllowListPolicy do
-  alias Pleroma.Config
-
   @moduledoc "Accept-list of users from specified instances"
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  alias Pleroma.Config
+
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   defp filter_by_list(activity, []), do: {:ok, activity}
 
   defp filter_by_list(%{"actor" => actor} = activity, allow_list) do

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

@@ -4,9 +4,10 @@
 
 defmodule Pleroma.Web.ActivityPub.MRF.VocabularyPolicy do
   @moduledoc "Filter activities which belong to certain activity vocabularies"
-
   @behaviour Pleroma.Web.ActivityPub.MRF.Policy
 
+  use Pleroma.Web.ActivityPub.MRF.Policy
+
   @impl true
   def filter(%{"type" => "Undo", "object" => object} = activity) do
     with {:ok, _} <- filter(object) do

lib/pleroma/web/activity_pub/views/user_view.ex

@@ -68,7 +68,7 @@ defmodule Pleroma.Web.ActivityPub.UserView do
     capabilities =
       if is_boolean(user.accepts_chat_messages) do
         %{
-          "acceptsChatMessages" => user.accepts_chat_messages
+          "acceptsChatMessages" => Pleroma.Chat.enabled?() and user.accepts_chat_messages
         }
       else
         %{}

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

@@ -71,7 +71,10 @@ defmodule Pleroma.Web.AdminAPI.ConfigController do
   end
 
   def descriptions(conn, _params) do
-    descriptions = Enum.filter(Pleroma.Docs.JSON.compiled_descriptions(), &whitelisted_config?/1)
+    descriptions =
+      Pleroma.Docs.JSON.compiled_descriptions()
+      |> Enum.filter(&allowed_config?(&1, :whitelist))
+      |> Enum.reject(&allowed_config?(&1, :blacklist))
 
     json(conn, translate_descriptions(descriptions))
   end
@@ -132,7 +135,8 @@ defmodule Pleroma.Web.AdminAPI.ConfigController do
     with :ok <- configurable_from_database() do
       results =
         configs
-        |> Enum.filter(&whitelisted_config?/1)
+        |> Enum.filter(&allowed_config?(&1, :whitelist))
+        |> Enum.reject(&allowed_config?(&1, :blacklist))
         |> Enum.map(fn
           %{group: group, key: key, delete: true} = params ->
             ConfigDB.delete(%{group: group, key: key, subkeys: params[:subkeys]})
@@ -174,27 +178,34 @@ defmodule Pleroma.Web.AdminAPI.ConfigController do
     end
   end
 
-  defp whitelisted_config?(":pleroma", ":database_config_whitelist"), do: false
+  defp allowed_config?(":pleroma", ":database_config_whitelist", _), do: true
+  defp allowed_config?(":pleroma", ":database_config_blacklist", _), do: true
 
-  defp whitelisted_config?(group, key) do
-    if whitelisted_configs = Config.get(:database_config_whitelist) do
-      Enum.any?(whitelisted_configs, fn
-        {whitelisted_group} ->
-          group == inspect(whitelisted_group)
+  defp allowed_config?(group, key, mode) do
+    configs =
+      case mode do
+        :whitelist -> Config.get(:database_config_whitelist)
+        :blacklist -> Config.get(:database_config_blacklist)
+      end
 
-        {whitelisted_group, whitelisted_key} ->
-          group == inspect(whitelisted_group) && key == inspect(whitelisted_key)
+    if configs do
+      Enum.any?(configs, fn
+        {allowed_group} ->
+          group == inspect(allowed_group)
+
+        {allowed_group, allowed_key} ->
+          group == inspect(allowed_group) && key == inspect(allowed_key)
       end)
     else
       true
     end
   end
 
-  defp whitelisted_config?(%{group: group, key: key}) do
-    whitelisted_config?(group, key)
+  defp allowed_config?(%{group: group, key: key}, mode) do
+    allowed_config?(group, key, mode)
   end
 
-  defp whitelisted_config?(%{group: group} = config) do
-    whitelisted_config?(group, config[:key])
+  defp allowed_config?(%{group: group} = config, mode) do
+    allowed_config?(group, config[:key], mode)
   end
 end

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

@@ -31,7 +31,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
             "The updated chat",
             "application/json",
             Chat
-          )
+          ),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{
@@ -56,7 +57,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
             "The read ChatMessage",
             "application/json",
             ChatMessage
-          )
+          ),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{
@@ -87,7 +89,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
             "The existing chat",
             "application/json",
             Chat
-          )
+          ),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{
@@ -118,7 +121,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
             "The created or existing chat",
             "application/json",
             Chat
-          )
+          ),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{
@@ -146,7 +150,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
         Operation.parameter(:pinned, :query, BooleanLike.schema(), "Include only pinned chats")
       ],
       responses: %{
-        200 => Operation.response("The chats of the user", "application/json", chats_response())
+        200 => Operation.response("The chats of the user", "application/json", chats_response()),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{
@@ -172,7 +177,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
         | pagination_params()
       ],
       responses: %{
-        200 => Operation.response("The chats of the user", "application/json", chats_response())
+        200 => Operation.response("The chats of the user", "application/json", chats_response()),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{
@@ -224,6 +230,7 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
             ChatMessage
           ),
         400 => Operation.response("Bad Request", "application/json", ApiError),
+        404 => Operation.response("Not Found", "application/json", ApiError),
         422 => Operation.response("MRF Rejection", "application/json", ApiError)
       },
       security: [
@@ -249,7 +256,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
             "The deleted ChatMessage",
             "application/json",
             ChatMessage
-          )
+          ),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{
@@ -268,7 +276,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
         Operation.parameter(:id, :path, :string, "The id of the chat", required: true)
       ],
       responses: %{
-        200 => Operation.response("The existing chat", "application/json", Chat)
+        200 => Operation.response("The existing chat", "application/json", Chat),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{
@@ -287,7 +296,8 @@ defmodule Pleroma.Web.ApiSpec.ChatOperation do
         Operation.parameter(:id, :path, :string, "The id of the chat", required: true)
       ],
       responses: %{
-        200 => Operation.response("The existing chat", "application/json", Chat)
+        200 => Operation.response("The existing chat", "application/json", Chat),
+        404 => Operation.response("Not Found", "application/json", ApiError)
       },
       security: [
         %{

lib/pleroma/web/endpoint.ex

@@ -139,6 +139,8 @@ defmodule Pleroma.Web.Endpoint do
     from: {:pleroma, "priv/static/adminfe/"}
   )
 
+  plug(Pleroma.Web.Plugs.StaticNotFoundPlug)
+
   # Code reloading can be explicitly enabled under the
   # :code_reloader configuration of your endpoint.
   if code_reloading? do

lib/pleroma/web/federator.ex

@@ -89,7 +89,8 @@ defmodule Pleroma.Web.Federator do
 
   # Job Worker Callbacks
 
-  @spec perform(atom(), any()) :: {:ok, any()} | {:error, any()}
+  @impl true
+  @spec perform(atom(), any()) :: :ok | {:ok, any()} | {:error, any()}
   def perform(:publish_one, params) do
     Publisher.prepare_one(params)
     |> Publisher.publish_one()

lib/pleroma/web/federator/publishing.ex

@@ -3,5 +3,6 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.Federator.Publishing do
+  @callback perform(atom(), any()) :: :ok | {:ok, any()} | {:error, any()}
   @callback publish(map()) :: any()
 end

lib/pleroma/web/feed/feed_view.ex

@@ -38,6 +38,20 @@ defmodule Pleroma.Web.Feed.FeedView do
     end
   end
 
+  def most_recent_update([], :rss), do: nil
+
+  def most_recent_update(activities, :rss) do
+    activities
+    |> Enum.map(fn activity ->
+      case activity.object do
+        %Object{updated_at: updated_at} -> updated_at
+        _ -> activity.updated_at
+      end
+    end)
+    |> Enum.max_by(&NaiveDateTime.to_erl/1)
+    |> to_rfc2822()
+  end
+
   def most_recent_update(activities, user, :atom) do
     (List.first(activities) || user).updated_at
     |> to_rfc3339()
@@ -97,6 +111,67 @@ defmodule Pleroma.Web.Feed.FeedView do
     end
   end
 
+  def rss_content_encoded(data) do
+    data
+    |> feed_content_encoded()
+    |> String.replace("]]>", "]]>")
+  end
+
+  def atom_content_encoded(data), do: feed_content_encoded(data)
+
+  defp feed_content_encoded(data) do
+    base_content =
+      case activity_content(data) do
+        "" ->
+          (data["summary"] || data["type"] || "")
+          |> escape()
+
+        content ->
+          content
+      end
+
+    attachments_html =
+      if sensitive?(data) do
+        ""
+      else
+        (data["attachment"] || [])
+        |> Enum.map(&rss_attachment_preview/1)
+        |> Enum.reject(&(&1 == ""))
+        |> Enum.join("<br/><br/>")
+      end
+
+    [base_content, attachments_html]
+    |> Enum.reject(&(&1 == ""))
+    |> Enum.join("<br/><br/>")
+  end
+
+  defp rss_attachment_preview(attachment) do
+    href = attachment_href(attachment)
+
+    if is_binary(href) and href != "" do
+      media_type = attachment_type(attachment) || "application/octet-stream"
+      escaped_href = escape(href)
+      name = attachment["name"] || "Attachment"
+      escaped_name = escape(name)
+
+      cond do
+        String.starts_with?(media_type, "image/") ->
+          ~s(<p><img src="#{escaped_href}" alt="#{escaped_name}" loading="lazy"/></p>)
+
+        String.starts_with?(media_type, "video/") ->
+          ~s(<p><video controls preload="metadata" src="#{escaped_href}"></video></p><p><a href="#{escaped_href}">#{escaped_name}</a></p>)
+
+        String.starts_with?(media_type, "audio/") ->
+          ~s(<p><audio controls preload="metadata" src="#{escaped_href}"></audio></p><p><a href="#{escaped_href}">#{escaped_name}</a></p>)
+
+        true ->
+          ~s(<p><a href="#{escaped_href}">#{escaped_name}</a></p>)
+      end
+    else
+      ""
+    end
+  end
+
   def activity_content(%{"content" => content}) do
     content
     |> String.replace(~r/[\n\r]/, "")
@@ -106,16 +181,104 @@ defmodule Pleroma.Web.Feed.FeedView do
 
   def activity_context(activity), do: escape(activity.data["context"])
 
+  def sensitive?(data), do: data["sensitive"] in [true, "true"]
+
+  def feed_self_url(conn) do
+    conn
+    |> Phoenix.Controller.current_url()
+    |> escape()
+  end
+
+  def activity_link(activity, data) do
+    cond do
+      activity.local -> data["id"] || data["url"]
+      true -> data["external_url"] || data["id"] || data["url"]
+    end
+  end
+
   def attachment_href(attachment) do
-    attachment["url"]
-    |> hd()
+    attachment
+    |> attachment_url_data()
     |> Map.get("href")
   end
 
   def attachment_type(attachment) do
-    attachment["url"]
-    |> hd()
-    |> Map.get("mediaType")
+    attachment
+    |> attachment_url_data()
+    |> Map.get("mediaType") || attachment["mediaType"]
+  end
+
+  def attachment_size(attachment) do
+    attachment
+    |> attachment_url_data()
+    |> Map.get("size", 0)
+  end
+
+  def attachment_size_positive(attachment) do
+    case attachment_size(attachment) do
+      size when is_integer(size) and size > 0 ->
+        size
+
+      size when is_binary(size) ->
+        case Integer.parse(size) do
+          {parsed, ""} when parsed > 0 -> parsed
+          _ -> nil
+        end
+
+      _ ->
+        nil
+    end
+  end
+
+  def attachment_medium(attachment) do
+    case attachment_type(attachment) do
+      "image/" <> _rest -> "image"
+      "video/" <> _rest -> "video"
+      "audio/" <> _rest -> "audio"
+      _ -> "document"
+    end
+  end
+
+  def rss_enclosure_attachment(attachments) do
+    Enum.find(attachments || [], &(not is_nil(attachment_size_positive(&1))))
+  end
+
+  def attachment_description(attachment) do
+    attachment["name"] || ""
+  end
+
+  def media_content_xml(_attachment, sensitive) when sensitive in [true, "true"], do: ""
+
+  def media_content_xml(attachment, _sensitive) do
+    href = escape(attachment_href(attachment) || "")
+    type = escape(attachment_type(attachment) || "application/octet-stream")
+    medium = escape(attachment_medium(attachment))
+
+    file_size_attr =
+      case attachment_size_positive(attachment) do
+        size when is_integer(size) -> ~s( fileSize="#{size}")
+        _ -> ""
+      end
+
+    case attachment_description(attachment) do
+      "" ->
+        ~s(<media:content url="#{href}" type="#{type}"#{file_size_attr} medium="#{medium}"/>\n)
+
+      description ->
+        """
+        <media:content url="#{href}" type="#{type}"#{file_size_attr} medium="#{medium}">
+          <media:description type="plain">#{escape(description)}</media:description>
+        </media:content>
+        """
+    end
+  end
+
+  defp attachment_url_data(attachment) do
+    case attachment["url"] do
+      [first | _] when is_map(first) -> first
+      %{} = map -> map
+      _ -> %{}
+    end
   end
 
   def get_href(id) do

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

@@ -716,7 +716,10 @@ defmodule Pleroma.Web.MastodonAPI.StatusController do
     end
   end
 
-  defp put_application(params, %{assigns: %{token: %Token{user: %User{} = user} = token}} = _conn) do
+  defp put_application(
+         params,
+         %{assigns: %{token: %Token{} = token, user: %User{} = user}} = _conn
+       ) do
     if user.disclose_client do
       %{client_name: client_name, website: website} = Repo.preload(token, :app).app
       Map.put(params, :generator, %{type: "Application", name: client_name, url: website})

lib/pleroma/web/mastodon_api/mastodon_api.ex

@@ -385,6 +385,15 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPI do
       |> cast_params()
       |> Map.update(:include_types, [], fn include_types -> include_types end)
 
+    if "pleroma:chat_mention" in options.include_types and not Pleroma.Chat.enabled?() do
+      options
+      |> Map.update(:include_types, [], fn include_types ->
+        List.delete(include_types, "pleroma:chat_mention")
+      end)
+    else
+      options
+    end
+
     if ("pleroma:report" not in options.include_types and
           User.privileged?(user, :reports_manage_reports)) or
          User.privileged?(user, :reports_manage_reports) do

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

@@ -287,6 +287,8 @@ defmodule Pleroma.Web.MastodonAPI.AccountView do
       user.last_status_at &&
         user.last_status_at |> NaiveDateTime.to_date() |> Date.to_iso8601()
 
+    accepts_chat_messages = Pleroma.Chat.enabled?() and user.accepts_chat_messages
+
     %{
       id: to_string(user.id),
       username: username_from_nickname(user.nickname),
@@ -336,7 +338,7 @@ defmodule Pleroma.Web.MastodonAPI.AccountView do
         relationship: relationship,
         skip_thread_containment: user.skip_thread_containment,
         background_image: image_url(user.background) |> MediaProxy.url(),
-        accepts_chat_messages: user.accepts_chat_messages,
+        accepts_chat_messages: accepts_chat_messages,
         favicon: favicon,
         avatar_description: avatar_description,
         header_description: header_description
@@ -391,6 +393,7 @@ defmodule Pleroma.Web.MastodonAPI.AccountView do
     |> Kernel.put_in([:source, :pleroma, :show_role], user.show_role)
     |> Kernel.put_in([:source, :pleroma, :no_rich_text], user.no_rich_text)
     |> Kernel.put_in([:source, :pleroma, :show_birthday], user.show_birthday)
+    |> Kernel.put_in([:source, :pleroma, :accepts_chat_messages], user.accepts_chat_messages)
   end
 
   defp maybe_put_settings(data, _, _, _), do: data

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

@@ -202,8 +202,12 @@ defmodule Pleroma.Web.MastodonAPI.InstanceView do
       end,
       "pleroma_emoji_reactions",
       "pleroma_custom_emoji_reactions",
-      "pleroma_chat_messages",
-      "pleroma:pin_chats",
+      if Pleroma.Chat.enabled?() do
+        "pleroma_chat_messages"
+      end,
+      if Pleroma.Chat.enabled?() do
+        "pleroma:pin_chats"
+      end,
       if Config.get([:instance, :show_reactions]) do
         "exposable_reactions"
       end,

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

@@ -620,6 +620,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do
         String.contains?(media_type, "image") -> "image"
         String.contains?(media_type, "video") -> "video"
         String.contains?(media_type, "audio") -> "audio"
+        attachment["type"] == "Image" -> "image"
         true -> "unknown"
       end
 

lib/pleroma/web/media_proxy/media_proxy_controller.ex

@@ -208,6 +208,7 @@ defmodule Pleroma.Web.MediaProxy.MediaProxyController do
 
   defp media_proxy_opts do
     Config.get([:media_proxy, :proxy_opts], [])
+    |> Keyword.put_new(:sniff_content_type, true)
   end
 
   defp sandbox(conn, _params) do

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

@@ -41,6 +41,7 @@ defmodule Pleroma.Web.PleromaAPI.ChatController do
   )
 
   plug(Pleroma.Web.ApiSpec.CastAndValidate, replace_params: false)
+  plug(:ensure_chats_enabled)
 
   defdelegate open_api_operation(action), to: Pleroma.Web.ApiSpec.ChatOperation
 
@@ -244,4 +245,16 @@ defmodule Pleroma.Web.PleromaAPI.ChatController do
       _ -> nil
     end
   end
+
+  defp ensure_chats_enabled(conn, _opts) do
+    if Chat.enabled?() do
+      conn
+    else
+      conn
+      |> put_status(:not_found)
+      |> put_view(Pleroma.Web.ErrorView)
+      |> render("404.json")
+      |> halt()
+    end
+  end
 end

lib/pleroma/web/plugs/ensure_host_matches_plug.ex

@@ -4,6 +4,7 @@
 
 defmodule Pleroma.Web.Plugs.EnsureHostMatchesPlug do
   @moduledoc "Ensures Host header matches instance"
+  @behaviour Plug
 
   alias Pleroma.Web.Endpoint
 

lib/pleroma/web/plugs/instance_static.ex

@@ -4,7 +4,7 @@
 
 defmodule Pleroma.Web.Plugs.InstanceStatic do
   require Pleroma.Constants
-  import Plug.Conn, only: [put_resp_header: 3, put_status: 2, send_resp: 3, halt: 1]
+  import Plug.Conn, only: [put_resp_header: 3]
 
   @moduledoc """
   This is a shim to call `Plug.Static` but with runtime `from` configuration.
@@ -51,37 +51,10 @@ defmodule Pleroma.Web.Plugs.InstanceStatic do
       |> Map.put(:from, from)
       |> Map.put(:content_types, false)
 
-    conn =
-      conn
-      |> set_content_type(conn.request_path)
-      |> Plug.Static.call(opts)
+    conn = set_content_type(conn, conn.request_path)
 
-    if conn.halted do
-      conn
-    else
-      path = String.trim_leading(conn.request_path, "/")
-
-      if not File.exists?(file_path(path)) do
-        conn
-        |> put_status(:not_found)
-        |> send_404()
-        |> halt()
-      else
-        conn
-      end
-    end
-  end
-
-  defp send_404(conn) do
-    if String.ends_with?(String.downcase(conn.request_path), ".json") do
-      conn
-      |> put_resp_header("content-type", "application/json")
-      |> send_resp(404, Jason.encode!(%{error: "not found"}))
-    else
-      conn
-      |> put_resp_header("content-type", "text/plain")
-      |> send_resp(404, "Not found")
-    end
+    # Call Plug.Static with our sanitized content-type
+    Plug.Static.call(conn, opts)
   end
 
   defp set_content_type(conn, "/emoji/" <> filepath) do

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) |> Repo.preload(:user),
+    with %Token{user_id: user_id} = token_record <- Repo.one(token_query),
          false <- is_nil(user_id),
          %User{} = user <- User.get_cached_by_id(user_id) do
       {:ok, user, token_record}

lib/pleroma/web/plugs/static_not_found_plug.ex

@@ -0,0 +1,38 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2026 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.Plugs.StaticNotFoundPlug do
+  @moduledoc """
+  A small plug that enforces returning 404 on static files when they don't exist.
+  """
+  @behaviour Plug
+
+  require Pleroma.Constants
+  import Plug.Conn, only: [put_resp_header: 3, send_resp: 3, halt: 1]
+
+  def init(opts), do: opts
+
+  for only <- Pleroma.Constants.static_only_files() do
+    def call(%{request_path: "/" <> unquote(only) <> _} = conn, _opts) do
+      send_404(conn)
+    end
+  end
+
+  def call(conn, _) do
+    conn
+  end
+
+  defp send_404(conn) do
+    if String.ends_with?(String.downcase(conn.request_path), ".json") do
+      conn
+      |> put_resp_header("content-type", "application/json")
+      |> send_resp(404, Jason.encode!(%{error: "not found"}))
+    else
+      conn
+      |> put_resp_header("content-type", "text/plain")
+      |> send_resp(404, "Not found")
+    end
+    |> halt()
+  end
+end

lib/pleroma/web/rich_media/helpers.ex

@@ -4,6 +4,8 @@
 
 defmodule Pleroma.Web.RichMedia.Helpers do
   alias Pleroma.Config
+  alias Pleroma.Gun.ConnectionPool
+  alias Pleroma.ReverseProxy.Client.Tesla, as: TeslaClient
 
   require Logger
 
@@ -19,12 +21,30 @@ defmodule Pleroma.Web.RichMedia.Helpers do
   end
 
   defp stream(url) do
-    with {_, {:ok, %Tesla.Env{status: 200, body: stream_body, headers: headers}}} <-
-           {:get, Pleroma.HTTP.get(url, req_headers(), http_options())},
-         {_, :ok} <- {:content_type, check_content_type(headers)},
-         {_, :ok} <- {:content_length, check_content_length(headers)},
-         {:read_stream, {:ok, body}} <- {:read_stream, read_stream(stream_body)} do
-      {:ok, body}
+    case Pleroma.HTTP.get(url, req_headers(), http_options()) do
+      {:ok, %Tesla.Env{status: 200, body: stream_body, headers: headers}} ->
+        result =
+          try do
+            with {_, :ok} <- {:content_type, check_content_type(headers)},
+                 {_, :ok} <- {:content_length, check_content_length(headers)},
+                 {:read_stream, {:ok, body}} <- {:read_stream, read_stream(stream_body)} do
+              {:ok, body}
+            end
+          rescue
+            exception ->
+              cleanup_stream(stream_body, :error)
+              reraise exception, __STACKTRACE__
+          end
+
+        cleanup_stream(stream_body, result)
+        result
+
+      {:ok, %Tesla.Env{body: stream_body}} = response ->
+        cleanup_stream(stream_body, :rejected)
+        {:get, response}
+
+      response ->
+        {:get, response}
     end
   end
 
@@ -95,6 +115,10 @@ defmodule Pleroma.Web.RichMedia.Helpers do
     end
   end
 
+  defp read_stream(%{pid: pid, stream: stream, opts: opts}) do
+    read_chunks(pid, stream, opts, "", 0, Keyword.get(http_options(), :max_body))
+  end
+
   defp read_stream(stream) do
     max_body = Keyword.get(http_options(), :max_body)
 
@@ -117,6 +141,33 @@ defmodule Pleroma.Web.RichMedia.Helpers do
     end
   end
 
+  defp read_chunks(pid, stream, opts, acc, total_bytes, max_body) do
+    case Tesla.Adapter.Gun.read_chunk(pid, stream, opts) do
+      {fin, chunk} when fin in [:fin, :nofin] ->
+        total_bytes = total_bytes + byte_size(chunk)
+
+        cond do
+          total_bytes > max_body ->
+            :error
+
+          fin == :fin ->
+            {:ok, acc <> chunk}
+
+          true ->
+            read_chunks(pid, stream, opts, acc <> chunk, total_bytes, max_body)
+        end
+
+      {:error, _reason} ->
+        :error
+    end
+  end
+
+  defp cleanup_stream(%{pid: pid, stream: stream}, {:ok, _body}),
+    do: ConnectionPool.release_stream(pid, stream)
+
+  defp cleanup_stream(%{pid: _pid} = client, _result), do: TeslaClient.close(client)
+  defp cleanup_stream(_stream, _result), do: :ok
+
   defp http_options do
     [
       pool: :rich_media,

lib/pleroma/web/streamer.ex

@@ -195,6 +195,9 @@ defmodule Pleroma.Web.Streamer do
     recipients = MapSet.new(item.recipients)
     domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)
 
+    following_ap_ids =
+      if user.domain_blocks == [], do: [], else: User.get_cached_user_friends_ap_ids(user)
+
     with parent <- Object.normalize(item, fetch: false) || item,
          true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
          true <- item.data["type"] != "Announce" || item.actor not in reblog_muted_ap_ids,
@@ -205,8 +208,12 @@ defmodule Pleroma.Web.Streamer do
          true <- MapSet.disjoint?(recipients, recipient_blocks),
          %{host: item_host} <- URI.parse(item.actor),
          %{host: parent_host} <- URI.parse(parent.data["actor"]),
-         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
-         false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
+         false <-
+           item.actor not in following_ap_ids and
+             Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
+         false <-
+           parent.data["actor"] not in following_ap_ids and
+             Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
          true <- thread_containment(item, user),
          false <- CommonAPI.thread_muted?(parent, user) do
       false
@@ -289,15 +296,17 @@ defmodule Pleroma.Web.Streamer do
 
   defp do_stream(topic, {user, %MessageReference{} = cm_ref})
        when topic in ["user", "user:pleroma_chat"] do
-    topic = "#{topic}:#{user.id}"
+    if Pleroma.Chat.enabled?() do
+      topic = "#{topic}:#{user.id}"
 
-    text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref}, topic)
+      text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref}, topic)
 
-    Registry.dispatch(@registry, topic, fn list ->
-      Enum.each(list, fn {pid, _auth} ->
-        send(pid, {:text, text})
+      Registry.dispatch(@registry, topic, fn list ->
+        Enum.each(list, fn {pid, _auth} ->
+          send(pid, {:text, text})
+        end)
       end)
-    end)
+    end
   end
 
   defp do_stream("user", %Activity{} = item) do

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

@@ -1,9 +1,9 @@
 <entry>
   <activity:object-type>http://activitystrea.ms/schema/1.0/note</activity:object-type>
   <activity:verb>http://activitystrea.ms/schema/1.0/post</activity:verb>
-  <id><%= @data["id"] %></id>
+  <id><%= escape(@data["id"] || "") %></id>
   <title><%= activity_title(@data, Keyword.get(@feed_config, :post_title, %{})) %></title>
-  <content type="html"><%= activity_description(@data) %></content>
+  <content type="html"><%= escape(atom_content_encoded(@data)) %></content>
   <published><%= to_rfc3339(@data["published"]) %></published>
   <updated><%= to_rfc3339(@data["published"]) %></updated>
   <ostatus:conversation ref="<%= activity_context(@activity) %>">
@@ -15,35 +15,26 @@
     <summary><%= escape(@data["summary"]) %></summary>
   <% end %>
 
-  <%= if @activity.local do %>
-    <link type="application/atom+xml" href='<%= @data["id"] %>' rel="self"/>
-    <link type="text/html" href='<%= @data["id"] %>' rel="alternate"/>
-  <% else %>
-    <link type="text/html" href='<%= @data["external_url"] %>' rel="alternate"/>
-  <% end %>
+  <link type="text/html" href='<%= escape(activity_link(@activity, @data) || "") %>' rel="alternate"/>
 
   <%= for tag <- Pleroma.Object.hashtags(@object) do %>
     <category term="<%= tag %>"></category>
   <% end %>
 
-  <%= for attachment <- @data["attachment"] || [] do %>
-    <link rel="enclosure" href="<%= attachment_href(attachment) %>" type="<%= attachment_type(attachment) %>"/>
+  <%= unless sensitive?(@data) do %>
+    <%= for attachment <- @data["attachment"] || [] do %>
+      <%= if size = attachment_size_positive(attachment) do %>
+        <link rel="enclosure" href="<%= escape(attachment_href(attachment) || "") %>" type="<%= escape(attachment_type(attachment) || "application/octet-stream") %>" length="<%= size %>"/>
+      <% else %>
+        <link rel="enclosure" href="<%= escape(attachment_href(attachment) || "") %>" type="<%= escape(attachment_type(attachment) || "application/octet-stream") %>"/>
+      <% end %>
+    <% end %>
   <% end %>
 
   <%= if @data["inReplyTo"] do %>
     <thr:in-reply-to ref='<%= @data["inReplyTo"] %>' href='<%= get_href(@data["inReplyTo"]) %>'/>
   <% end %>
 
-  <%= for id <- @activity.recipients do %>
-    <%= if id == Pleroma.Constants.as_public() do %>
-      <link rel="mentioned" ostatus:object-type="http://activitystrea.ms/schema/1.0/collection" href="http://activityschema.org/collection/public"/>
-    <% else %>
-      <%= unless Regex.match?(~r/^#{Pleroma.Web.Endpoint.url()}.+followers$/, id) do %>
-        <link rel="mentioned" ostatus:object-type="http://activitystrea.ms/schema/1.0/person" href="<%= id %>"/>
-      <% end %>
-    <% end %>
-  <% end %>
-
   <%= for {emoji, file} <- @data["emoji"] || %{} do %>
     <link name="<%= emoji %>" rel="emoji" href="<%= file %>"/>
   <% end %>

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

@@ -1,45 +1,34 @@
 <item>
   <activity:object-type>http://activitystrea.ms/schema/1.0/note</activity:object-type>
   <activity:verb>http://activitystrea.ms/schema/1.0/post</activity:verb>
-  <guid><%= @data["id"] %></guid>
+  <guid isPermaLink="true"><%= escape(@data["id"] || "") %></guid>
   <title><%= activity_title(@data, Keyword.get(@feed_config, :post_title, %{})) %></title>
   <description><%= activity_description(@data) %></description>
+  <content:encoded><![CDATA[<%= rss_content_encoded(@data) %>]]></content:encoded>
   <pubDate><%= to_rfc2822(@data["published"]) %></pubDate>
   <ostatus:conversation ref="<%= activity_context(@activity) %>">
     <%= activity_context(@activity) %>
   </ostatus:conversation>
 
-  <%= if @activity.local do %>
-    <link><%= @data["id"] %></link>
-  <% else %>
-    <link><%= @data["external_url"] %></link>
-  <% end %>
-
-  <link rel="ostatus:conversation"><%= activity_context(@activity) %></link>
+  <link><%= escape(activity_link(@activity, @data) || "") %></link>
 
   <%= for tag <- Pleroma.Object.hashtags(@object) do %>
-    <category term="<%= tag %>"></category>
+    <%= if tag && tag != "" do %>
+      <category><%= tag %></category>
+    <% end %>
+  <% end %>
+
+  <%= unless sensitive?(@data) do %>
+    <%= with attachment when not is_nil(attachment) <- rss_enclosure_attachment(@data["attachment"]) do %>
+      <enclosure url="<%= escape(attachment_href(attachment) || "") %>" type="<%= escape(attachment_type(attachment) || "application/octet-stream") %>" length="<%= attachment_size_positive(attachment) %>" />
+    <% end %>
   <% end %>
 
   <%= for attachment <- @data["attachment"] || [] do %>
-    <enclosure url="<%= attachment_href(attachment) %>" type="<%= attachment_type(attachment) %>" />
+    <%= media_content_xml(attachment, sensitive?(@data)) %>
   <% end %>
 
   <%= if @data["inReplyTo"] do %>
     <thr:in-reply-to ref='<%= @data["inReplyTo"] %>' href='<%= get_href(@data["inReplyTo"]) %>'/>
   <% end %>
-
-  <%= for id <- @activity.recipients do %>
-    <%= if id == Pleroma.Constants.as_public() do %>
-      <link rel="mentioned" ostatus:object-type="http://activitystrea.ms/schema/1.0/collection">http://activityschema.org/collection/public</link>
-    <% else %>
-      <%= unless Regex.match?(~r/^#{Pleroma.Web.Endpoint.url()}.+followers$/, id) do %>
-        <link rel="mentioned" ostatus:object-type="http://activitystrea.ms/schema/1.0/person"><%= id %></link>
-      <% end %>
-    <% end %>
-  <% end %>
-
-  <%= for {emoji, file} <- @data["emoji"] || %{} do %>
-    <link name="<%= emoji %>" rel="emoji"><%= file %></link>
-  <% end %>
 </item>

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

@@ -4,9 +4,9 @@
 
     <%= render Phoenix.Controller.view_module(@conn), "_tag_author.atom", assigns %>
 
-  <id><%= @data["id"] %></id>
+  <id><%= escape(@data["id"] || "") %></id>
   <title><%= activity_title(@data, Keyword.get(@feed_config, :post_title, %{})) %></title>
-  <content type="html"><%= activity_description(@data) %></content>
+  <content type="html"><%= escape(atom_content_encoded(@data)) %></content>
   <published><%= to_rfc3339(@data["published"]) %></published>
   <updated><%= to_rfc3339(@data["published"]) %></updated>
   <ostatus:conversation ref="<%= activity_context(@activity) %>">
@@ -15,34 +15,25 @@
   <link href="<%= activity_context(@activity) %>" rel="ostatus:conversation"/>
 
   <%= if @data["summary"] != "" do %>
-    <summary><%= @data["summary"] %></summary>
+    <summary><%= escape(@data["summary"]) %></summary>
   <% end %>
 
-  <%= if @activity.local do %>
-    <link type="application/atom+xml" href='<%= @data["id"] %>' rel="self"/>
-    <link type="text/html" href='<%= @data["id"] %>' rel="alternate"/>
-  <% else %>
-    <link type="text/html" href='<%= @data["external_url"] %>' rel="alternate"/>
-  <% end %>
-
-  <%= for id <- @activity.recipients do %>
-    <%= if id == Pleroma.Constants.as_public() do %>
-      <link rel="mentioned"
-            ostatus:object-type="http://activitystrea.ms/schema/1.0/collection"
-            href="http://activityschema.org/collection/public"/>
-    <% else %>
-      <%= unless Regex.match?(~r/^#{Pleroma.Web.Endpoint.url()}.+followers$/, id) do %>
-        <link rel="mentioned"
-              ostatus:object-type="http://activitystrea.ms/schema/1.0/person"
-              href="<%= id %>" />
-      <% end %>
-    <% end %>
-  <% end %>
+  <link type="text/html" href='<%= escape(activity_link(@activity, @data) || "") %>' rel="alternate"/>
 
   <%= for tag <- Pleroma.Object.hashtags(@object) do %>
     <category term="<%= tag %>"></category>
   <% end %>
 
+  <%= unless sensitive?(@data) do %>
+    <%= for attachment <- @data["attachment"] || [] do %>
+      <%= if size = attachment_size_positive(attachment) do %>
+        <link rel="enclosure" href="<%= escape(attachment_href(attachment) || "") %>" type="<%= escape(attachment_type(attachment) || "application/octet-stream") %>" length="<%= size %>"/>
+      <% else %>
+        <link rel="enclosure" href="<%= escape(attachment_href(attachment) || "") %>" type="<%= escape(attachment_type(attachment) || "application/octet-stream") %>"/>
+      <% end %>
+    <% end %>
+  <% end %>
+
   <%= for {emoji, file} <- @data["emoji"] || %{} do %>
     <link name="<%= emoji %>" rel="emoji" href="<%= file %>"/>
   <% end %>

lib/pleroma/web/templates/feed/feed/_tag_activity.xml.eex

@@ -1,14 +1,24 @@
 <item>
-  <title><%= activity_title(@data, Keyword.get(@feed_config, :post_title, %{})) %></title>
-
-
-  <guid isPermalink="true"><%= activity_context(@activity) %></guid>
-  <link><%= activity_context(@activity) %></link>
+  <guid isPermaLink="true"><%= escape(activity_link(@activity, @data) || "") %></guid>
+  <link><%= escape(activity_link(@activity, @data) || "") %></link>
   <pubDate><%= to_rfc2822(@data["published"]) %></pubDate>
-
+  <title><%= activity_title(@data, Keyword.get(@feed_config, :post_title, %{})) %></title>
   <description><%= activity_description(@data) %></description>
-  <%= for attachment <- @data["attachment"] || [] do %>
-    <enclosure url="<%= attachment_href(attachment) %>" type="<%= attachment_type(attachment) %>"/>
+  <content:encoded><![CDATA[<%= rss_content_encoded(@data) %>]]></content:encoded>
+
+  <%= unless sensitive?(@data) do %>
+    <%= with attachment when not is_nil(attachment) <- rss_enclosure_attachment(@data["attachment"]) do %>
+      <enclosure url="<%= escape(attachment_href(attachment) || "") %>" type="<%= escape(attachment_type(attachment) || "application/octet-stream") %>" length="<%= attachment_size_positive(attachment) %>"/>
+    <% end %>
   <% end %>
 
+  <%= for attachment <- @data["attachment"] || [] do %>
+    <%= media_content_xml(attachment, sensitive?(@data)) %>
+  <% end %>
+
+  <%= for tag <- Pleroma.Object.hashtags(@object) do %>
+    <%= if tag && tag != "" do %>
+      <category><%= tag %></category>
+    <% end %>
+  <% end %>
 </item>

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

@@ -9,10 +9,12 @@
 
   <id><%= Routes.tag_feed_url(@conn, :feed, @tag) <> ".atom" %></id>
   <title>#<%= @tag %></title>
-  <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>
+  <%= with subtitle when subtitle not in [nil, ""] <- 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) do %>
+    <subtitle><%= subtitle %></subtitle>
+  <% end %>
   <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"/>
+  <updated><%= most_recent_update(@activities) || to_rfc3339(NaiveDateTime.utc_now()) %></updated>
+  <link rel="self" href="<%= feed_self_url(@conn) %>" 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

@@ -1,6 +1,8 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <rss version="2.0"
      xmlns:webfeeds="http://webfeeds.org/rss/1.0"
+     xmlns:content="http://purl.org/rss/1.0/modules/content/"
+     xmlns:media="http://search.yahoo.com/mrss/"
      xmlns:thr="http://purl.org/syndication/thread/1.0">
   <channel>
 
@@ -8,7 +10,12 @@
     <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>
     <webfeeds:logo><%= feed_logo() %></webfeeds:logo>
+    <webfeeds:icon><%= feed_logo() %></webfeeds:icon>
     <webfeeds:accentColor>2b90d9</webfeeds:accentColor>
+    <%= with last_build_date when not is_nil(last_build_date) <- most_recent_update(@activities, :rss) do %>
+      <lastBuildDate><%= last_build_date %></lastBuildDate>
+    <% end %>
+    <generator><%= Pleroma.Application.named_version() %></generator>
     <%= for activity <- @activities do %>
     <%= render Phoenix.Controller.view_module(@conn), "_tag_activity.xml", Map.merge(assigns, prepare_activity(activity)) %>
     <% end %>

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

@@ -8,10 +8,12 @@
 
   <id><%= Routes.user_feed_url(@conn, :feed, @user.nickname) <> ".atom" %></id>
   <title><%= @user.nickname <> "'s timeline" %></title>
-  <subtitle><%= escape(@user.bio) %></subtitle>
+  <%= if @user.bio && @user.bio != "" do %>
+    <subtitle><%= escape(@user.bio) %></subtitle>
+  <% end %>
   <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="<%= feed_self_url(@conn) %>" type="application/atom+xml"/>
 
   <%= render Phoenix.Controller.view_module(@conn), "_author.atom", assigns %>
 

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

@@ -1,26 +1,32 @@
 <?xml version="1.0" encoding="UTF-8" ?>
 <rss version="2.0"
      xmlns:atom="http://www.w3.org/2005/Atom"
+     xmlns:content="http://purl.org/rss/1.0/modules/content/"
+     xmlns:webfeeds="http://webfeeds.org/rss/1.0"
+     xmlns:media="http://search.yahoo.com/mrss/"
      xmlns:thr="http://purl.org/syndication/thread/1.0"
      xmlns:activity="http://activitystrea.ms/spec/1.0/"
      xmlns:ostatus="http://ostatus.org/schema/1.0"
      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>
-    <atom:link href="<%= Routes.user_feed_url(@conn, :feed, @user.nickname) <> ".atom" %>"
+    <atom:link href="<%= feed_self_url(@conn) %>"
                rel="self" type="application/rss+xml" />
+    <link><%= "#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.rss" %></link>
     <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>
     </image>
+    <lastBuildDate><%= most_recent_update(@activities, @user, :rss) %></lastBuildDate>
+    <webfeeds:icon><%= logo(@user) %></webfeeds:icon>
+    <generator><%= Pleroma.Application.named_version() %></generator>
 
     <%= 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>
+      <atom:link rel="next" href="<%= "#{Routes.user_feed_url(@conn, :feed, @user.nickname)}.rss?max_id=#{last_activity(@activities).id}" %>" />
     <% end %>
 
     <%= for activity <- @activities do %>

lib/pleroma/workers/publisher_worker.ex

@@ -4,25 +4,39 @@
 
 defmodule Pleroma.Workers.PublisherWorker do
   alias Pleroma.Activity
+  alias Pleroma.Config
   alias Pleroma.Instances
   alias Pleroma.Web.Federator
 
   use Oban.Worker, queue: :federator_outgoing, max_attempts: 13
 
+  @publish_one_param_keys %{
+    "activity_id" => :activity_id,
+    "cc" => :cc,
+    "inbox" => :inbox,
+    "unreachable_since" => :unreachable_since
+  }
+
   @impl true
   def perform(%Job{args: %{"op" => "publish", "activity_id" => activity_id}}) do
     activity = Activity.get_by_id(activity_id)
-    Federator.perform(:publish, activity)
+    federator().perform(:publish, activity)
   end
 
   def perform(%Job{args: %{"op" => "publish_one", "params" => params}} = job) do
-    params = Map.new(params, fn {k, v} -> {String.to_atom(k), v} end)
+    params =
+      Enum.reduce(params, %{}, fn {key, value}, acc ->
+        case @publish_one_param_keys do
+          %{^key => atom_key} -> Map.put(acc, atom_key, value)
+          _ -> acc
+        end
+      end)
 
     # Cancel / skip the job if this server believed to be unreachable now
     if not Instances.reachable?(params.inbox) do
       {:cancel, :unreachable}
     else
-      case Federator.perform(:publish_one, params) do
+      case federator().perform(:publish_one, params) do
         {:ok, _} ->
           :ok
 
@@ -44,6 +58,8 @@ defmodule Pleroma.Workers.PublisherWorker do
   @impl true
   def timeout(_job), do: :timer.seconds(10)
 
+  defp federator, do: Config.get([:pipeline, :federator], Federator)
+
   @base_backoff 15
   @pow 5
   @impl true