retry behaviour

config/config.exs

@@ -888,6 +888,8 @@
   url: "http://127.0.0.1:5000",
   api_key: nil
 
+config :pleroma, :queue, module: Pleroma.Broadway
+
 # Import environment specific config. This must remain at the bottom
 # of this file so it overrides the configuration defined above.
 import_config "#{Mix.env()}.exs"

lib/mix/tasks/pleroma/database.ex

@@ -178,7 +178,7 @@ def run(["ensure_expiration"]) do
           |> DateTime.from_naive!("Etc/UTC")
           |> Timex.shift(days: days)
 
-        Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
+        Pleroma.Workers.PurgeExpiredActivity.schedule(%{
           activity_id: activity.id,
           expires_at: expires_at
         })

lib/mix/tasks/pleroma/queue.ex

@@ -0,0 +1,18 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Mix.Tasks.Pleroma.Queue do
+  use Mix.Task
+
+  import Mix.Pleroma
+
+  def run(["queues"]) do
+    start_pleroma()
+
+    Pleroma.Config.get([Oban, :queues])
+    |> Keyword.keys()
+    |> Enum.join("\n")
+    |> shell_info()
+  end
+end

lib/pleroma/application.ex

@@ -68,11 +68,11 @@ def start(_type, _args) do
       ] ++
         cachex_children() ++
         http_children() ++
+        queue_children() ++
         [
           Pleroma.Stats,
           Pleroma.JobQueueMonitor,
           {Majic.Pool, [name: Pleroma.MajicPool, pool_size: Config.get([:majic_pool, :size], 2)]},
-          {Oban, Config.get(Oban)},
           Pleroma.Web.Endpoint
         ] ++
         elasticsearch_children() ++
@@ -267,4 +267,16 @@ defp http_children do
 
     [{Finch, config}]
   end
+
+  defp queue_children do
+    queue_module = Config.get([:queue, :module])
+
+    case queue_module do
+      Oban ->
+        [{Oban, Config.get(Oban)}]
+
+      Pleroma.Broadway ->
+        Pleroma.Broadway.children()
+    end
+  end
 end

lib/pleroma/broadway.ex

@@ -0,0 +1,190 @@
+defmodule Pleroma.Broadway do
+  use Broadway
+  alias Broadway.Message
+  require Logger
+
+  @queue "akkoma"
+  @exchange "akkoma_exchange"
+  @retry_header "x-retries"
+  @delay_header "x-delay"
+
+  def start_link(_args) do
+    Broadway.start_link(__MODULE__,
+      name: __MODULE__,
+      producer: [
+        module:
+          {BroadwayRabbitMQ.Producer,
+           queue: @queue,
+           after_connect: &declare_rabbitmq/1,
+           metadata: [:routing_key, :headers],
+           on_failure: :reject}
+      ],
+      processors: [
+        default: [
+          concurrency: 10
+        ]
+      ],
+      batchers: [
+        default: [
+          batch_size: 10,
+          batch_timeout: 100,
+          concurrency: 10
+        ]
+      ]
+    )
+  end
+
+  defp declare_rabbitmq(amqp_channel) do
+    declare_exchanges(amqp_channel)
+    declare_queues(amqp_channel)
+    declare_bindings(amqp_channel)
+  end
+
+  defp declare_exchanges(amqp_channel) do
+    # Main exchange, all messages go here
+    :ok =
+      AMQP.Exchange.declare(amqp_channel, @exchange, :"x-delayed-message",
+        durable: true,
+        arguments: [{"x-delayed-type", :longstr, "topic"}]
+      )
+  end
+
+  defp declare_queues(amqp_channel) do
+    # Main queue, bound to main exchange
+    {:ok, _} = AMQP.Queue.declare(amqp_channel, @queue, durable: true)
+  end
+
+  defp declare_bindings(amqp_channel) do
+    :ok = AMQP.Queue.bind(amqp_channel, @queue, @exchange, routing_key: "#")
+  end
+
+  defp retry_count(:undefined), do: 0
+  defp retry_count(headers) do
+    match = Enum.find(headers, fn {k, _t, _v} -> k == @retry_header end)
+    if is_nil(match) do
+      0
+    else
+      elem(match, 2)
+    end
+  end
+
+  @impl true
+  def handle_message(_, %Message{data: data, metadata: %{routing_key: routing_key, headers: headers}} = message, _) do
+    Logger.debug("Received message on #{routing_key}")
+    with {:ok, data} <- Jason.decode(data),
+         {module, data} <- Map.pop(data, "__module__"),
+         module <- String.to_existing_atom(module),
+         :ok <- perform_message(module, data) do
+      message
+    else
+      err ->
+        retries = retry_count(headers)
+        if retries > 5 do
+          Message.failed(message, err)
+        else
+          {:ok, _} = produce(routing_key, data, scheduled_in: 5000, retry_count: retries + 1)
+          message
+        end
+    end
+  end
+
+  defp perform_message(module, args) do
+    case module.perform(%Oban.Job{args: args}) do
+      :ok ->
+        :ok
+
+      {:ok, _} ->
+        :ok
+
+      err ->
+        err
+    end
+  end
+
+  @impl true
+  def handle_batch(_, batch, _, _) do
+    batch
+  end
+
+  @impl true
+  def handle_failed(messages, _) do
+    for message <- messages do
+      %Message{data: data, metadata: %{routing_key: topic}, status: {:failed, reason}} = message
+      {:ok, %{"op" => op}} = Jason.decode(data)
+      Logger.error("Processing task on #{topic}(#{op}) failed: #{inspect(reason)}")
+    end
+
+    messages
+  end
+
+  def topics do
+    Pleroma.Config.get([Oban, :queues])
+    |> Keyword.keys()
+  end
+
+  def children do
+    [Pleroma.Broadway]
+  end
+
+  defp add_headers([headers: headers] = opts, key, type, value) when is_list(headers) do
+    Keyword.put(opts, :headers, [{key, type, value} | headers])
+  end
+
+  defp add_headers(opts, key, type, value) do
+    Keyword.put(opts, :headers, [{key, type, value}])
+  end
+
+  defp maybe_with_priority(opts, params) do
+    if !is_nil(params[:priority]) do
+      Keyword.put(opts, :priority, params[:priority])
+    else
+      opts
+    end
+  end
+
+  defp maybe_schedule_at(opts, params) do
+    if !is_nil(params[:scheduled_at]) do
+      time_in_ms = DateTime.diff(params[:scheduled_at], DateTime.utc_now())
+      opts
+      |> add_headers(@delay_header, :long, time_in_ms)
+    else
+      opts
+    end
+  end
+
+  defp maybe_schedule_in(opts, params) do
+    if !is_nil(params[:scheduled_in]) do
+      opts
+      |> add_headers(@delay_header, :long, params[:scheduled_in])
+    else
+      opts
+    end
+  end
+
+
+  defp maybe_with_retry_count(opts, params) do
+    if !is_nil(params[:retry_count]) do
+      opts
+      |> add_headers(@retry_header, :long, params[:retry_count])
+    else
+      opts
+    end
+  end
+
+  def produce(topic, args, opts \\ []) do
+    {:ok, connection} = AMQP.Connection.open()
+    {:ok, channel} = AMQP.Channel.open(connection)
+
+    publish_options =
+      []
+      |> maybe_with_priority(opts)
+      |> maybe_schedule_at(opts)
+      |> maybe_schedule_in(opts)
+      |> maybe_with_retry_count(opts)
+
+    Logger.debug("Sending to #{topic} with #{inspect(publish_options)}")
+    :ok = AMQP.Basic.publish(channel, @exchange, topic, args, publish_options)
+    :ok = AMQP.Connection.close(connection)
+    {:ok, args}
+  end
+end

lib/pleroma/filter.ex

@@ -133,7 +133,7 @@ defp maybe_add_expires_at(changeset, %{expires_in: nil}) do
   defp maybe_add_expires_at(changeset, _), do: changeset
 
   defp maybe_add_expiration_job(%{expires_at: %NaiveDateTime{} = expires_at} = filter) do
-    Pleroma.Workers.PurgeExpiredFilter.enqueue(%{
+    Pleroma.Workers.PurgeExpiredFilter.schedule(%{
       filter_id: filter.id,
       expires_at: DateTime.from_naive!(expires_at, "Etc/UTC")
     })

lib/pleroma/mfa/token.ex

@@ -52,7 +52,7 @@ defp expired?(%__MODULE__{valid_until: valid_until}) do
   @spec create(User.t(), Authorization.t() | nil) :: {:ok, t()} | {:error, Ecto.Changeset.t()}
   def create(user, authorization \\ nil) do
     with {:ok, token} <- do_create(user, authorization) do
-      Pleroma.Workers.PurgeExpiredToken.enqueue(%{
+      Pleroma.Workers.PurgeExpiredToken.schedule(%{
         token_id: token.id,
         valid_until: DateTime.from_naive!(token.valid_until, "Etc/UTC"),
         mod: __MODULE__

lib/pleroma/user/backup.ex

@@ -34,7 +34,7 @@ defmodule Pleroma.User.Backup do
   def create(user, admin_id \\ nil) do
     with :ok <- validate_limit(user, admin_id),
          {:ok, backup} <- user |> new() |> Repo.insert() do
-      BackupWorker.process(backup, admin_id)
+      BackupWorker.enqueue("process", %{"backup_id" => backup.id, "admin_user_id" => admin_id})
     end
   end
 

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -230,7 +230,7 @@ defp maybe_create_activity_expiration(
          %{data: %{"expires_at" => %DateTime{} = expires_at}} = activity
        ) do
     with {:ok, _job} <-
-           Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
+           Pleroma.Workers.PurgeExpiredActivity.schedule(%{
              activity_id: activity.id,
              expires_at: expires_at
            }) do

lib/pleroma/web/activity_pub/side_effects.ex

@@ -384,7 +384,7 @@ def handle(%{data: %{"type" => "Remove"} = data} = object, meta) do
         {:ok, expires_at} =
           Pleroma.EctoType.ActivityPub.ObjectValidators.DateTime.cast(meta[:expires_at])
 
-        Pleroma.Workers.PurgeExpiredActivity.enqueue(%{
+        Pleroma.Workers.PurgeExpiredActivity.schedule(%{
           activity_id: meta[:activity_id],
           expires_at: expires_at
         })

lib/pleroma/web/o_auth/token.ex

@@ -117,7 +117,7 @@ defp put_valid_until(changeset, attrs) do
   def create(%App{} = app, %User{} = user, attrs \\ %{}) do
     with {:ok, token} <- do_create(app, user, attrs) do
       if Pleroma.Config.get([:oauth2, :clean_expired_tokens]) do
-        Pleroma.Workers.PurgeExpiredToken.enqueue(%{
+        Pleroma.Workers.PurgeExpiredToken.schedule(%{
           token_id: token.id,
           valid_until: DateTime.from_naive!(token.valid_until, "Etc/UTC"),
           mod: __MODULE__

lib/pleroma/workers/backup_worker.ex

@@ -3,36 +3,27 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Workers.BackupWorker do
-  use Oban.Worker, queue: :backup, max_attempts: 1
+  use Pleroma.Workers.WorkerHelper, queue: "backup", max_attempts: 1
-
   alias Oban.Job
   alias Pleroma.User.Backup
 
-  def process(backup, admin_user_id \\ nil) do
-    %{"op" => "process", "backup_id" => backup.id, "admin_user_id" => admin_user_id}
-    |> new()
-    |> Oban.insert()
-  end
-
   @impl Oban.Worker
   def timeout(_job) do
     Pleroma.Config.get([:workers, :timeout, :backup], :timer.minutes(1))
   end
 
+  @spec schedule_deletion(Backup.t()) ::
+          {:error, any} | {:ok, Oban.Job.t()}
   def schedule_deletion(backup) do
     days = Pleroma.Config.get([Backup, :purge_after_days])
     time = 60 * 60 * 24 * days
     scheduled_at = Calendar.NaiveDateTime.add!(backup.inserted_at, time)
 
-    %{"op" => "delete", "backup_id" => backup.id}
+    enqueue("delete", %{"backup_id" => backup.id}, scheduled_at: scheduled_at)
-    |> new(scheduled_at: scheduled_at)
-    |> Oban.insert()
   end
 
   def delete(backup) do
-    %{"op" => "delete", "backup_id" => backup.id}
+    enqueue("delete", %{"backup_id" => backup.id})
-    |> new()
-    |> Oban.insert()
   end
 
   @impl true

lib/pleroma/workers/purge_expired_activity.ex

@@ -7,23 +7,20 @@ defmodule Pleroma.Workers.PurgeExpiredActivity do
   Worker which purges expired activity.
   """
 
-  use Oban.Worker, queue: :activity_expiration, max_attempts: 1, unique: [period: :infinity]
+  use Pleroma.Workers.WorkerHelper,
+    queue: "activity_expiration",
+    max_attempts: 1,
+    unique: [period: :infinity]
 
   import Ecto.Query
 
   alias Pleroma.Activity
 
-  @spec enqueue(map()) ::
+  def schedule(args) do
-          {:ok, Oban.Job.t()}
-          | {:error, :expired_activities_disabled}
-          | {:error, :expiration_too_close}
-  def enqueue(args) do
     with true <- enabled?() do
       {scheduled_at, args} = Map.pop(args, :expires_at)
 
-      args
+      enqueue("delete", args, scheduled_at: scheduled_at)
-      |> new(scheduled_at: scheduled_at)
-      |> Oban.insert()
     end
   end
 

lib/pleroma/workers/purge_expired_filter.ex

@@ -7,21 +7,20 @@ defmodule Pleroma.Workers.PurgeExpiredFilter do
   Worker which purges expired filters
   """
 
-  use Oban.Worker, queue: :filter_expiration, max_attempts: 1, unique: [period: :infinity]
+  use Pleroma.Workers.WorkerHelper,
+    queue: "filter_expiration",
+    max_attempts: 1,
+    unique: [period: :infinity]
 
   import Ecto.Query
 
   alias Oban.Job
   alias Pleroma.Repo
 
-  @spec enqueue(%{filter_id: integer(), expires_at: DateTime.t()}) ::
+  def schedule(args) do
-          {:ok, Job.t()} | {:error, Ecto.Changeset.t()}
-  def enqueue(args) do
     {scheduled_at, args} = Map.pop(args, :expires_at)
 
-    args
+    enqueue("delete", args, scheduled_at: scheduled_at)
-    |> new(scheduled_at: scheduled_at)
-    |> Oban.insert()
   end
 
   @impl Oban.Worker

lib/pleroma/workers/purge_expired_token.ex

@@ -7,16 +7,12 @@ defmodule Pleroma.Workers.PurgeExpiredToken do
   Worker which purges expired OAuth tokens
   """
 
-  use Oban.Worker, queue: :token_expiration, max_attempts: 1
+  use Pleroma.Workers.WorkerHelper, queue: "token_expiration", max_attempts: 1
 
-  @spec enqueue(%{token_id: integer(), valid_until: DateTime.t(), mod: module()}) ::
+  def schedule(args) do
-          {:ok, Oban.Job.t()} | {:error, Ecto.Changeset.t()}
-  def enqueue(args) do
     {scheduled_at, args} = Map.pop(args, :valid_until)
 
-    args
+    enqueue("delete", args, scheduled_at: scheduled_at)
-    |> __MODULE__.new(scheduled_at: scheduled_at)
-    |> Oban.insert()
   end
 
   @impl Oban.Worker

lib/pleroma/workers/worker_helper.ex

@@ -25,30 +25,55 @@ def sidekiq_backoff(attempt, pow \\ 4, base_backoff \\ 15) do
   defmacro __using__(opts) do
     caller_module = __CALLER__.module
     queue = Keyword.fetch!(opts, :queue)
+    queue_system = Config.get([:queue, :module])
 
-    quote do
+    case queue_system do
-      # Note: `max_attempts` is intended to be overridden in `new/2` call
+      Oban ->
-      use Oban.Worker,
+        quote do
-        queue: unquote(queue),
+          # Note: `max_attempts` is intended to be overridden in `new/2` call
-        max_attempts: 1
+          use Oban.Worker,
+            queue: unquote(queue),
+            max_attempts: 1
 
-      alias Oban.Job
+          alias Oban.Job
 
-      def enqueue(op, params, worker_args \\ []) do
+          def enqueue(op, params, worker_args \\ []) do
-        params = Map.merge(%{"op" => op}, params)
+            params = Map.merge(%{"op" => op}, params)
-        queue_atom = String.to_atom(unquote(queue))
+            queue_atom = String.to_atom(unquote(queue))
-        worker_args = worker_args ++ WorkerHelper.worker_args(queue_atom)
+            worker_args = worker_args ++ WorkerHelper.worker_args(queue_atom)
 
-        unquote(caller_module)
+            unquote(caller_module)
-        |> apply(:new, [params, worker_args])
+            |> apply(:new, [params, worker_args])
-        |> Oban.insert()
+            |> Oban.insert()
-      end
+          end
 
-      @impl Oban.Worker
+          @impl Oban.Worker
-      def timeout(_job) do
+          def timeout(_job) do
-        queue_atom = String.to_atom(unquote(queue))
+            queue_atom = String.to_atom(unquote(queue))
-        Config.get([:workers, :timeout, queue_atom], :timer.minutes(1))
+            Config.get([:workers, :timeout, queue_atom], :timer.minutes(1))
-      end
+          end
+        end
+
+      Pleroma.Broadway ->
+        quote do
+          @topic unquote(queue)
+          use Oban.Worker,
+            queue: unquote(queue),
+            max_attempts: 1
+
+          alias Oban.Job
+
+          def enqueue(op, params, worker_args \\ []) do
+            worker = to_string(__MODULE__)
+
+            params =
+              params
+              |> Map.put("__module__", worker)
+              |> Map.put("op", op)
+
+            Pleroma.Broadway.produce(unquote(queue), Jason.encode!(params), worker_args)
+          end
+        end
     end
   end
 end

mix.exs

@@ -190,6 +190,8 @@ defp deps do
        git: "https://akkoma.dev/AkkomaGang/mfm-parser.git",
        ref: "912fba81152d4d572e457fd5427f9875b2bc3dbe"},
       {:poison, ">= 0.0.0"},
+      {:broadway, "~> 1.0"},
+      {:broadway_rabbitmq, "~> 0.7"},
 
       ## dev & test
       {:ex_doc, "~> 0.22", only: :dev, runtime: false},

mix.lock

@@ -1,8 +1,12 @@
 %{
+  "amqp": {:hex, :amqp, "3.2.0", "51e85e06e4d283d98f21dce95906e42cb181fc7ceeb967b33b3be73d6ea41aa5", [:mix], [{:amqp_client, "~> 3.9.1", [hex: :amqp_client, repo: "hexpm", optional: false]}], "hexpm", "1439570336df6e79000239938fb055a0944dc9a768b4dec0af1375404508a014"},
+  "amqp_client": {:hex, :amqp_client, "3.9.27", "98e8ecb4e3717b6937c941c93874e284ca219a4bc5169bc7a1602fcf77833e3d", [:make, :rebar3], [{:rabbit_common, "3.9.27", [hex: :rabbit_common, repo: "hexpm", optional: false]}], "hexpm", "12f5ca7252b5ac70944e10e682defa11c67f17d0311911ed2860e95bc0dbcd75"},
   "base62": {:hex, :base62, "1.2.2", "85c6627eb609317b70f555294045895ffaaeb1758666ab9ef9ca38865b11e629", [:mix], [{:custom_base, "~> 0.2.1", [hex: :custom_base, repo: "hexpm", optional: false]}], "hexpm", "d41336bda8eaa5be197f1e4592400513ee60518e5b9f4dcf38f4b4dae6f377bb"},
   "bbcode_pleroma": {:hex, :bbcode_pleroma, "0.2.0", "d36f5bca6e2f62261c45be30fa9b92725c0655ad45c99025cb1c3e28e25803ef", [:mix], [{:nimble_parsec, "~> 0.5", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "19851074419a5fedb4ef49e1f01b30df504bb5dbb6d6adfc135238063bebd1c3"},
   "bcrypt_elixir": {:hex, :bcrypt_elixir, "2.3.1", "5114d780459a04f2b4aeef52307de23de961b69e13a5cd98a911e39fda13f420", [:make, :mix], [{:comeonin, "~> 5.3", [hex: :comeonin, repo: "hexpm", optional: false]}, {:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}], "hexpm", "42182d5f46764def15bf9af83739e3bf4ad22661b1c34fc3e88558efced07279"},
   "benchee": {:hex, :benchee, "1.1.0", "f3a43817209a92a1fade36ef36b86e1052627fd8934a8b937ac9ab3a76c43062", [:mix], [{:deep_merge, "~> 1.0", [hex: :deep_merge, repo: "hexpm", optional: false]}, {:statistex, "~> 1.0", [hex: :statistex, repo: "hexpm", optional: false]}], "hexpm", "7da57d545003165a012b587077f6ba90b89210fd88074ce3c60ce239eb5e6d93"},
+  "broadway": {:hex, :broadway, "1.0.5", "54593b1ed00b0d63c00aca8583697d0f8a8bb59320b67a8e5fe61b82069940e7", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.7 or ~> 0.4.0 or ~> 0.5.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "4fd031f5a670daa48b507c80bbc7830ffc66f91ec14951e4bf2e13d4bd325f51"},
+  "broadway_rabbitmq": {:hex, :broadway_rabbitmq, "0.7.2", "d2cc2cecf893e2a8646e74670827e4ee4240c1482473c511c3dd2c18b982728f", [:mix], [{:amqp, "~> 1.3 or ~> 2.0 or ~> 3.0", [hex: :amqp, repo: "hexpm", optional: false]}, {:broadway, "~> 1.0", [hex: :broadway, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.3.5 or ~> 0.4.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "e3a1d27b452cefc1da96bc15e12b34ff5fe3c9b2ada1341e6cdcae7f71f81383"},
   "bunt": {:hex, :bunt, "0.2.1", "e2d4792f7bc0ced7583ab54922808919518d0e57ee162901a16a1b6664ef3b14", [:mix], [], "hexpm", "a330bfb4245239787b15005e66ae6845c9cd524a288f0d141c148b02603777a5"},
   "cachex": {:hex, :cachex, "3.4.0", "868b2959ea4aeb328c6b60ff66c8d5123c083466ad3c33d3d8b5f142e13101fb", [:mix], [{:eternal, "~> 1.2", [hex: :eternal, repo: "hexpm", optional: false]}, {:jumper, "~> 1.0", [hex: :jumper, repo: "hexpm", optional: false]}, {:sleeplocks, "~> 1.1", [hex: :sleeplocks, repo: "hexpm", optional: false]}, {:unsafe, "~> 1.0", [hex: :unsafe, repo: "hexpm", optional: false]}], "hexpm", "370123b1ab4fba4d2965fb18f87fd758325709787c8c5fce35b3fe80645ccbe5"},
   "calendar": {:hex, :calendar, "1.0.0", "f52073a708528482ec33d0a171954ca610fe2bd28f1e871f247dc7f1565fa807", [:mix], [{:tzdata, "~> 0.1.201603 or ~> 0.5.20 or ~> 1.0", [hex: :tzdata, repo: "hexpm", optional: false]}], "hexpm", "990e9581920c82912a5ee50e62ff5ef96da6b15949a2ee4734f935fdef0f0a6f"},
@@ -17,6 +21,7 @@
   "cowboy": {:hex, :cowboy, "2.9.0", "865dd8b6607e14cf03282e10e934023a1bd8be6f6bacf921a7e2a96d800cd452", [:make, :rebar3], [{:cowlib, "2.11.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "2c729f934b4e1aa149aff882f57c6372c15399a20d54f65c8d67bef583021bde"},
   "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.3.1", "ebd1a1d7aff97f27c66654e78ece187abdc646992714164380d8a041eda16754", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "3a6efd3366130eab84ca372cbd4a7d3c3a97bdfcfb4911233b035d117063f0af"},
   "cowlib": {:hex, :cowlib, "2.11.0", "0b9ff9c346629256c42ebe1eeb769a83c6cb771a6ee5960bd110ab0b9b872063", [:make, :rebar3], [], "hexpm", "2b3e9da0b21c4565751a6d4901c20d1b4cc25cbb7fd50d91d2ab6dd287bc86a9"},
+  "credentials_obfuscation": {:hex, :credentials_obfuscation, "3.1.0", "2c405ea0c5db7b3344aa5a99f86c33e7b6ecea97d2cb613371e1cf0d192ef2c6", [:rebar3], [], "hexpm", "04884e62b1c6cdfba999d4d6b3e99bc0a59d5e439517bc5c01767255afb7b778"},
   "credo": {:git, "https://github.com/rrrene/credo.git", "1c1b99ea41a457761383d81aaf6a606913996fe7", [ref: "1c1b99ea41a457761383d81aaf6a606913996fe7"]},
   "crypt": {:git, "https://github.com/msantos/crypt.git", "f75cd55325e33cbea198fb41fe41871392f8fb76", [ref: "f75cd55325e33cbea198fb41fe41871392f8fb76"]},
   "custom_base": {:hex, :custom_base, "0.2.1", "4a832a42ea0552299d81652aa0b1f775d462175293e99dfbe4d7dbaab785a706", [:mix], [], "hexpm", "8df019facc5ec9603e94f7270f1ac73ddf339f56ade76a721eaa57c1493ba463"},
@@ -49,6 +54,7 @@
   "flake_id": {:hex, :flake_id, "0.1.0", "7716b086d2e405d09b647121a166498a0d93d1a623bead243e1f74216079ccb3", [:mix], [{:base62, "~> 1.2", [hex: :base62, repo: "hexpm", optional: false]}, {:ecto, ">= 2.0.0", [hex: :ecto, repo: "hexpm", optional: true]}], "hexpm", "31fc8090fde1acd267c07c36ea7365b8604055f897d3a53dd967658c691bd827"},
   "floki": {:hex, :floki, "0.34.0", "002d0cc194b48794d74711731db004fafeb328fe676976f160685262d43706a8", [:mix], [], "hexpm", "9c3a9f43f40dde00332a589bd9d389b90c1f518aef500364d00636acc5ebc99c"},
   "gen_smtp": {:hex, :gen_smtp, "0.15.0", "9f51960c17769b26833b50df0b96123605a8024738b62db747fece14eb2fbfcc", [:rebar3], [], "hexpm", "29bd14a88030980849c7ed2447b8db6d6c9278a28b11a44cafe41b791205440f"},
+  "gen_stage": {:hex, :gen_stage, "1.1.2", "b1656cd4ba431ed02c5656fe10cb5423820847113a07218da68eae5d6a260c23", [:mix], [], "hexpm", "9e39af23140f704e2b07a3e29d8f05fd21c2aaf4088ff43cb82be4b9e3148d02"},
   "gettext": {:hex, :gettext, "0.20.0", "75ad71de05f2ef56991dbae224d35c68b098dd0e26918def5bb45591d5c8d429", [:mix], [], "hexpm", "1c03b177435e93a47441d7f681a7040bd2a816ece9e2666d1c9001035121eb3d"},
   "hackney": {:hex, :hackney, "1.18.1", "f48bf88f521f2a229fc7bae88cf4f85adc9cd9bcf23b5dc8eb6a1788c662c4f6", [:rebar3], [{:certifi, "~> 2.9.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.3.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "a4ecdaff44297e9b5894ae499e9a070ea1888c84afdd1fd9b7b2bc384950128e"},
   "hpax": {:hex, :hpax, "0.1.2", "09a75600d9d8bbd064cdd741f21fc06fc1f4cf3d0fcc335e5aa19be1a7235c84", [:mix], [], "hexpm", "2c87843d5a23f5f16748ebe77969880e29809580efdaccd615cd3bed628a8c13"},
@@ -60,6 +66,7 @@
   "jason": {:hex, :jason, "1.4.0", "e855647bc964a44e2f67df589ccf49105ae039d4179db7f6271dfd3843dc27e6", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "79a3791085b2a0f743ca04cec0f7be26443738779d09302e01318f97bdb82121"},
   "joken": {:hex, :joken, "2.5.0", "09be497d804b8115eb6f07615cef2e60c2a1008fb89dc0aef0d4c4b4609b99aa", [:mix], [{:jose, "~> 1.11.2", [hex: :jose, repo: "hexpm", optional: false]}], "hexpm", "22b25c89617c5ed8ca7b31026340a25ea0f9ca7160f9706b79be9ed81fdf74e7"},
   "jose": {:hex, :jose, "1.11.2", "f4c018ccf4fdce22c71e44d471f15f723cb3efab5d909ab2ba202b5bf35557b3", [:mix, :rebar3], [], "hexpm", "98143fbc48d55f3a18daba82d34fe48959d44538e9697c08f34200fa5f0947d2"},
+  "jsx": {:hex, :jsx, "3.1.0", "d12516baa0bb23a59bb35dccaf02a1bd08243fcbb9efe24f2d9d056ccff71268", [:rebar3], [], "hexpm", "0c5cc8fdc11b53cc25cf65ac6705ad39e54ecc56d1c22e4adb8f5a53fb9427f3"},
   "jumper": {:hex, :jumper, "1.0.1", "3c00542ef1a83532b72269fab9f0f0c82bf23a35e27d278bfd9ed0865cecabff", [:mix], [], "hexpm", "318c59078ac220e966d27af3646026db9b5a5e6703cb2aa3e26bcfaba65b7433"},
   "linkify": {:git, "https://akkoma.dev/AkkomaGang/linkify.git", "2567e2c1073fa371fd26fd66dfa5bc77b6919c16", [branch: "bugfix/line-ending-buffer"]},
   "mail": {:hex, :mail, "0.2.3", "2c6bb5f8a5f74845fa50ecd0fb45ea16b164026f285f45104f1c4c078cd616d4", [:mix], [], "hexpm", "932b398fa9c69fdf290d7ff63175826e0f1e24414d5b0763bb00a2acfc6c6bf5"},
@@ -76,7 +83,7 @@
   "mock": {:hex, :mock, "0.3.7", "75b3bbf1466d7e486ea2052a73c6e062c6256fb429d6797999ab02fa32f29e03", [:mix], [{:meck, "~> 0.9.2", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm", "4da49a4609e41fd99b7836945c26f373623ea968cfb6282742bcb94440cf7e5c"},
   "mogrify": {:hex, :mogrify, "0.9.2", "b360984adea7dd6a55f18028e6327973c58de7f548fdb86c9859848aa904d5b0", [:mix], [], "hexpm", "c18d10fd70ca20e2585301616c89f6e4f7159d92efc9cc8ee579e00c886f699d"},
   "mox": {:hex, :mox, "1.0.2", "dc2057289ac478b35760ba74165b4b3f402f68803dd5aecd3bfd19c183815d64", [:mix], [], "hexpm", "f9864921b3aaf763c8741b5b8e6f908f44566f1e427b2630e89e9a73b981fef2"},
-  "nimble_options": {:hex, :nimble_options, "0.5.1", "5c166f7669e40333191bea38e3bd3811cc13f459f1e4be49e89128a21b5d8c4d", [:mix], [], "hexpm", "d176cf7baa4fef0ceb301ca3eb8b55bd7de3e45f489c4f8b4f2849f1f114ef3e"},
+  "nimble_options": {:hex, :nimble_options, "0.4.0", "c89babbab52221a24b8d1ff9e7d838be70f0d871be823165c94dd3418eea728f", [:mix], [], "hexpm", "e6701c1af326a11eea9634a3b1c62b475339ace9456c1a23ec3bc9a847bca02d"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.2.3", "244836e6e3f1200c7f30cb56733fd808744eca61fd182f731eac4af635cc6d0b", [:mix], [], "hexpm", "c8d789e39b9131acf7b99291e93dae60ab48ef14a7ee9d58c6964f59efb570b0"},
   "nimble_pool": {:hex, :nimble_pool, "0.2.6", "91f2f4c357da4c4a0a548286c84a3a28004f68f05609b4534526871a22053cde", [:mix], [], "hexpm", "1c715055095d3f2705c4e236c18b618420a35490da94149ff8b580a2144f653f"},
   "oban": {:hex, :oban, "2.12.1", "f604d7e6a8be9fda4a9b0f6cebbd633deba569f85dbff70c4d25d99a6f023177", [:mix], [{:ecto_sql, "~> 3.6", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "9b1844c2b74e0d788b73e5144b0c9d5674cb775eae29d88a36f3c3b48d42d058"},
@@ -99,6 +106,7 @@
   "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"},
   "postgrex": {:hex, :postgrex, "0.16.5", "fcc4035cc90e23933c5d69a9cd686e329469446ef7abba2cf70f08e2c4b69810", [:mix], [{:connection, "~> 1.1", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "edead639dc6e882618c01d8fc891214c481ab9a3788dfe38dd5e37fd1d5fb2e8"},
   "pot": {:hex, :pot, "1.0.2", "13abb849139fdc04ab8154986abbcb63bdee5de6ed2ba7e1713527e33df923dd", [:rebar3], [], "hexpm", "78fe127f5a4f5f919d6ea5a2a671827bd53eb9d37e5b4128c0ad3df99856c2e0"},
+  "rabbit_common": {:hex, :rabbit_common, "3.9.27", "f0d152b7b400cf531e0fc23d4c7b2442da95a384da1429f8c6cebc8ba8433426", [:make, :rebar3], [{:credentials_obfuscation, "3.1.0", [hex: :credentials_obfuscation, repo: "hexpm", optional: false]}, {:jsx, "3.1.0", [hex: :jsx, repo: "hexpm", optional: false]}, {:recon, "2.5.2", [hex: :recon, repo: "hexpm", optional: false]}], "hexpm", "6a696026ad237d9c89afcaff80b3dc9e71f97938b3fd39cea15bfebd6eec2648"},
   "ranch": {:hex, :ranch, "1.8.0", "8c7a100a139fd57f17327b6413e4167ac559fbc04ca7448e9be9057311597a1d", [:make, :rebar3], [], "hexpm", "49fbcfd3682fab1f5d109351b61257676da1a2fdbe295904176d5e521a2ddfe5"},
   "recon": {:hex, :recon, "2.5.2", "cba53fa8db83ad968c9a652e09c3ed7ddcc4da434f27c3eaa9ca47ffb2b1ff03", [:mix, :rebar3], [], "hexpm", "2c7523c8dee91dff41f6b3d63cba2bd49eb6d2fe5bf1eec0df7f87eb5e230e1c"},
   "remote_ip": {:hex, :remote_ip, "1.1.0", "cb308841595d15df3f9073b7c39243a1dd6ca56e5020295cb012c76fbec50f2d", [:mix], [{:combine, "~> 0.10", [hex: :combine, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "616ffdf66aaad6a72fc546dabf42eed87e2a99e97b09cbd92b10cc180d02ed74"},