aerospike_driver

defmodule Aerospike.Cluster.NodeTransport do
  @moduledoc """
  Behaviour for per-node I/O, isolating cluster logic from transport details.

  The Tender, Router, and command modules call this behaviour — they must
  never reach `:gen_tcp` directly. The same contract is served by a real
  TCP implementation in production and a scripted fake in tests, so every
  cluster-logic path is exercised deterministically without sockets.

  This behaviour is intentionally narrow: `info/2`, unary `command/4`,
  multi-frame `command_stream/4`, optional `login/2`, and optional stream
  open/read/close callbacks are the execution shapes supported today.
  Streaming replies and fan-out orchestration are separate contracts, not
  hidden variants of `command/4`.

  Implementations are free to choose their own `conn` representation
  (a socket, a pid, a reference, a struct). The opaque type prevents
  transport internals from leaking into callers.
  """

  @typedoc "Opaque connection handle returned by `c:connect/3`."
  @type conn :: term()

  @typedoc "Authentication mode used by transports that implement `c:login/2`."
  @type auth_mode :: :internal | :external | :pki

  @typedoc """
  Connection option accepted by `c:connect/3`.

  Implementations may support additional keys. The built-in TCP/TLS
  transports document their concrete option sets in
  `Aerospike.Transport.Tcp` and `Aerospike.Transport.Tls`.
  """
  @type connect_option :: {atom(), term()}

  @typedoc "Keyword list of connection options accepted by `c:connect/3`."
  @type connect_opts :: [connect_option()]

  @doc """
  Opens a connection to `host:port`.

  On success, returns an opaque connection handle suitable for passing to
  the other callbacks. On failure, returns an `Aerospike.Error` tagged
  with a transport-appropriate code (e.g. `:connection_error`, `:timeout`).
  """
  @callback connect(host :: String.t(), port :: :inet.port_number(), opts :: connect_opts()) ::
              {:ok, conn()} | {:error, Aerospike.Error.t()}

  @doc """
  Closes the connection. Must be idempotent — calling `close/1` on an
  already-closed handle returns `:ok`.
  """
  @callback close(conn()) :: :ok

  @doc """
  Issues one info request containing `commands` and returns the parsed
  key/value map from the server's reply.

  The single-round-trip shape matches the Aerospike info protocol: all
  commands ship in one request, all results come back in one response.
  """
  @callback info(conn(), commands :: [String.t()]) ::
              {:ok, %{String.t() => String.t()}} | {:error, Aerospike.Error.t()}

  @typedoc """
  Options accepted by `c:command/4`.

    * `:use_compression` — when `true`, requests whose encoded size exceeds
      a fixed 128-byte threshold are wrapped in a type-4
      (`AS_MSG_COMPRESSED`) proto frame before being sent. Smaller requests
      are sent plain even when the flag is set. Implementations that ignore
      compression must still accept the option without error. Defaults to
      `false`.
    * `:message_type` — `:as_msg` (default) or `:admin`. `:admin` is the
      Aerospike admin-protocol reply type used by security commands.
    * `:attempt` — zero-based retry attempt index attached to telemetry
      metadata. Defaults to `0` when omitted.
  """
  @type command_option ::
          {:use_compression, boolean()}
          | {:message_type, :as_msg | :admin}
          | {:attempt, non_neg_integer()}

  @typedoc "Keyword list accepted by `c:command/4` and `c:command_stream/4`."
  @type command_opts :: [command_option()]

  @doc """
  Sends a pre-encoded request and returns the full reply bytes.

  The request is expected to be complete wire bytes (proto header + body)
  produced by the command encoder. The reply is the full response payload;
  framing/parsing is the caller's responsibility.

  `deadline_ms` is a per-socket-read deadline in milliseconds applied to
  each `:gen_tcp.recv/3` call (header and body are read separately on
  passive `{:packet, :raw}` sockets; see `Aerospike.Transport.Tcp`). It is
  deliberately separate from the caller's total-operation budget: a slow
  node can blow its read deadline without the caller having to track a
  monotonic deadline manually. The caller remains responsible for the
  overall operation budget — the transport does not enforce it.

  `opts` is a keyword list documented by `t:command_opts/0`.

  Single request, single response — streaming and multi-frame replies
  (scan, query) are out of scope for this behaviour.
  """
  @callback command(
              conn(),
              request :: iodata(),
              deadline_ms :: non_neg_integer(),
              opts :: command_opts()
            ) ::
              {:ok, binary()} | {:error, Aerospike.Error.t()}

  @doc """
  Sends a pre-encoded request and reads a multi-frame reply through
  the terminal marker, returning the concatenated reply frames or bodies
  appropriate for `opts[:message_type]`.

  This seam exists for batch-style requests that still complete as one
  bounded command but arrive as multiple protocol frames. Unary callers
  must continue using `c:command/4`; long-lived incremental consumers
  should use the stream callbacks.
  """
  @callback command_stream(
              conn(),
              request :: iodata(),
              deadline_ms :: non_neg_integer(),
              opts :: command_opts()
            ) ::
              {:ok, binary()} | {:error, Aerospike.Error.t()}

  @typedoc """
  Option accepted by `c:stream_open/4`.

  The built-in TCP/TLS transports accept `:use_compression` and `:attempt`,
  matching `t:command_opts/0`; implementations may ignore keys they do not
  support.
  """
  @type stream_option ::
          {:use_compression, boolean()}
          | {:attempt, non_neg_integer()}

  @typedoc "Keyword list accepted by `c:stream_open/4`."
  @type stream_opts :: [stream_option()]

  @typedoc "Opaque stream handle returned by `c:stream_open/4`."
  @type stream :: term()

  @doc """
  Opens a streaming request and returns an opaque handle for reading frames.

  The transport owns the long-lived socket state and delivery mechanics; the
  caller owns any parsing of returned frames. The request bytes are already
  encoded when they reach the transport.
  """
  @callback stream_open(
              conn(),
              request :: iodata(),
              deadline_ms :: non_neg_integer(),
              opts :: stream_opts()
            ) :: {:ok, stream()} | {:error, Aerospike.Error.t()}

  @doc """
  Reads the next frame from an open stream.

  Returns `{:ok, frame}` for a delivered frame, `:done` when the stream has
  reached end-of-stream, or `{:error, error}` for a transport-class failure.
  """
  @callback stream_read(stream(), deadline_ms :: non_neg_integer()) ::
              {:ok, binary()} | :done | {:error, Aerospike.Error.t()}

  @doc """
  Closes an open stream handle. Must be idempotent.
  """
  @callback stream_close(stream()) :: :ok

  @typedoc """
  Parsed admin-protocol login reply returned by `c:login/2`.

    * `:ok_no_token` — server accepted the login but issued no session
      token (e.g. PKI with no user mapping).
    * `{:session, token, ttl_seconds_or_nil}` — server accepted the login
      and issued `token`. `ttl_seconds` is the server-reported TTL (nil
      if omitted).
    * `:security_not_enabled` — server has security disabled (result
      code 52); callers treat this as success with no token.
  """
  @type login_reply ::
          :ok_no_token
          | {:session, binary(), non_neg_integer() | nil}
          | :security_not_enabled

  @typedoc """
  Login option accepted by `c:login/2`.

    * `:user` — username. Required for password login and session-token
      authenticate, omitted for PKI login.
    * `:password` — cleartext password for internal/external login.
    * `:auth_mode` — `:internal`, `:external`, or `:pki`.
    * `:session_token` — token used to run AUTHENTICATE instead of LOGIN.
    * `:login_timeout_ms` — read deadline applied to the login reply.
  """
  @type login_option ::
          {:user, String.t()}
          | {:password, String.t()}
          | {:auth_mode, auth_mode()}
          | {:session_token, binary()}
          | {:login_timeout_ms, pos_integer()}

  @typedoc "Keyword list accepted by `c:login/2`."
  @type login_opts :: [login_option()]

  @doc """
  Runs the admin-protocol login (or session-token authenticate) handshake
  on an already-connected socket and returns the parsed reply.

  Optional — transports that do not support authentication raise
  `UndefinedFunctionError`; callers that use this callback must ensure
  the configured transport implements it.

  Opts:

    * `:user` — username (required when `:session_token` is absent).
    * `:password` — cleartext password (required when `:session_token`
      is absent; the transport hashes it as the server requires).
    * `:auth_mode` — `:internal`, `:external`, or `:pki`.
    * `:session_token` — when present, runs AUTHENTICATE instead of a
      fresh LOGIN. `:user` is still required.
    * `:login_timeout_ms` — read deadline applied to the login reply.
      Transport-specific default.
  """
  @callback login(conn(), opts :: login_opts()) ::
              {:ok, login_reply()} | {:error, Aerospike.Error.t()}

  @optional_callbacks login: 2, stream_open: 4, stream_read: 2, stream_close: 1
end