aerospike_driver

defmodule Aerospike.Cluster.NodeSupervisor do
  @moduledoc false

  @default_pool_size 10

  # Aerospike's default `proto-fd-idle-ms` is 60_000 ms: the server
  # drops client sockets that sit idle longer than that. Evict at 55_000
  # so the client closes first and the next checkout sees a fresh
  # worker instead of a server-initiated RST.
  @default_idle_timeout_ms 55_000

  # Cap how many idle workers NimblePool drops per ping cycle so a big
  # pool cannot lose every worker in one tick. 2 matches the guidance
  # in NimblePool's docs and keeps at least some capacity hot through
  # a quiet period.
  @default_max_idle_pings 2

  @doc false
  def child_spec(opts) when is_list(opts) do
    name = Keyword.fetch!(opts, :name)

    %{
      id: {__MODULE__, name},
      start: {__MODULE__, :start_link, [opts]},
      type: :supervisor,
      restart: :permanent,
      shutdown: :infinity
    }
  end

  @doc """
  Starts the per-cluster `DynamicSupervisor`.

  Options:

    * `:name` — cluster name atom. The supervisor is registered under
      `sup_name/1`.
  """
  @spec start_link(keyword()) :: Supervisor.on_start()
  def start_link(opts) when is_list(opts) do
    name = Keyword.fetch!(opts, :name)

    is_atom(name) or
      raise ArgumentError, "Aerospike.Cluster.NodeSupervisor: :name must be an atom"

    DynamicSupervisor.start_link(strategy: :one_for_one, name: sup_name(name))
  end

  @doc """
  Returns the registered name atom used for the cluster's NodeSupervisor.
  """
  @spec sup_name(atom()) :: atom()
  def sup_name(name) when is_atom(name), do: :"#{name}_node_sup"

  @doc """
  Starts an `Aerospike.Cluster.NodePool` child under this supervisor.

  Options:

    * `:node_name` — string identifier of the Aerospike node (required).
    * `:transport` — module implementing `Aerospike.Cluster.NodeTransport`
      (required).
    * `:host` — host the pool workers connect to (required).
    * `:port` — TCP port the pool workers connect to (required).
    * `:connect_opts` — keyword list passed as the third argument to
      `transport.connect/3` (required).
    * `:pool_size` — positive integer, defaults to `#{@default_pool_size}`.
    * `:min_connections_per_node` — non-negative integer floor requested
      by startup policy. `0` starts workers lazily; positive values keep
      the existing eager pool behavior, which maintains at least that many
      reusable workers.
    * `:idle_timeout_ms` — milliseconds a worker may sit idle before
      NimblePool evicts it via `handle_ping/2`. Defaults to
      `#{@default_idle_timeout_ms}` to stay under Aerospike's default
      `proto-fd-idle-ms` of 60_000 ms.
    * `:max_idle_pings` — positive integer bounding how many idle
      workers NimblePool may drop per verification cycle. Defaults to
      `#{@default_max_idle_pings}`.
    * `:counters` — optional `Aerospike.Cluster.NodeCounters.t()` reference
      owned by the `Aerospike.Cluster.Tender`. When present, pool callbacks
      increment `:in_flight` on checkout and decrement on checkin /
      cancellation, and bump `:failed` when the caller signals a
      transport-class failure via `{:close, :failure}`. Omitting this
      key leaves counter writes as no-ops, which is the intended
      behaviour for tests and cluster-state-only modes.
    * `:cluster_name` — optional cluster identity atom used by internal
      runtime metrics hooks. When absent, connection metrics are skipped.
    * `:features` — optional `MapSet` of capability tokens captured
      from the node's `features` info-key reply at registration. Pool
      consumers (capability-gated send paths) read this to decide
      whether to enable optional features per node. Defaults to an
      empty `MapSet`.
  """
  @spec start_pool(pid() | atom(), keyword()) :: DynamicSupervisor.on_start_child()
  def start_pool(supervisor, opts) when is_list(opts) do
    node_name = Keyword.fetch!(opts, :node_name)
    transport = Keyword.fetch!(opts, :transport)
    host = Keyword.fetch!(opts, :host)
    port = Keyword.fetch!(opts, :port)
    connect_opts = Keyword.fetch!(opts, :connect_opts)
    pool_size = Keyword.get(opts, :pool_size, @default_pool_size)
    min_connections_per_node = Keyword.get(opts, :min_connections_per_node, pool_size)
    idle_timeout_ms = Keyword.get(opts, :idle_timeout_ms, @default_idle_timeout_ms)
    max_idle_pings = Keyword.get(opts, :max_idle_pings, @default_max_idle_pings)
    counters = Keyword.get(opts, :counters)
    cluster_name = Keyword.get(opts, :cluster_name)
    features = Keyword.get(opts, :features, MapSet.new())

    worker_opts =
      [
        transport: transport,
        host: host,
        port: port,
        connect_opts: connect_opts,
        node_name: node_name,
        counters: counters,
        cluster_name: cluster_name,
        features: features
      ]

    lazy? = min_connections_per_node == 0

    child = %{
      id: {:node_pool, node_name},
      start:
        {NimblePool, :start_link,
         [
           [
             worker: {Aerospike.Cluster.NodePool, worker_opts},
             pool_size: pool_size,
             lazy: lazy?,
             worker_idle_timeout: idle_timeout_ms,
             max_idle_pings: max_idle_pings
           ]
         ]},
      restart: :temporary,
      shutdown: 5_000
    }

    case sup_pid(supervisor) do
      nil -> {:error, :not_found}
      sup -> DynamicSupervisor.start_child(sup, child)
    end
  end

  @doc """
  Terminates the pool child identified by `pool_pid`.

  Returns `:ok` on success, `{:error, :not_found}` when the pid is not a
  child of this supervisor (including when the supervisor itself is not
  reachable by name).
  """
  @spec stop_pool(pid() | atom(), pid()) :: :ok | {:error, :not_found}
  def stop_pool(supervisor, pool_pid) when is_pid(pool_pid) do
    case sup_pid(supervisor) do
      nil ->
        {:error, :not_found}

      sup ->
        case DynamicSupervisor.terminate_child(sup, pool_pid) do
          :ok -> :ok
          {:error, _} -> {:error, :not_found}
        end
    end
  end

  defp sup_pid(supervisor) when is_atom(supervisor), do: Process.whereis(supervisor)
  defp sup_pid(pid) when is_pid(pid), do: pid
end