ulfnet_amqpx

defmodule AMQPX.Receiver.Standard.Retry do
  defstruct [
    :table,
    :limit,
    :identity,
    :delay
  ]

  @doc """
  Retry options.

  `identity` specifies the list of methods to generate a unique term for a
  message. The first non-`nil` result is used. If all methods evaluate to
  `nil`, retry tracking is not used for that message.

  `delay` specifies the time to wait (in milliseconds) before rejecting the
  delivery, to prevent a hot retry loop.
  """
  @type option ::
          {:limit, integer()}
          | {:identity, [AMQPX.MessageIdentity.identity_function()]}
          | {:delay, nil | integer()}

  def init(nil), do: nil

  def init(opts) do
    table = :ets.new(__MODULE__, [:public])

    limit = opts |> Keyword.fetch!(:limit)
    identity = opts |> Keyword.get(:identity, [:message_id, :payload_hash])
    delay = opts |> Keyword.get(:delay)

    %__MODULE__{
      table: table,
      limit: limit,
      identity: identity,
      delay: delay
    }
  end

  def exhausted?(payload, meta, handler, state)
  def exhausted?(_, _, _, nil), do: false

  def exhausted?(
        payload,
        meta,
        handler,
        %__MODULE__{table: table, limit: limit, identity: identity}
      ) do
    case AMQPX.MessageIdentity.get(payload, meta, handler, identity) do
      nil ->
        false

      key ->
        seen_times =
          case :ets.lookup(table, key) do
            [] -> 0
            [{^key, n}] -> n
          end

        # A retry limit of 0 allows us to see a message once. A retry limit
        # of 1 lets us see a message twice, and so on.
        if seen_times > limit do
          true
        else
          :ets.insert(table, {key, seen_times + 1})
          false
        end
    end
  end

  def delay(state)
  def delay(%__MODULE__{delay: delay}) when is_integer(delay), do: Process.sleep(delay)
  def delay(_), do: nil

  def clear(payload, meta, handler, state)
  def clear(_, _, _, nil), do: nil

  def clear(payload, meta, handler, %__MODULE__{table: table, identity: identity}) do
    case AMQPX.MessageIdentity.get(payload, meta, handler, identity) do
      nil ->
        nil

      key ->
        :ets.delete(table, key)
    end
  end
end