+defmodule Pleroma.Web.Federator.RetryQueue do
+ use GenServer
+ alias Pleroma.Web.{WebFinger, Websub}
+ alias Pleroma.Web.ActivityPub.ActivityPub
+ require Logger
+
+ @websub Application.get_env(:pleroma, :websub)
+ @ostatus Application.get_env(:pleroma, :websub)
+ @httpoison Application.get_env(:pleroma, :websub)
+ @instance Application.get_env(:pleroma, :websub)
+ # initial timeout, 5 min
+ @initial_timeout 30_000
+ @max_retries 5
+
+ def init(args) do
+ {:ok, args}
+ end
+
+ def start_link() do
+ GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+ end
+
+ def enqueue(data, transport, retries \\ 0) do
+ GenServer.cast(__MODULE__, {:maybe_enqueue, data, transport, retries + 1})
+ end
+
+ def handle_cast({:maybe_enqueue, data, transport, retries}, state) do
+ if retries > @max_retries do
+ Logger.debug("Maximum retries reached on #{inspect(data)}")
+ {:noreply, state}
+ else
+ Process.send_after(
+ __MODULE__,
+ {:send, data, transport, retries},
+ growth_function(retries)
+ )
+
+ {:noreply, state}
+ end
+ end
+
+ def handle_info({:send, %{topic: topic} = data, :websub, retries}, state) do
+ Logger.debug("RetryQueue: Retrying to send object #{topic}")
+
+ case Websub.publish_one(data) do
+ {:ok, _} ->
+ {:noreply, state}
+
+ {:error, reason} ->
+ enqueue(data, :websub, retries)
+ {:noreply, state}
+ end
+ end
+
+ def handle_info({:send, %{id: id} = data, :activitypub, retries}, state) do
+ Logger.debug("RetryQueue: Retrying to send object #{id}")
+
+ case ActivityPub.publish_one(data) do
+ {:ok, _} ->
+ {:noreply, state}
+
+ {:error, reason} ->
+ enqueue(data, :activitypub, retries)
+ {:noreply, state}
+ end
+ end
+
+ def handle_info(unknown, state) do
+ Logger.debug("RetryQueue: don't know what to do with #{inspect(unknown)}, ignoring")
+ {:noreply, state}
+ end
+
+ defp growth_function(retries) do
+ round(@initial_timeout * :math.pow(retries, 3))
+ end
+end