aerospike_driver

defmodule Aerospike.CircuitBreaker do
  @moduledoc false

  alias Aerospike.Error
  alias Aerospike.Tables

  @type conn_name :: atom()
  @type node_name :: String.t()

  @increment_event [:aerospike, :circuit_breaker, :increment]
  @reject_event [:aerospike, :circuit_breaker, :reject]
  @reset_event [:aerospike, :circuit_breaker, :reset]

  @doc """
  Returns whether the circuit breaker is enabled for the connection.

  Breaker state is driven by the `:breaker_config` row in the meta ETS table:
  `%{max_error_rate: non_neg_integer, error_rate_window: pos_integer}`.

  Missing or malformed config intentionally disables breaker logic so request paths
  never crash due to partial startup state or transient table inconsistencies.
  """
  @spec enabled?(conn_name()) :: boolean()
  def enabled?(conn) when is_atom(conn) do
    case breaker_config(conn) do
      {:ok, %{max_error_rate: max_error_rate}} -> max_error_rate > 0
      :error -> false
    end
  end

  @doc """
  Performs the request-path threshold check for a node.

  Rejection boundary is `count >= max_error_rate` (inclusive), so a node is
  rejected as soon as it reaches the configured threshold. This matches the
  fixed-threshold semantics used by the cluster tend window.

  Emits `[:aerospike, :circuit_breaker, :reject]` on rejection with:
  - measurements: `%{count: count, max_error_rate: threshold}`
  - metadata: `%{conn: conn, node: node_name}`
  """
  @spec allow_request?(conn_name(), node_name()) :: :ok | {:error, Error.t()}
  def allow_request?(conn, node_name) when is_atom(conn) and is_binary(node_name) do
    if enabled?(conn) do
      {:ok, %{max_error_rate: max_error_rate}} = breaker_config(conn)
      count = error_count(conn, node_name)
      maybe_reject(conn, node_name, count, max_error_rate)
    else
      :ok
    end
  end

  @doc """
  Atomically increments the per-node error counter for breaker tracking.

  Uses `:ets.update_counter/4` with row shape `{node_name, count}` so increments
  remain lock-free and safe across concurrent request processes.

  The counter is additive within the current tend window and is only cleared by
  `maybe_reset_window/3`. It is never decremented in request paths.

  Emits `[:aerospike, :circuit_breaker, :increment]` with:
  - measurements: `%{count: updated_count}`
  - metadata: `%{conn: conn, node: node_name, reason: reason}`

  The `reason` metadata is part of the maintainer-facing contract used to
  differentiate transport and server-overload signals in telemetry handlers.
  """
  @spec record_error(conn_name(), node_name(), atom()) :: :ok
  def record_error(conn, node_name, reason)
      when is_atom(conn) and is_binary(node_name) and is_atom(reason) do
    if enabled?(conn) do
      count = :ets.update_counter(Tables.breaker(conn), node_name, {2, 1}, {node_name, 0})
      emit_increment(conn, node_name, reason, count)
    end

    :ok
  end

  @doc """
  Resets all node counters at the configured tend-window boundary.

  Reset happens when `rem(tend_tick, error_rate_window) == 0`, matching the
  cluster tend cadence so windows are consistent across all request paths.

  Emits `[:aerospike, :circuit_breaker, :reset]` for each node whose
  `count_before_reset` is greater than zero.
  """
  @spec maybe_reset_window(conn_name(), non_neg_integer(), pos_integer()) :: :ok
  def maybe_reset_window(conn, tend_tick, error_rate_window)
      when is_atom(conn) and is_integer(tend_tick) and tend_tick >= 0 and
             is_integer(error_rate_window) and error_rate_window > 0 do
    if enabled?(conn) and rem(tend_tick, error_rate_window) == 0 do
      reset_window(conn)
    end

    :ok
  end

  defp maybe_reject(conn, node_name, count, max_error_rate) when count >= max_error_rate do
    emit_reject(conn, node_name, count, max_error_rate)
    {:error, Error.from_result_code(:max_error_rate, node: node_name)}
  end

  defp maybe_reject(_conn, _node_name, _count, _max_error_rate), do: :ok

  # Config reads are intentionally total: malformed or missing rows return :error
  # so public request-path functions can safely degrade to breaker-disabled mode.
  defp breaker_config(conn) do
    case :ets.lookup(Tables.meta(conn), :breaker_config) do
      [{:breaker_config, %{max_error_rate: max_error_rate, error_rate_window: error_rate_window}}]
      when is_integer(max_error_rate) and max_error_rate >= 0 and is_integer(error_rate_window) and
             error_rate_window > 0 ->
        {:ok, %{max_error_rate: max_error_rate, error_rate_window: error_rate_window}}

      _ ->
        :error
    end
  end

  # Unknown rows are treated as count 0 to avoid negative coupling with node
  # discovery order and to keep pre-checks idempotent during topology churn.
  defp error_count(conn, node_name) do
    case :ets.lookup(Tables.breaker(conn), node_name) do
      [{^node_name, count}] when is_integer(count) and count >= 0 -> count
      _ -> 0
    end
  end

  # Emit reset events before deleting rows so observers receive the final
  # window totals for each node.
  defp reset_window(conn) do
    conn
    |> Tables.breaker()
    |> :ets.tab2list()
    |> Enum.each(&emit_reset(conn, &1))

    :ets.delete_all_objects(Tables.breaker(conn))
  end

  defp emit_increment(conn, node_name, reason, count) do
    :telemetry.execute(@increment_event, %{count: count}, %{
      conn: conn,
      node: node_name,
      reason: reason
    })
  end

  defp emit_reject(conn, node_name, count, max_error_rate) do
    :telemetry.execute(
      @reject_event,
      %{count: count, max_error_rate: max_error_rate},
      %{conn: conn, node: node_name}
    )
  end

  defp emit_reset(conn, {node_name, count})
       when is_binary(node_name) and is_integer(count) and count > 0 do
    :telemetry.execute(
      @reset_event,
      %{count_before_reset: count},
      %{conn: conn, node: node_name}
    )
  end

  defp emit_reset(_conn, _entry), do: :ok
end