hare_mq

defmodule HareMq.Connection do
  @moduledoc """
  GenServer that manages a single AMQP connection.

  ## Lifecycle

  On start it immediately sends itself a `:connect` message. If the broker is
  unavailable or misconfigured it logs the error and retries after
  `reconnect_interval_ms` (from `config :hare_mq, :configuration`;
  default 10 000 ms). All configuration is read at runtime so changes via
  `Application.put_env` take effect on the next reconnect.

  The underlying `AMQP.Connection` is monitored. On a clean broker-initiated
  close (status 200) the GenServer goes idle. On any unexpected drop it logs
  and schedules a reconnect.

  ## Named instances (multi-vhost)

  By default the process registers as `{:global, HareMq.Connection}`. Pass
  `name:` to `start_link/1` to run multiple independent connections:

      {HareMq.Connection, name: {:global, :conn_vhost_b}}

  Consumers and publishers accept a matching `connection_name:` option.
  """
  use GenServer
  use AMQP
  require Logger

  defp reconnect_interval,
    do:
      (Application.get_env(:hare_mq, :configuration) || [])[:reconnect_interval_ms] || 10_000

  # Exponential backoff capped at 60 s with per-attempt jitter to avoid
  # thundering-herd reconnection storms after an outage.
  defp backoff_delay(attempt) do
    base = reconnect_interval()
    delay = trunc(base * :math.pow(2, min(attempt - 1, 5)))
    jitter = :rand.uniform(max(base, 1))
    min(delay + jitter, 60_000)
  end

  def start_link(opts \\ []) do
    opts = opts || []
    name = Keyword.get(opts, :name, {:global, __MODULE__})
    GenServer.start_link(__MODULE__, name, name: name)
    |> HareMq.CodeFlow.successful_start()
  end

  def init(name) do
    send(self(), :connect)
    {:ok, %{name: name, conn: nil, attempt: 0}}
  end

  @doc """
  Returns the current AMQP connection.

  Accepts an optional `name` argument (default `{:global, HareMq.Connection}`) to
  query a specific named instance.

  Returns:
  - `{:ok, %AMQP.Connection{}}` when connected.
  - `{:error, :not_connected}` when the GenServer is not running or not yet connected.
  """
  def get_connection(name \\ {:global, __MODULE__}) do
    case GenServer.whereis(name) do
      nil ->
        {:error, :not_connected}

      pid ->
        case GenServer.call(pid, :get_connection, 1_000) do
          %AMQP.Connection{} = conn -> {:ok, conn}
          _ -> {:error, :not_connected}
        end
    end
  end

  @doc """
  Closes the current AMQP connection.

  Accepts an optional `name` argument (default `{:global, HareMq.Connection}`).
  Sets internal state to `nil` so subsequent `get_connection/1` calls return
  `{:error, :not_connected}` immediately.

  Returns:
  - `{:ok, %AMQP.Connection{}}` when the connection was open and is now closed.
  - `{:error, :not_connected}` when there was no active connection.
  """
  def close_connection(name \\ {:global, __MODULE__}) do
    case GenServer.call(name, :close_connection) do
      nil -> {:error, :not_connected}
      %AMQP.Connection{} = conn -> {:ok, conn}
    end
  end

  def handle_call(:get_connection, _, %{conn: %AMQP.Connection{} = conn} = state) do
    {:reply, conn, state}
  end

  def handle_call(:get_connection, _, state) do
    {:reply, nil, state}
  end

  def handle_call(:close_connection, _, %{conn: %AMQP.Connection{} = conn} = state) do
    Connection.close(conn)
    {:reply, conn, %{state | conn: nil}}
  end

  def handle_call(:close_connection, _, state) do
    {:reply, nil, state}
  end

  def handle_info(:connect, state) do
    configs = Application.get_env(:hare_mq, :amqp)
    attempt = state.attempt + 1

    case configs[:url] do
      nil ->
        delay = backoff_delay(attempt)
        Logger.error("[connection] Missing :amqp config. Retrying in #{delay}ms (attempt #{attempt})...")
        :telemetry.execute(
          [:hare_mq, :connection, :reconnecting],
          %{retry_delay_ms: delay},
          %{connection_name: state.name, reason: :missing_config}
        )
        Process.send_after(self(), :connect, delay)
        {:noreply, %{state | attempt: attempt}}

      host ->
        case Connection.open(host) do
          {:ok, %AMQP.Connection{} = conn} ->
            Process.monitor(conn.pid)
            :telemetry.execute(
              [:hare_mq, :connection, :connected],
              %{system_time: System.system_time()},
              %{connection_name: state.name, host: host}
            )
            {:noreply, %{state | conn: conn, attempt: 0}}

          {:error, reason} ->
            delay = backoff_delay(attempt)
            Logger.error("[connection] Failed to connect #{host}. Reconnecting in #{delay}ms (attempt #{attempt})...")
            :telemetry.execute(
              [:hare_mq, :connection, :reconnecting],
              %{retry_delay_ms: delay},
              %{connection_name: state.name, host: host, reason: reason}
            )
            Process.send_after(self(), :connect, delay)
            {:noreply, %{state | attempt: attempt}}
        end
    end
  end

  def handle_info({:DOWN, _, :process, _pid, {:shutdown, :normal}}, state) do
    {:noreply, %{state | conn: nil}}
  end

  def handle_info({:DOWN, _, :process, _pid, {:shutdown, {:server_initiated_close, 200, _}}}, state) do
    {:noreply, %{state | conn: nil}}
  end

  def handle_info({:DOWN, _, :process, _pid, reason}, state) do
    Logger.error("[connection] Connection lost: #{inspect(reason)}. Reconnecting...")
    :telemetry.execute(
      [:hare_mq, :connection, :disconnected],
      %{system_time: System.system_time()},
      %{connection_name: state.name, reason: reason}
    )
    # Reset attempt counter — first reconnect after a real disconnect should be prompt.
    Process.send_after(self(), :connect, reconnect_interval())
    {:noreply, %{state | conn: nil, attempt: 0}}
  end

  def handle_info(reason, state) do
    # Stop GenServer. Will be restarted by Supervisor.
    {:stop, {:connection_lost, reason}, state}
  end
end