aerospike_driver

defmodule Aerospike do
  @moduledoc """
  Public entry point for the Aerospike Elixir driver.

  The public package identity is `aerospike_driver`. The useful contract here
  is still narrower than "full client parity": start-up should be explicit,
  supported command families should be discoverable, and the named proof suites
  should match the public surface.

  The driver currently proves a small unary command family on one shared execution
  path plus batch, scan, query, and transaction helpers over the same
  supervised cluster runtime:

    * `get/3` reads all bins for a key
    * `get_header/2` reads only record metadata for a key
    * `put/4` writes a bin map
    * `put_payload/4` sends a caller-built single-record write/delete frame
    * `apply_udf/6` executes one record UDF against one key
    * `register_udf/3`, `register_udf/4`, `remove_udf/2`,
      `remove_udf/3`, and `list_udfs/1` / `list_udfs/2` manage
      server-side UDF packages and return explicit metadata/task
      handles
    * `truncate/2`, `truncate/3`, and `truncate/4` expose one-node
      operator truncation helpers with explicit namespace and set forms
    * `create_user/5`, `create_pki_user/4`, `drop_user/3`,
      `change_password/4`, `grant_roles/4`, `revoke_roles/4`, `query_user/3`,
      `query_users/2`, `create_role/4`, `drop_role/3`,
      `set_whitelist/4`, `set_quotas/5`, `grant_privileges/4`,
      `revoke_privileges/4`, `query_role/3`, and `query_roles/2` expose the
      enterprise security-admin seam
    * `add/4`, `append/4`, and `prepend/4` expose thin unary write
      helpers for common counter and string mutations
    * `metrics_enabled?/1`, `enable_metrics/2`, `disable_metrics/1`,
      `stats/1`, and `warm_up/2` expose opt-in runtime metrics and an
      explicit operator pool probe over the already-started workers
    * `touch/2` updates record metadata
    * `delete/2` removes a record
    * `exists/2` performs a header-only existence probe
    * `operate/4` runs simple, CDT-style, and expression operation lists built
      with `Aerospike.Op`, `Aerospike.Op.List`, `Aerospike.Op.Map`,
      `Aerospike.Op.Exp`, and `Aerospike.Ctx`
    * unary commands (`get/3`, `get_header/2`, `put/4`, `exists/2`,
      `touch/2`, `delete/2`, `operate/4`, `apply_udf/6`, `add/4`,
      `append/4`, and `prepend/4`) accept `%Aerospike.Exp{}` via `:filter`
      for server-side execution filtering
    * `Aerospike.Exp` builds server-side expression values, including values
      usable as expression-backed secondary-index sources
    * `create_expression_index/5` creates expression-backed secondary indexes
      on servers that support them and returns a pollable index task
    * `batch_get/4`, `batch_get_header/3`, `batch_exists/3`,
      `batch_get_operate/4`, `batch_delete/3`, and `batch_udf/6` operate on
      multiple keys and return per-key results in caller order
    * `Aerospike.Batch` and `batch_operate/3` expose a curated heterogeneous
      batch surface for mixing reads, writes, deletes, operations, and record
      UDF calls while returning one `%Aerospike.BatchResult{}` per input
    * `child_spec/1`, `close/2`, `key/3`, and `key_digest/3` round out
      the root lifecycle and key-construction boundary
    * `info/3`, `info_node/4`, `nodes/1`, and `node_names/1` expose
      one-node operator reads over the published cluster view; `info_node/4`
      targets one named active node discovered from `node_names/1` or `nodes/1`
    * `set_xdr_filter/4` sets or clears Enterprise XDR expression filters
      through a one-node info command
    * `query_stream!/3`, `query_all/3`, `query_count/3`,
      `query_aggregate/6`, and `query_aggregate_result/6` run
      secondary-index queries through the same
      node-preparation pipeline, with lazy outer streams but
      node-buffered record delivery
      and optional `%Aerospike.Exp{}` filters through `Query.filter/2`
      when a secondary-index predicate from `Query.where/2` is also used.
    * `query_execute/4` and `query_udf/6` run background query jobs on
      that same setup path and return pollable task handles
    * `scan_stream/3`, `scan_stream!/3`, `scan_all/3`, `scan_all!/3`,
      `scan_count/3`, `scan_count!/3`, `scan_page/3`, and `scan_page!/3`
      run scan fan-out across the same scan/runtime setup, again with lazy
      outer streams, node-buffered record delivery, and resumable pages
    * scan/query helpers that already support node targeting accept
      `node: node_name` in `opts`; discover names with `node_names/1` or
      `nodes/1`

  Quick-start shape:

      {:ok, _sup} =
        Aerospike.start_link(
          name: :aerospike,
          transport: Aerospike.Transport.Tcp,
          hosts: ["127.0.0.1:3000"],
          namespaces: ["test"],
          pool_size: 2
        )

      Aerospike.Cluster.ready?(:aerospike)

  The repo README names the supported validation profiles:

    * Community Edition single-node on `localhost:3000`
    * Community Edition three-node cluster from this repo's `docker-compose.yml`
    * Enterprise Edition variants from this repo's `docker-compose.yml`

  The public `Stream` helpers are lazy only at the API boundary. The
  current runtime still drains each node stream into memory before
  yielding that node's records downstream, so it does not promise
  frame-by-frame cross-node backpressure or cancellation coordination.

  Expression builders now cover scalar, metadata, arithmetic, conditional,
  variable, CDT, bit, and HyperLogLog helper families. Public policy remains
  keyword-based at facade calls, with centralized normalization for
  single-record, batch, scan, query, startup, auth, and discovery options.

  Caller-facing policy validation and default materialization is shared across
  command families. Public command functions still accept keyword opts, but
  command paths no longer merge or validate those policy families ad hoc.
  """

  alias Aerospike.Batch
  alias Aerospike.Cluster
  alias Aerospike.Cluster.Supervisor, as: ClusterSupervisor
  alias Aerospike.Command.Admin
  alias Aerospike.Command.ApplyUdf
  alias Aerospike.Command.Batch, as: MixedBatch
  alias Aerospike.Command.BatchCommand.Entry
  alias Aerospike.Command.BatchDelete
  alias Aerospike.Command.BatchGet
  alias Aerospike.Command.BatchUdf
  alias Aerospike.Command.Delete
  alias Aerospike.Command.Exists
  alias Aerospike.Command.Get
  alias Aerospike.Command.Operate
  alias Aerospike.Command.Put
  alias Aerospike.Command.PutPayload
  alias Aerospike.Command.ScanOps
  alias Aerospike.Command.Touch
  alias Aerospike.Command.WriteOp
  alias Aerospike.Error
  alias Aerospike.ExecuteTask
  alias Aerospike.Exp
  alias Aerospike.IndexTask
  alias Aerospike.Key
  alias Aerospike.Page
  alias Aerospike.Policy
  alias Aerospike.Privilege
  alias Aerospike.Protocol.AsmMsg.Operation
  alias Aerospike.Protocol.OperateFlags
  alias Aerospike.Query
  alias Aerospike.RegisterTask
  alias Aerospike.RetryPolicy
  alias Aerospike.Role
  alias Aerospike.Runtime.TxnRoll
  alias Aerospike.RuntimeMetrics
  alias Aerospike.Scan
  alias Aerospike.Txn
  alias Aerospike.UDF
  alias Aerospike.User

  @typedoc """
  Identifier for a running cluster facade.

  Read-side helpers accept the registered cluster name or a pid
  registered under that name. Arbitrary `GenServer.server()` forms are
  not supported.
  """
  @type cluster :: named_cluster() | pid()

  @typedoc """
  Registered atom name for a running cluster.

  Lifecycle and transaction helpers resolve supervisor and ETS resources from
  this name, so they do not currently accept arbitrary `GenServer.server()`
  identities.
  """
  @type named_cluster :: atom()

  @typedoc """
  One active cluster node as returned by `nodes/1`.
  """
  @type node_info :: %{name: String.t(), host: String.t(), port: :inet.port_number()}

  @typedoc "Security user metadata returned by `query_user/3` and `query_users/2`."
  @type user_info :: User.t()

  @typedoc "Security role metadata returned by `query_role/3` and `query_roles/2`."
  @type role_info :: Role.t()

  @typedoc "Security privilege metadata used by role administration APIs."
  @type privilege :: Privilege.t()

  @typedoc """
  Non-negative timeout value in milliseconds.

  `:timeout` is the total client-side budget for a command. `:socket_timeout`
  is the per-attempt idle socket deadline; `0` asks the command path to use the
  remaining total budget.
  """
  @type timeout_ms :: non_neg_integer()

  @typedoc """
  Record TTL accepted by write commands.

    * non-negative integer - explicit TTL in seconds
    * `:default` - use the namespace default TTL
    * `:never_expire` - ask the server to keep the record indefinitely
    * `:dont_update` - preserve the existing record TTL
  """
  @type ttl :: non_neg_integer() | :default | :never_expire | :dont_update

  @typedoc """
  AP namespace read consistency.

  `:one` reads from one available replica. `:all` asks the server to consult
  all relevant replicas for the read.
  """
  @type read_mode_ap :: :one | :all

  @typedoc """
  Strong-consistency namespace read consistency.

  The default is `:session`. `:linearize` requests linearized reads.
  `:allow_replica` and `:allow_unavailable` relax consistency for availability
  when the server namespace configuration allows it.
  """
  @type read_mode_sc :: :session | :linearize | :allow_replica | :allow_unavailable

  @typedoc """
  Record existence behavior for write commands.

    * `:update` - create or update bins (default)
    * `:update_only` - update an existing record only
    * `:create_or_replace` - create a new record or replace the existing record
    * `:replace_only` - replace an existing record only
    * `:create_only` - create only when the key is absent
  """
  @type record_exists_action ::
          :update | :update_only | :create_or_replace | :replace_only | :create_only

  @typedoc """
  Generation check behavior for write commands.

  `:none` disables generation checks. `:expect_equal` requires the server
  record generation to equal `:generation`. `:expect_gt` requires it to be
  greater than `:generation`. When `:generation` is positive and
  `:generation_policy` is omitted, the driver defaults to `:expect_equal`.
  """
  @type generation_policy :: :none | :expect_equal | :expect_gt

  @typedoc """
  Commit acknowledgement level for write commands.

  `:all` waits for the server's configured replica commit level. `:master`
  accepts acknowledgement from the master partition owner.
  """
  @type commit_level :: :all | :master

  @typedoc """
  Retry option accepted by single-record, scan, and query command families.

    * `:max_retries` - retries after the first attempt
    * `:sleep_between_retries_ms` - fixed delay between retry attempts
    * `:replica_policy` - read retry routing policy, `:master` or `:sequence`

  Retry defaults are configured at cluster start and can be overridden per
  command where this type appears.
  """
  @type retry_opt ::
          {:max_retries, non_neg_integer()}
          | {:sleep_between_retries_ms, non_neg_integer()}
          | {:replica_policy, RetryPolicy.replica_policy()}

  @typedoc """
  Read option accepted by `get/4`, `get_header/3`, and `exists/3`.

  Read helpers accept timeout/retry opts, consistency opts, `:send_key`,
  `:use_compression`, and a server-side expression `:filter`.
  """
  @type read_opt ::
          {:timeout, timeout_ms()}
          | {:socket_timeout, timeout_ms()}
          | retry_opt()
          | {:read_mode_ap, read_mode_ap()}
          | {:read_mode_sc, read_mode_sc()}
          | {:read_touch_ttl_percent, -1 | 0..100}
          | {:send_key, boolean()}
          | {:use_compression, boolean() | nil}
          | {:filter, Exp.t() | nil}

  @typedoc "Keyword options accepted by read helpers."
  @type read_opts :: [read_opt()]

  @typedoc """
  Write option accepted by single-record write, delete, UDF, and operate helpers.

  Write helpers accept timeout/retry opts, record TTL and generation policy,
  existence policy, commit level, durable delete, `:respond_per_op`, `:send_key`,
  read-touch metadata opts used by mixed operations, per-command compression,
  a server-side expression `:filter`, and an optional transaction handle.
  """
  @type write_opt ::
          {:timeout, timeout_ms()}
          | {:socket_timeout, timeout_ms()}
          | retry_opt()
          | {:ttl, ttl()}
          | {:generation, non_neg_integer()}
          | {:generation_policy, generation_policy()}
          | {:exists, record_exists_action()}
          | {:commit_level, commit_level()}
          | {:durable_delete, boolean()}
          | {:respond_per_op, boolean()}
          | {:send_key, boolean()}
          | {:read_mode_ap, read_mode_ap()}
          | {:read_mode_sc, read_mode_sc()}
          | {:read_touch_ttl_percent, -1 | 0..100}
          | {:use_compression, boolean() | nil}
          | {:filter, Exp.t() | nil}
          | {:txn, Txn.t()}

  @typedoc "Keyword options accepted by single-record write, delete, UDF, and operate helpers."
  @type write_opts :: [write_opt()]

  @typedoc """
  Parent batch dispatch option.

  Parent options describe the batch request as a whole: deadline, node
  concurrency, partial-result handling, response shape, and inline execution
  hints. Per-record read/write policy belongs on `batch_record_read_opt/0` and
  `batch_record_write_opt/0`.
  """
  @type batch_parent_opt ::
          {:timeout, timeout_ms()}
          | {:socket_timeout, timeout_ms()}
          | {:max_concurrent_nodes, non_neg_integer()}
          | {:allow_partial_results, boolean()}
          | {:respond_all_keys, boolean()}
          | {:allow_inline, boolean()}
          | {:allow_inline_ssd, boolean()}

  @typedoc "Parent batch options accepted by `batch_operate/3`."
  @type batch_parent_opts :: [batch_parent_opt()]

  @typedoc """
  Batch read option accepted by per-entry batch reads.

  These options are encoded with one batch read entry. Parent dispatch options
  are only accepted by homogeneous batch read helpers and `batch_operate/3`.
  """
  @type batch_record_read_opt ::
          {:filter, Exp.t() | nil}
          | {:read_mode_ap, read_mode_ap()}
          | {:read_mode_sc, read_mode_sc()}
          | {:read_touch_ttl_percent, -1 | 0..100}

  @typedoc "Keyword options accepted by per-entry batch reads."
  @type batch_record_read_opts :: [batch_record_read_opt()]

  @typedoc "Batch read option accepted by homogeneous batch-read helpers."
  @type batch_read_opt :: batch_parent_opt() | batch_record_read_opt()

  @typedoc "Keyword options accepted by batch read helpers."
  @type batch_read_opts :: [batch_read_opt()]

  @typedoc """
  Batch write option accepted by per-entry batch writes.

  These options are encoded with one batch write/delete/UDF entry. Parent
  dispatch options are only accepted by homogeneous batch write helpers and
  `batch_operate/3`.
  """
  @type batch_record_write_opt ::
          {:ttl, ttl()}
          | {:generation, non_neg_integer()}
          | {:generation_policy, generation_policy()}
          | {:exists, record_exists_action()}
          | {:commit_level, commit_level()}
          | {:durable_delete, boolean()}
          | {:respond_per_op, boolean()}
          | {:send_key, boolean()}
          | {:filter, Exp.t() | nil}
          | {:read_mode_ap, read_mode_ap()}
          | {:read_mode_sc, read_mode_sc()}
          | {:read_touch_ttl_percent, -1 | 0..100}

  @typedoc "Keyword options accepted by per-entry batch writes."
  @type batch_record_write_opts :: [batch_record_write_opt()]

  @typedoc "Batch write option accepted by homogeneous batch-write helpers."
  @type batch_write_opt :: batch_parent_opt() | batch_record_write_opt()

  @typedoc "Keyword options accepted by batch write helpers."
  @type batch_write_opts :: [batch_write_opt()]

  @typedoc "Node-targeting option accepted by scan/query helpers that support single-node execution."
  @type node_opt :: {:node, String.t()}

  @typedoc """
  Runtime option accepted by all scan and query helpers.

  These options tune client runtime behavior and server scan/query fields:
  command deadlines, retries, task timeout, pool checkout timeout, fan-out
  concurrency, throttling, bin-data inclusion, expected duration, task id, and
  cursor resume state.
  """
  @type scan_query_runtime_opt ::
          {:timeout, timeout_ms()}
          | {:socket_timeout, timeout_ms()}
          | retry_opt()
          | {:task_timeout, timeout_ms() | :infinity}
          | {:pool_checkout_timeout, timeout_ms()}
          | {:max_concurrent_nodes, non_neg_integer()}
          | {:records_per_second, non_neg_integer()}
          | {:include_bin_data, boolean()}
          | {:expected_duration, :long | :short | :long_relax_ap}
          | {:task_id, pos_integer()}
          | {:cursor, Aerospike.Cursor.t()}

  @typedoc """
  Runtime option accepted by scan/query helpers that may target one node.

  `:node` must be an active node name returned by `node_names/1` or `nodes/1`.
  Final aggregate reduction does not accept `:node` because it must consume all
  server partials required to produce one local result.
  """
  @type scan_query_opt :: scan_query_runtime_opt() | node_opt()

  @typedoc "Keyword options accepted by scan and query helpers."
  @type scan_query_opts :: [scan_query_opt()]

  @typedoc """
  Additional local-source option for finalized aggregate queries.

  Exactly one of these is required by `query_aggregate_result/6`: inline Lua
  source via `:source`, or a readable local file path via `:source_path`.
  """
  @type aggregate_result_opt :: {:source, String.t()} | {:source_path, Path.t()}

  @typedoc "Keyword options accepted by `query_aggregate_result/6` and `query_aggregate_result!/6`."
  @type aggregate_result_opts :: [scan_query_runtime_opt() | aggregate_result_opt()]

  @typedoc """
  Option accepted by one-node info/admin helpers.

  `:pool_checkout_timeout` bounds checkout from the selected node pool before
  the one-node info/admin command is sent.
  """
  @type admin_opt :: {:pool_checkout_timeout, timeout_ms()}

  @typedoc "Keyword options accepted by one-node info/admin helpers."
  @type admin_opts :: [admin_opt()]

  @typedoc """
  Option accepted by security-admin helpers.

  `:timeout` bounds the admin protocol operation. `:pool_checkout_timeout`
  bounds checkout from the selected node pool before the command is sent.
  """
  @type security_admin_opt :: {:timeout, timeout_ms()} | {:pool_checkout_timeout, timeout_ms()}

  @typedoc "Keyword options accepted by security-admin helpers."
  @type security_admin_opts :: [security_admin_opt()]

  @typedoc "Secondary-index collection type."
  @type index_collection :: :list | :mapkeys | :mapvalues

  @typedoc "Secondary-index particle/source type."
  @type index_type :: :numeric | :string | :geo2dsphere

  @typedoc """
  Option accepted by `create_index/4`.

  `:bin`, `:name`, and `:type` are required. `:collection` creates a collection
  index over list values, map keys, or map values. `:ctx` targets a nested CDT
  path built with `Aerospike.Ctx`.
  """
  @type create_index_opt ::
          {:bin, String.t()}
          | {:name, String.t()}
          | {:type, index_type()}
          | {:collection, index_collection()}
          | {:ctx, [Aerospike.Ctx.step()]}
          | admin_opt()

  @typedoc "Keyword options accepted by `create_index/4`."
  @type create_index_opts :: [create_index_opt()]

  @typedoc """
  Option accepted by `create_expression_index/5`.

  `:name` and `:type` are required. Expression indexes use the provided
  `%Aerospike.Exp{}` as their source and therefore do not accept `:bin` or
  nested CDT `:ctx`.
  """
  @type create_expression_index_opt ::
          {:name, String.t()}
          | {:type, index_type()}
          | {:collection, index_collection()}
          | admin_opt()

  @typedoc "Keyword options accepted by `create_expression_index/5`."
  @type create_expression_index_opts :: [create_expression_index_opt()]

  @typedoc """
  Option accepted by `truncate/3` and `truncate/4`.

  `:before` limits truncation to records last updated before the given
  timestamp. `:pool_checkout_timeout` bounds checkout for the admin request.
  """
  @type truncate_opt :: {:before, DateTime.t()} | admin_opt()

  @typedoc "Keyword options accepted by truncate helpers."
  @type truncate_opts :: [truncate_opt()]

  @typedoc "Option accepted by `enable_metrics/2`."
  @type metrics_opt :: {:reset, boolean()}

  @typedoc "Keyword options accepted by `enable_metrics/2`."
  @type metrics_opts :: [metrics_opt()]

  @typedoc """
  Option accepted by `warm_up/2`.

  `:count` is the number of worker checkout probes per active node. `0` means
  up to the configured pool size. `:pool_checkout_timeout` bounds each probe.
  """
  @type warm_up_opt :: {:count, non_neg_integer()} | {:pool_checkout_timeout, timeout_ms()}

  @typedoc "Keyword options accepted by `warm_up/2`."
  @type warm_up_opts :: [warm_up_opt()]

  @typedoc "Role-creation option accepted by `create_role/4`."
  @type create_role_opt ::
          {:whitelist, [String.t()]}
          | {:read_quota, non_neg_integer()}
          | {:write_quota, non_neg_integer()}
          | security_admin_opt()

  @typedoc "Keyword options accepted by `create_role/4`."
  @type create_role_opts :: [create_role_opt()]

  @typedoc "Transaction options accepted by `transaction/3` when a handle is not supplied."
  @type transaction_opts :: [{:timeout, timeout_ms()}]

  @typedoc "Operation input accepted by `operate/4`."
  @type operate_operation ::
          {:write, String.t() | atom(), term()}
          | {:read, String.t() | atom()}
          | {:add, String.t() | atom(), integer() | float()}
          | {:append, String.t() | atom(), String.t()}
          | {:prepend, String.t() | atom(), String.t()}
          | :touch
          | :delete
          | Aerospike.Op.t()

  @doc """
  Starts a supervised cluster.

  Internally this delegates to `Aerospike.Cluster.Supervisor.start_link/1`,
  which boots the cluster-state owner, per-node pool supervisor, partition-map
  writer, and tend-cycle process under one `rest_for_one` tree. The `:name`
  option is the public cluster identity later passed to `get/3`,
  `Aerospike.Cluster.ready?/1`, and the other facade/read-side helpers.

  Required options: `:name`, `:transport`, `:hosts`, `:namespaces`.

  Startup validation happens synchronously at this boundary. Shape
  errors for required opts, retry/breaker knobs, TLS/connect opts, and
  auth pairs fail `start_link/1` immediately instead of surfacing later
  from the first tend cycle or pool worker. The `:name` remains the
  public cluster identity for facade calls such as `get/3`, `info/3`,
  and `Aerospike.Cluster.ready?/1`.

  Cluster lifecycle knobs:

    * `:tend_trigger` — `:timer` (default) or `:manual`.
    * `:tend_interval_ms` — automatic tend period in milliseconds.
      Positive integer.
    * `:failure_threshold` — consecutive tend failures before the
      tend cycle demotes a node. Non-negative integer.

  Pool-level knobs (forwarded internally on each node-pool start):

    * `:pool_size` — workers per node. Positive integer.
    * `:min_connections_per_node` — minimum warm connection target per
      node. Non-negative integer. `0` allows lazy worker creation.
    * `:idle_timeout_ms` — milliseconds a worker may sit idle before the
      pool verification step evicts it. Positive integer. Defaults stay under
      Aerospike's `proto-fd-idle-ms`.
    * `:max_idle_pings` — bound on how many idle workers NimblePool
      may drop per verification cycle. Positive integer.

  Breaker and retry knobs:

    * `:circuit_open_threshold` — consecutive node failures tolerated
      before new commands are refused. Non-negative integer.
    * `:max_concurrent_ops_per_node` — in-flight plus queued command
      cap enforced per node. Positive integer.
    * `:max_retries` — retries after the initial attempt. Non-negative
      integer.
    * `:sleep_between_retries_ms` — fixed delay between retries.
      Non-negative integer.
    * `:replica_policy` — `:master` or `:sequence`.

  Cluster feature toggles:

    * `:use_compression` — boolean cluster-wide request-compression
      opt-in, gated per node by advertised capabilities.
    * `:use_services_alternate` — boolean toggle for
      `peers-clear-alt` discovery.
    * `:seed_only_cluster` — boolean. When true, discovery is limited to
      configured seed addresses.
    * `:cluster_name` — expected server cluster name. Nodes reporting a
      different name are rejected.
    * `:application_id` — application identity sent to capable servers for
      server-side client correlation.

  Auth opts:

    * `:auth_mode` — `:internal` (default), `:external`, or `:pki`.
      External and PKI modes require `Aerospike.Transport.Tls`.
    * `:user` / `:password` — cluster-wide credentials. Must be passed
      together or omitted together. PKI auth uses the TLS client certificate
      and must omit both.
    * `:login_timeout_ms` — startup/login read deadline in milliseconds.

  TCP-level tuning knobs (passed verbatim to the TCP transport via the
  `:connect_opts` keyword):

    * `:connect_timeout_ms` — handshake + write-buffer drain deadline.
    * `:info_timeout` — read deadline applied to every `info/2` call.
      Defaults to `:connect_timeout_ms`.
    * `:tcp_nodelay` — boolean, default `true`.
    * `:tcp_keepalive` — boolean, default `true`.
    * `:tcp_sndbuf` / `:tcp_rcvbuf` — positive integer kernel buffer
      sizes. Unset lets the kernel pick.

  `Aerospike.Cluster.Supervisor` documents the underlying validation and child
  ownership details.
  """
  @spec start_link([ClusterSupervisor.option()]) :: Supervisor.on_start()
  def start_link(opts) when is_list(opts) do
    ClusterSupervisor.start_link(opts)
  end

  @doc """
  Returns a child specification for one supervised cluster.

  This delegates to the same cluster-supervisor implementation as
  `start_link/1`, so the accepted options and validation boundary match.
  """
  @spec child_spec([ClusterSupervisor.option()]) :: Supervisor.child_spec()
  def child_spec(opts) when is_list(opts) do
    ClusterSupervisor.child_spec(opts)
  end

  @doc """
  Stops the supervised cluster registered under the atom `cluster`.

  This targets the registered cluster supervisor name derived from `cluster`
  and returns `:ok` when the supervisor exits or is already absent.
  """
  @spec close(named_cluster(), timeout :: non_neg_integer()) :: :ok
  def close(cluster, timeout \\ 15_000)
      when is_atom(cluster) and is_integer(timeout) and timeout >= 0 do
    case Process.whereis(ClusterSupervisor.sup_name(cluster)) do
      nil -> :ok
      pid -> Supervisor.stop(pid, :normal, timeout)
    end
  end

  @doc """
  Builds a key from namespace, set, and a user key.

  This is a thin wrapper over `Aerospike.Key.new/3`.
  """
  @spec key(String.t(), String.t(), String.t() | integer()) :: Key.t()
  def key(namespace, set, user_key), do: Key.new(namespace, set, user_key)

  @doc """
  Builds a key from namespace, set, and an existing 20-byte digest.

  This is a thin wrapper over `Aerospike.Key.from_digest/3`.
  """
  @spec key_digest(String.t(), String.t(), <<_::160>>) :: Key.t()
  def key_digest(namespace, set, digest), do: Key.from_digest(namespace, set, digest)

  @doc """
  Reads `key` from `cluster`.

  Pass `:all` to read every bin, or pass a non-empty list of string or atom bin
  names to project only those bins.

  Options:

    * `:timeout` — total op-budget milliseconds for the call, shared
      across the initial send and every retry. Default `5_000`.
    * `:socket_timeout` — per-attempt socket idle deadline in milliseconds,
      capped by the remaining total budget. `0` uses the remaining total
      budget.
    * `:max_retries` — overrides the cluster-default retry cap for this
      call. `0` disables retry entirely. See `Aerospike.RetryPolicy`.
    * `:sleep_between_retries_ms` — fixed delay between retry attempts.
    * `:replica_policy` — `:master` (all attempts against the master)
      or `:sequence` (walk the replica list by attempt index).
    * `:read_mode_ap` — `:one` or `:all`.
    * `:read_mode_sc` — `:session`, `:linearize`, `:allow_replica`, or
      `:allow_unavailable`.
    * `:read_touch_ttl_percent` — `-1`, `0`, or an integer from `1` through
      `100`.
    * `:send_key` — include the user key in the request when `true`.
    * `:use_compression` — per-command compression override; `nil` keeps the
      cluster/node default.
    * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter
      expression, or `nil` for no filter.

  Returns `{:ok, %Aerospike.Record{}}` on hit,
  `{:error, %Aerospike.Error{code: :key_not_found}}` on miss, or a
  routing atom (`:cluster_not_ready`, `:no_master`, `:unknown_node`)
  when the cluster view cannot serve the request.

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec get(cluster(), Key.key_input(), :all | [String.t() | atom()], read_opts()) ::
          {:ok, Aerospike.Record.t()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def get(cluster, key, bins \\ :all, opts \\ [])

  def get(cluster, key, bins, opts) do
    with {:ok, key} <- coerce_key(key) do
      Get.execute(cluster, key, bins, opts)
    end
  end

  @doc """
  Reads only record metadata for `key` from `cluster`.

  This is the explicit single-record header helper. It reuses the same unary
  read path as `get/3`, but requests only generation/TTL metadata and returns
  a `%Aerospike.Record{}` with `bins: %{}` on hit.

  Options:

    * `:timeout` — total op-budget milliseconds for the call, shared
      across the initial send and every retry. Default `5_000`.
    * `:socket_timeout` — per-attempt socket idle deadline in milliseconds,
      capped by the remaining total budget. `0` uses the remaining total
      budget.
    * `:max_retries` — overrides the cluster-default retry cap for this
      call. `0` disables retry entirely. See `Aerospike.RetryPolicy`.
    * `:sleep_between_retries_ms` — fixed delay between retry attempts.
    * `:replica_policy` — `:master` (all attempts against the master)
      or `:sequence` (walk the replica list by attempt index).
    * `:read_mode_ap` — `:one` or `:all`.
    * `:read_mode_sc` — `:session`, `:linearize`, `:allow_replica`, or
      `:allow_unavailable`.
    * `:read_touch_ttl_percent` — `-1`, `0`, or an integer from `1` through
      `100`.
    * `:send_key` — include the user key in the request when `true`.
    * `:use_compression` — per-command compression override; `nil` keeps the
      cluster/node default.
    * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter
      expression, or `nil` for no filter.

  Returns `{:ok, %Aerospike.Record{bins: %{}}}` on hit,
  `{:error, %Aerospike.Error{code: :key_not_found}}` on miss, or a
  routing atom (`:cluster_not_ready`, `:no_master`, `:unknown_node`)
  when the cluster view cannot serve the request.

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec get_header(cluster(), Key.key_input(), read_opts()) ::
          {:ok, Aerospike.Record.t()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def get_header(cluster, key, opts \\ [])

  def get_header(cluster, key, opts) when is_list(opts) do
    with {:ok, key} <- coerce_key(key) do
      Get.execute(cluster, key, :header, opts)
    end
  end

  @doc """
  Reads multiple `keys` from `cluster` in one batch request per target node.

  The result list stays in the same order as `keys`. Each list item is
  either `{:ok, %Aerospike.Record{}}` for a hit or an indexed error for
  that key (`{:error, %Aerospike.Error{}}`, `{:error, :no_master}`, or
  `{:error, :unknown_node}`).

  Pass `:all` to read every bin, or pass a non-empty list of string or atom bin
  names to project only those bins. Batch opts include parent dispatch fields
  such as `:timeout`, `:socket_timeout`, `:max_concurrent_nodes`,
  `:allow_partial_results`, `:respond_all_keys`, `:allow_inline`, and
  `:allow_inline_ssd`, plus encodable read fields such as `:filter`,
  `:read_mode_ap`, `:read_mode_sc`, and `:read_touch_ttl_percent`.

  Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples.
  """
  @spec batch_get(cluster(), [Key.key_input()], :all | [String.t() | atom()], batch_read_opts()) ::
          {:ok,
           [
             {:ok, Aerospike.Record.t()}
             | {:error, Aerospike.Error.t()}
             | {:error, :no_master | :unknown_node}
           ]}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready}
  def batch_get(cluster, keys, bins \\ :all, opts \\ [])

  def batch_get(cluster, keys, bins, opts) when is_list(keys) do
    with {:ok, keys} <- coerce_keys(keys) do
      BatchGet.execute(cluster, keys, bins, opts)
    end
  end

  @doc """
  Reads record headers for multiple `keys` from `cluster`.

  The result list stays in the same order as `keys`. Each list item is
  either `{:ok, %Aerospike.Record{bins: %{}}}` for a hit or an indexed
  error for that key (`{:error, %Aerospike.Error{}}`,
  `{:error, :no_master}`, or `{:error, :unknown_node}`).

  This helper accepts the same batch read opts as `batch_get/4`.

  Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples.
  """
  @spec batch_get_header(cluster(), [Key.key_input()], batch_read_opts()) ::
          {:ok,
           [
             {:ok, Aerospike.Record.t()}
             | {:error, Aerospike.Error.t()}
             | {:error, :no_master | :unknown_node}
           ]}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready}
  def batch_get_header(cluster, keys, opts \\ [])

  def batch_get_header(cluster, keys, opts) when is_list(keys) and is_list(opts) do
    with {:ok, keys} <- coerce_keys(keys) do
      BatchGet.execute(cluster, keys, :header, opts)
    end
  end

  @doc """
  Checks existence for multiple `keys` from `cluster`.

  The result list stays in the same order as `keys`. Each list item is
  either `{:ok, true}` / `{:ok, false}` for the targeted key or an
  indexed error for that key (`{:error, %Aerospike.Error{}}`,
  `{:error, :no_master}`, or `{:error, :unknown_node}`).

  This helper accepts the same batch read opts as `batch_get/4`.

  Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples.
  """
  @spec batch_exists(cluster(), [Key.key_input()], batch_read_opts()) ::
          {:ok,
           [
             {:ok, boolean()}
             | {:error, Aerospike.Error.t()}
             | {:error, :no_master | :unknown_node}
           ]}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready}
  def batch_exists(cluster, keys, opts \\ [])

  def batch_exists(cluster, keys, opts) when is_list(keys) and is_list(opts) do
    with {:ok, keys} <- coerce_keys(keys) do
      BatchGet.execute(cluster, keys, :exists, opts)
    end
  end

  @doc """
  Runs one read-only operation list for multiple `keys` from `cluster`.

  The result list stays in the same order as `keys`. Each list item is
  either `{:ok, %Aerospike.Record{}}` for a hit or an indexed error for
  that key (`{:error, %Aerospike.Error{}}`, `{:error, :no_master}`, or
  `{:error, :unknown_node}`). Missing keys remain explicit per-key
  errors; this helper does not collapse misses to `nil`.

  This helper accepts the same batch read opts as `batch_get/4`.

  Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples.
  """
  @spec batch_get_operate(cluster(), [Key.key_input()], [Aerospike.Op.t()], batch_read_opts()) ::
          {:ok,
           [
             {:ok, Aerospike.Record.t()}
             | {:error, Aerospike.Error.t()}
             | {:error, :no_master | :unknown_node}
           ]}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready}
  def batch_get_operate(cluster, keys, operations, opts \\ [])

  def batch_get_operate(cluster, keys, operations, opts)
      when is_list(keys) and is_list(operations) and is_list(opts) do
    with {:ok, keys} <- coerce_keys(keys) do
      BatchGet.execute_operate(cluster, keys, operations, opts)
    end
  end

  @doc """
  Deletes multiple `keys` from `cluster`.

  The result list stays in the same order as `keys`. Each list item is a
  `%Aerospike.BatchResult{}` with `status: :ok` for a deleted record or
  `status: :error` for a per-key failure such as a missing key, routing
  failure, or node transport error. Missing keys remain explicit error
  results with the server result code; this helper does not collapse them to
  booleans.

  This helper accepts parent batch opts and encodable write fields such as
  `:generation_policy`, `:generation`, `:commit_level`, `:durable_delete`,
  `:send_key`, and `:filter`.

  Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples.
  """
  @spec batch_delete(cluster(), [Key.key_input()], batch_write_opts()) ::
          {:ok, [Aerospike.BatchResult.t()]}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready}
  def batch_delete(cluster, keys, opts \\ [])

  def batch_delete(cluster, keys, opts) when is_list(keys) and is_list(opts) do
    with {:ok, keys} <- coerce_keys(keys) do
      BatchDelete.execute(cluster, keys, opts)
    end
  end

  @doc """
  Executes one record UDF for multiple `keys` from `cluster`.

  The result list stays in the same order as `keys`. Each list item is a
  `%Aerospike.BatchResult{}` with `status: :ok` for a successful UDF row or
  `status: :error` for a per-key failure such as a missing key, missing UDF,
  routing failure, or node transport error. Successful UDF rows may include a
  returned `%Aerospike.Record{}` when the server sends return bins.

  This helper accepts the same parent batch and write opts as
  `batch_delete/3`.

  Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples.
  """
  @spec batch_udf(
          cluster(),
          [Key.key_input()],
          String.t(),
          String.t(),
          list(),
          batch_write_opts()
        ) ::
          {:ok, [Aerospike.BatchResult.t()]}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready}
  def batch_udf(cluster, keys, package, function, args, opts \\ [])

  def batch_udf(cluster, keys, package, function, args, opts)
      when is_list(keys) and is_binary(package) and is_binary(function) and is_list(args) and
             is_list(opts) do
    with {:ok, keys} <- coerce_keys(keys) do
      BatchUdf.execute(cluster, keys, package, function, args, opts)
    end
  end

  @doc """
  Executes heterogeneous batch entries built with `Aerospike.Batch`.

  The result list stays in the same order as the input entries. Each list item
  is a `%Aerospike.BatchResult{}` with `status: :ok` for a successful row or
  `status: :error` for a per-key failure such as a missing key, routing
  failure, server error, or node transport error.

  Parent batch policy opts are accepted in `opts`. Per-entry read/write policy
  opts can be attached with the `Aerospike.Batch` builders.
  """
  @spec batch_operate(cluster(), [Batch.t()], batch_parent_opts()) ::
          {:ok, [Aerospike.BatchResult.t()]}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready}
  def batch_operate(cluster, entries, opts \\ [])

  def batch_operate(_cluster, [], opts) when is_list(opts) do
    with {:ok, _policy} <- Policy.batch(opts) do
      {:ok, []}
    end
  end

  def batch_operate(cluster, entries, opts) when is_list(entries) and is_list(opts) do
    with {:ok, _validated} <- Policy.batch(opts),
         {:ok, policy} <- batch_policy(cluster, opts),
         {:ok, command_entries} <- batch_command_entries(entries, policy),
         {:ok, results} <- MixedBatch.execute(cluster, command_entries, opts) do
      {:ok, MixedBatch.to_public_results(results)}
    end
  end

  @doc """
  Returns a lazy `Stream` of records from a scan.

  The returned stream is lazy at the Enumerable boundary, but the current
  runtime drains each node response fully before yielding that node's
  records downstream. It does not promise frame-by-frame backpressure or
  an explicit cancellation API. Pass `node: node_name` in `opts` to scan one
  active node, using a name returned by `node_names/1` or `nodes/1`.
  """
  @spec scan_stream(cluster(), Scan.t(), scan_query_opts()) ::
          {:ok, Enumerable.t()} | {:error, Aerospike.Error.t()}
  def scan_stream(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    ScanOps.stream(cluster, scan, opts)
  end

  @doc """
  Same as `scan_stream/3` but raises on error.
  """
  @spec scan_stream!(cluster(), Scan.t(), scan_query_opts()) :: Enumerable.t()
  def scan_stream!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    case scan_stream(cluster, scan, opts) do
      {:ok, stream} -> stream
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Deprecated alias for `scan_stream!/3`.
  """
  @deprecated "Use scan_stream!/3 instead."
  @spec stream!(cluster(), Scan.t(), scan_query_opts()) :: Enumerable.t()
  def stream!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    scan_stream!(cluster, scan, opts)
  end

  @doc """
  Returns a lazy `Stream` of records from a secondary-index query.

  Like `scan_stream/3`, this is lazy only at the outer Enumerable boundary.
  The current runtime buffers each node's query results before yielding
  them to the caller. Pass `node: node_name` in `opts` to query one active
  node, using a name returned by `node_names/1` or `nodes/1`.
  """
  @spec query_stream(cluster(), Query.t(), scan_query_opts()) ::
          {:ok, Enumerable.t()} | {:error, Aerospike.Error.t()}
  def query_stream(cluster, %Query{} = query, opts \\ []) when is_list(opts) do
    ScanOps.query_stream(cluster, query, opts)
  end

  @doc """
  Same as `query_stream/3` but raises on error.
  """
  @spec query_stream!(cluster(), Query.t(), scan_query_opts()) :: Enumerable.t()
  def query_stream!(cluster, %Query{} = query, opts \\ []) when is_list(opts) do
    case query_stream(cluster, query, opts) do
      {:ok, stream} -> stream
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Eagerly collects scan records into a list.

  Pass `node: node_name` in `opts` to scan one active node, using a name
  returned by `node_names/1` or `nodes/1`.
  """
  @spec scan_all(cluster(), Scan.t(), scan_query_opts()) ::
          {:ok, [Aerospike.Record.t()]} | {:error, Aerospike.Error.t()}
  def scan_all(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    ScanOps.all(cluster, scan, opts)
  end

  @doc """
  Deprecated alias for `scan_all/3`.
  """
  @deprecated "Use scan_all/3 instead."
  @spec all(cluster(), Scan.t(), scan_query_opts()) ::
          {:ok, [Aerospike.Record.t()]} | {:error, Aerospike.Error.t()}
  def all(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    scan_all(cluster, scan, opts)
  end

  @doc """
  Eagerly collects query records into a list.

  `query.max_records` must be set because this helper advances through
  the query in repeated page-sized steps until the cursor is exhausted.
  Pass `node: node_name` in `opts` to query one active node, using a name
  returned by `node_names/1` or `nodes/1`.
  """
  @spec query_all(cluster(), Query.t(), scan_query_opts()) ::
          {:ok, [Aerospike.Record.t()]} | {:error, Aerospike.Error.t()}
  def query_all(cluster, %Query{} = query, opts \\ []) when is_list(opts) do
    ScanOps.query_all(cluster, query, opts)
  end

  @doc """
  Same as `query_all/3` but returns the list or raises `Aerospike.Error`.
  """
  @spec query_all!(cluster(), Query.t(), scan_query_opts()) :: [Aerospike.Record.t()]
  def query_all!(cluster, %Query{} = query, opts \\ []) when is_list(opts) do
    case query_all(cluster, query, opts) do
      {:ok, records} -> records
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Same as `scan_all/3` but returns the list or raises `Aerospike.Error`.
  """
  @spec scan_all!(cluster(), Scan.t(), scan_query_opts()) :: [Aerospike.Record.t()]
  def scan_all!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    case scan_all(cluster, scan, opts) do
      {:ok, records} -> records
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Returns one collected scan page and a resumable cursor when more records
  remain.

  `scan.max_records` is required because it seeds the partition-tracker
  budget for the page walk. On multi-node scans that budget is distributed
  across active nodes, so a page is resumable but not guaranteed to contain
  exactly `scan.max_records` records. The cursor resumes partition progress
  from the prior page; it is not a stable snapshot token. Pass
  `node: node_name` in `opts` to collect a page from one active node, using
  a name returned by `node_names/1` or `nodes/1`.
  """
  @spec scan_page(cluster(), Scan.t(), scan_query_opts()) ::
          {:ok, Page.t()} | {:error, Aerospike.Error.t()}
  def scan_page(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    ScanOps.scan_page(cluster, scan, opts)
  end

  @doc """
  Same as `scan_page/3` but returns the page or raises `Aerospike.Error`.
  """
  @spec scan_page!(cluster(), Scan.t(), scan_query_opts()) :: Page.t()
  def scan_page!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    case scan_page(cluster, scan, opts) do
      {:ok, page} -> page
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Deprecated alias for `scan_all!/3`.
  """
  @deprecated "Use scan_all!/3 instead."
  @spec all!(cluster(), Scan.t(), scan_query_opts()) :: [Aerospike.Record.t()]
  def all!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    scan_all!(cluster, scan, opts)
  end

  @doc """
  Counts scan matches without materializing the records.

  Pass `node: node_name` in `opts` to count records from one active node,
  using a name returned by `node_names/1` or `nodes/1`.
  """
  @spec scan_count(cluster(), Scan.t(), scan_query_opts()) ::
          {:ok, non_neg_integer()} | {:error, Aerospike.Error.t()}
  def scan_count(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    ScanOps.count(cluster, scan, opts)
  end

  @doc """
  Deprecated alias for `scan_count/3`.
  """
  @deprecated "Use scan_count/3 instead."
  @spec count(cluster(), Scan.t(), scan_query_opts()) ::
          {:ok, non_neg_integer()} | {:error, Aerospike.Error.t()}
  def count(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    scan_count(cluster, scan, opts)
  end

  @doc """
  Counts query matches without materializing the records.

  This still walks the query stream and counts client-side. It is not a
  separate server-side count primitive. Pass `node: node_name` in `opts` to
  count records from one active node, using a name returned by `node_names/1`
  or `nodes/1`.
  """
  @spec query_count(cluster(), Query.t(), scan_query_opts()) ::
          {:ok, non_neg_integer()} | {:error, Aerospike.Error.t()}
  def query_count(cluster, %Query{} = query, opts \\ []) when is_list(opts) do
    ScanOps.query_count(cluster, query, opts)
  end

  @doc """
  Same as `query_count/3` but returns the count or raises `Aerospike.Error`.
  """
  @spec query_count!(cluster(), Query.t(), scan_query_opts()) :: non_neg_integer()
  def query_count!(cluster, %Query{} = query, opts \\ []) when is_list(opts) do
    case query_count(cluster, query, opts) do
      {:ok, count} -> count
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Sends one info command to one active cluster node and returns that node's reply.
  """
  @spec info(cluster(), String.t(), admin_opts()) ::
          {:ok, String.t()} | {:error, Aerospike.Error.t()}
  def info(cluster, command, opts \\ [])
      when is_binary(command) and is_list(opts) do
    Admin.info(cluster, command, opts)
  end

  @doc """
  Sends one info command to the named active cluster node and returns that
  node's reply.

  `node_name` must be one of the names returned by `node_names/1` or
  `nodes/1`. Stale or unknown names return an
  `%Aerospike.Error{code: :invalid_node}` instead of falling back to a
  different node.

      {:ok, [node_name | _]} = Aerospike.node_names(:aerospike)
      {:ok, response} = Aerospike.info_node(:aerospike, node_name, "statistics")

  Supported options:

    * `:pool_checkout_timeout` — non-negative pool checkout timeout in
      milliseconds.
  """
  @spec info_node(cluster(), String.t(), String.t(), admin_opts()) ::
          {:ok, String.t()} | {:error, Aerospike.Error.t()}
  def info_node(cluster, node_name, command, opts \\ [])
      when is_binary(node_name) and is_binary(command) and is_list(opts) do
    Admin.info_node(cluster, node_name, command, opts)
  end

  @doc """
  Returns the published active cluster nodes with their direct-connect host and port.
  """
  @spec nodes(cluster()) :: {:ok, [node_info()]}
  def nodes(cluster) do
    {:ok, Cluster.nodes(cluster)}
  end

  @doc """
  Returns the published active cluster node-name snapshot.
  """
  @spec node_names(cluster()) :: {:ok, [String.t()]}
  def node_names(cluster) do
    {:ok, Cluster.node_names(cluster)}
  end

  @doc """
  Returns whether internal runtime metrics are enabled for `cluster`.

  Metrics are opt-in. The collector is initialized at cluster start so the
  config rows exist early, but command, pool, and tender counters remain idle
  until `enable_metrics/2` is called.
  """
  @spec metrics_enabled?(cluster()) :: boolean()
  def metrics_enabled?(cluster) do
    RuntimeMetrics.metrics_enabled?(cluster)
  end

  @doc """
  Enables internal runtime metrics for `cluster`.

  Supported options:

    * `:reset` — boolean. When `true`, clears the existing runtime counters
      before enabling collection.
  """
  @spec enable_metrics(cluster(), metrics_opts()) :: :ok | {:error, Aerospike.Error.t()}
  def enable_metrics(cluster, opts \\ []) when is_list(opts) do
    with :ok <- validate_enable_metrics_opts(opts) do
      RuntimeMetrics.enable(cluster, opts)
    end
  end

  @doc """
  Disables internal runtime metrics for `cluster`.

  Counter rows remain in place so `stats/1` still reports the last collected
  values until metrics are re-enabled or reset.
  """
  @spec disable_metrics(cluster()) :: :ok | {:error, Aerospike.Error.t()}
  def disable_metrics(cluster) do
    RuntimeMetrics.disable(cluster)
  end

  @doc """
  Returns the current internal runtime metrics snapshot for `cluster`.

  The returned map is intentionally limited to the counters and cluster
  metadata the runtime collector actually records today. It is not a
  generalized exporter surface.
  """
  @spec stats(cluster()) :: map()
  def stats(cluster) do
    RuntimeMetrics.stats(cluster)
  end

  @doc """
  Verifies that the active node pools can serve checkouts through the normal path.

  This is an explicit operator action; it does not toggle metrics and it does
  not change the pool startup mode. The current driver pools are already eager at
  cluster start. `warm_up/2` simply proves that the active pools can hand out
  the requested number of workers right now.

  Supported options:

    * `:count` — non-negative integer. `0` (default) means "up to the configured
      pool size per active node".
    * `:pool_checkout_timeout` — non-negative integer timeout in milliseconds
      for each checkout probe.
  """
  @spec warm_up(cluster(), warm_up_opts()) :: {:ok, map()} | {:error, Aerospike.Error.t()}
  def warm_up(cluster, opts \\ []) when is_list(opts) do
    with :ok <- validate_warm_up_opts(opts) do
      Cluster.warm_up(cluster, opts)
    end
  end

  @doc """
  Creates a secondary index and returns a pollable task handle.

  Required options:

    * `:bin` — non-empty bin name.
    * `:name` — non-empty index name.
    * `:type` — one of `:numeric`, `:string`, or `:geo2dsphere`.

  Optional options:

    * `:collection` — one of `:list`, `:mapkeys`, or `:mapvalues`.
    * `:ctx` — non-empty nested CDT path built with `Aerospike.Ctx`.
    * `:pool_checkout_timeout` — non-negative pool checkout timeout in
      milliseconds.

  Geo indexes use `type: :geo2dsphere` and can be queried with
  `Aerospike.Filter.geo_within/2` or `Aerospike.Filter.geo_contains/2`.

      {:ok, task} =
        Aerospike.create_index(cluster, "test", "places",
          bin: "loc",
          name: "places_loc_geo_idx",
          type: :geo2dsphere
        )

      :ok = Aerospike.IndexTask.wait(task)
  """
  @spec create_index(cluster(), String.t(), String.t(), create_index_opts()) ::
          {:ok, IndexTask.t()} | {:error, Aerospike.Error.t()}
  def create_index(cluster, namespace, set, opts \\ [])
      when is_binary(namespace) and is_binary(set) and is_list(opts) do
    Admin.create_index(cluster, namespace, set, opts)
  end

  @doc """
  Creates an expression-backed secondary index and returns a pollable task
  handle.

  Required options:

    * `:name` — non-empty index name.
    * `:type` — one of `:numeric`, `:string`, or `:geo2dsphere`.

  Optional options:

    * `:collection` — one of `:list`, `:mapkeys`, or `:mapvalues`.
    * `:pool_checkout_timeout` — non-negative pool checkout timeout in
      milliseconds.

  The source must be a `%Aerospike.Exp{}` with non-empty wire bytes. Expression
  indexes use the expression as the source and therefore do not accept `:bin`.
  Servers older than Aerospike 8.1 reject expression-backed index creation
  before the create command is sent.

      {:ok, task} =
        Aerospike.create_expression_index(cluster, "test", "users", Exp.int_bin("age"),
          name: "users_age_expr_idx",
          type: :numeric
        )

      :ok = Aerospike.IndexTask.wait(task)
  """
  @spec create_expression_index(
          cluster(),
          String.t(),
          String.t(),
          Exp.t(),
          create_expression_index_opts()
        ) ::
          {:ok, IndexTask.t()} | {:error, Aerospike.Error.t()}
  def create_expression_index(cluster, namespace, set, %Exp{} = expression, opts \\ [])
      when is_binary(namespace) and is_binary(set) and is_list(opts) do
    with {:ok, _policy} <- Policy.expression_index_create(expression, opts) do
      Admin.create_expression_index(cluster, namespace, set, expression, opts)
    end
  end

  @doc """
  Sets or clears the Enterprise XDR filter for one datacenter and namespace.

  Pass a non-empty `%Aerospike.Exp{}` to set the filter, or `nil` to clear the
  current filter. `datacenter` and `namespace` must be non-empty info-command
  identifiers and cannot contain command delimiters.

  Live application requires an Enterprise server with XDR configured. Community
  Edition or unconfigured clusters may reject the command after local
  validation.

      filter = Exp.eq(Exp.int_bin("active"), Exp.int(1))
      :ok = Aerospike.set_xdr_filter(cluster, "dc-west", "test", filter)
      :ok = Aerospike.set_xdr_filter(cluster, "dc-west", "test", nil)
  """
  @spec set_xdr_filter(cluster(), String.t(), String.t(), Exp.t() | nil) ::
          :ok | {:error, Aerospike.Error.t()}
  def set_xdr_filter(cluster, datacenter, namespace, filter)
      when is_binary(datacenter) and is_binary(namespace) do
    with :ok <- Policy.xdr_filter(datacenter, namespace, filter) do
      Admin.set_xdr_filter(cluster, datacenter, namespace, filter)
    end
  end

  @doc """
  Drops a secondary index.
  """
  @spec drop_index(cluster(), String.t(), String.t(), admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def drop_index(cluster, namespace, index_name, opts \\ [])
      when is_binary(namespace) and is_binary(index_name) and is_list(opts) do
    Admin.drop_index(cluster, namespace, index_name, opts)
  end

  @doc """
  Lists the registered server-side UDF packages visible from one active node.

  This is package lifecycle state, not record execution. Use `apply_udf/6`
  to invoke one function against one key.
  """
  @spec list_udfs(cluster(), admin_opts()) :: {:ok, [UDF.t()]} | {:error, Aerospike.Error.t()}
  def list_udfs(cluster, opts \\ []) when is_list(opts) do
    Admin.list_udfs(cluster, opts)
  end

  @doc """
  Uploads a UDF package from inline source or a readable local `.lua` path.

  Returns a pollable `Aerospike.RegisterTask` once the server accepts the
  upload. The package may still be propagating, so call `RegisterTask.wait/2`
  before relying on it from `apply_udf/6` or background UDF jobs.
  """
  @spec register_udf(cluster(), String.t(), String.t(), admin_opts()) ::
          {:ok, RegisterTask.t()} | {:error, Aerospike.Error.t()}
  def register_udf(cluster, path_or_content, server_name, opts \\ [])
      when is_binary(path_or_content) and is_binary(server_name) and is_list(opts) do
    Admin.register_udf(cluster, path_or_content, server_name, opts)
  end

  @doc """
  Removes a registered UDF package by server filename.

  This is idempotent: removing an already-absent package still returns `:ok`.
  """
  @spec remove_udf(cluster(), String.t(), admin_opts()) :: :ok | {:error, Aerospike.Error.t()}
  def remove_udf(cluster, server_name, opts \\ [])
      when is_binary(server_name) and is_list(opts) do
    Admin.remove_udf(cluster, server_name, opts)
  end

  @doc """
  Returns one collected query page and a resumable cursor when more
  records remain.

  `query.max_records` is required because it seeds the partition-tracker
  budget for the page walk. On multi-node queries that budget is
  distributed across active nodes, so a page is resumable but not
  guaranteed to contain exactly `query.max_records` records. The cursor
  resumes partition progress from the prior page; it is not a stable
  snapshot token. Pass `node: node_name` in `opts` to collect a page from one
  active node, using a name returned by `node_names/1` or `nodes/1`.
  """
  @spec query_page(cluster(), Query.t(), scan_query_opts()) ::
          {:ok, Page.t()} | {:error, Aerospike.Error.t()}
  def query_page(cluster, %Query{} = query, opts \\ []) when is_list(opts) do
    ScanOps.query_page(cluster, query, opts)
  end

  @doc """
  Same as `query_page/3` but returns the page or raises `Aerospike.Error`.
  """
  @spec query_page!(cluster(), Query.t(), scan_query_opts()) :: Page.t()
  def query_page!(cluster, %Query{} = query, opts \\ []) when is_list(opts) do
    case query_page(cluster, query, opts) do
      {:ok, page} -> page
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Truncates all records in `namespace`.

  This sends one truncate info command through the shared admin seam. The
  server then distributes the truncate across the cluster.

  Options:

    * `:before` — `%DateTime{}` — truncate only records whose last-update
      time is older than the provided timestamp
    * `:pool_checkout_timeout` — non-negative integer checkout timeout in
      milliseconds for the one-node admin request
  """
  @spec truncate(cluster(), String.t(), truncate_opts()) :: :ok | {:error, Aerospike.Error.t()}
  def truncate(cluster, namespace, opts \\ [])

  def truncate(cluster, namespace, set)
      when is_binary(namespace) and is_binary(set) do
    truncate(cluster, namespace, set, [])
  end

  def truncate(cluster, namespace, opts)
      when is_binary(namespace) and is_list(opts) do
    with {:ok, opts} <- validate_truncate_opts(opts, "Aerospike.truncate/3") do
      Admin.truncate(cluster, namespace, opts)
    end
  end

  @doc """
  Truncates all records in `namespace` and `set`.

  Like `truncate/3`, this uses the shared one-node admin info seam. The optional
  `:before` filter is forwarded to the server as a last-update cutoff.

  Options:

    * `:before` — `%DateTime{}` — truncate only records whose last-update
      time is older than the provided timestamp
    * `:pool_checkout_timeout` — non-negative integer checkout timeout in
      milliseconds for the one-node admin request
  """
  @spec truncate(cluster(), String.t(), String.t(), truncate_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def truncate(cluster, namespace, set, opts)
      when is_binary(namespace) and is_binary(set) and is_list(opts) do
    with {:ok, opts} <- validate_truncate_opts(opts, "Aerospike.truncate/4") do
      Admin.truncate(cluster, namespace, set, opts)
    end
  end

  @doc """
  Creates a password-authenticated security user.

  This requires Aerospike Enterprise with security enabled and a cluster
  connection authenticated as a user that holds the `user-admin` privilege.

  Supported opts are:

    * `:timeout`
    * `:pool_checkout_timeout`
  """
  @spec create_user(cluster(), String.t(), String.t(), [String.t()], security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def create_user(cluster, user_name, password, roles, opts \\ [])
      when is_binary(user_name) and is_binary(password) and is_list(roles) and is_list(opts) do
    with {:ok, role_names} <- validate_role_names(roles),
         {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.create_user(cluster, user_name, password, role_names, opts)
    end
  end

  @doc """
  Creates a PKI-authenticated security user.

  The user is created with a no-password credential and is intended for TLS
  certificate authentication. This requires Aerospike Enterprise with security
  enabled, server support for PKI users, and a cluster connection
  authenticated as a user that holds the `user-admin` privilege.

  Supported opts are:

    * `:timeout`
    * `:pool_checkout_timeout`
  """
  @spec create_pki_user(cluster(), String.t(), [String.t()], security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def create_pki_user(cluster, user_name, roles, opts \\ [])
      when is_binary(user_name) and is_list(roles) and is_list(opts) do
    with {:ok, role_names} <- validate_role_names(roles),
         {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.create_pki_user(cluster, user_name, role_names, opts)
    end
  end

  @doc """
  Drops a security user.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec drop_user(cluster(), String.t(), security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def drop_user(cluster, user_name, opts \\ [])
      when is_binary(user_name) and is_list(opts) do
    with {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.drop_user(cluster, user_name, opts)
    end
  end

  @doc """
  Changes a security user's password.

  When `user_name` matches the credentials configured on `cluster`, the driver
  uses the self-service password-change command and rotates the running
  cluster's in-memory credentials for future reconnects. For other users it
  uses the user-admin password-set command.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec change_password(cluster(), String.t(), String.t(), security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def change_password(cluster, user_name, password, opts \\ [])
      when is_binary(user_name) and is_binary(password) and is_list(opts) do
    with {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.change_password(cluster, user_name, password, opts)
    end
  end

  @doc """
  Grants roles to a security user.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec grant_roles(cluster(), String.t(), [String.t()], security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def grant_roles(cluster, user_name, roles, opts \\ [])
      when is_binary(user_name) and is_list(roles) and is_list(opts) do
    with {:ok, role_names} <- validate_role_names(roles),
         {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.grant_roles(cluster, user_name, role_names, opts)
    end
  end

  @doc """
  Revokes roles from a security user.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec revoke_roles(cluster(), String.t(), [String.t()], security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def revoke_roles(cluster, user_name, roles, opts \\ [])
      when is_binary(user_name) and is_list(roles) and is_list(opts) do
    with {:ok, role_names} <- validate_role_names(roles),
         {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.revoke_roles(cluster, user_name, role_names, opts)
    end
  end

  @doc """
  Queries one security user.

  Returns `{:ok, nil}` when the named user does not exist.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec query_user(cluster(), String.t(), security_admin_opts()) ::
          {:ok, user_info() | nil} | {:error, Aerospike.Error.t()}
  def query_user(cluster, user_name, opts \\ [])
      when is_binary(user_name) and is_list(opts) do
    with {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.query_user(cluster, user_name, opts)
    end
  end

  @doc """
  Queries all security users visible to the authenticated cluster user.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec query_users(cluster(), security_admin_opts()) ::
          {:ok, [user_info()]} | {:error, Aerospike.Error.t()}
  def query_users(cluster, opts \\ []) when is_list(opts) do
    with {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.query_users(cluster, opts)
    end
  end

  @doc """
  Creates a security role.

  This requires Aerospike Enterprise with security enabled.

  Supported opts are:

    * `:whitelist` — list of client address strings
    * `:read_quota` — non-negative integer operations-per-second limit
    * `:write_quota` — non-negative integer operations-per-second limit
    * `:timeout`
    * `:pool_checkout_timeout`
  """
  @spec create_role(cluster(), String.t(), [privilege()], create_role_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def create_role(cluster, role_name, privileges, opts \\ [])
      when is_binary(role_name) and is_list(privileges) and is_list(opts) do
    with {:ok, privileges} <- validate_privileges(privileges),
         {:ok, {whitelist, read_quota, write_quota, call_opts}} <-
           validate_create_role_opts(opts),
         {:ok, _policy} <- Policy.security_admin(call_opts) do
      Admin.create_role(
        cluster,
        role_name,
        privileges,
        whitelist,
        read_quota,
        write_quota,
        call_opts
      )
    end
  end

  @doc """
  Drops a security role.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec drop_role(cluster(), String.t(), security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def drop_role(cluster, role_name, opts \\ [])
      when is_binary(role_name) and is_list(opts) do
    with {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.drop_role(cluster, role_name, opts)
    end
  end

  @doc """
  Sets or clears a security role's client-address whitelist.

  Pass an empty list to clear the role's whitelist. This requires Aerospike
  Enterprise with security enabled.
  """
  @spec set_whitelist(cluster(), String.t(), [String.t()], security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def set_whitelist(cluster, role_name, whitelist, opts \\ [])
      when is_binary(role_name) and is_list(opts) do
    with {:ok, whitelist} <- validate_role_whitelist_value(whitelist),
         {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.set_whitelist(cluster, role_name, whitelist, opts)
    end
  end

  @doc """
  Sets read and write quota limits for a security role.

  Pass `0` for either quota to clear that limit. Quotas require server security
  configuration with quotas enabled.
  """
  @spec set_quotas(
          cluster(),
          String.t(),
          non_neg_integer(),
          non_neg_integer(),
          security_admin_opts()
        ) ::
          :ok | {:error, Aerospike.Error.t()}
  def set_quotas(cluster, role_name, read_quota, write_quota, opts \\ [])
      when is_binary(role_name) and is_list(opts) do
    with {:ok, read_quota} <- validate_role_quota_value(read_quota, :read_quota),
         {:ok, write_quota} <- validate_role_quota_value(write_quota, :write_quota),
         {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.set_quotas(cluster, role_name, read_quota, write_quota, opts)
    end
  end

  @doc """
  Grants privileges to a security role.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec grant_privileges(cluster(), String.t(), [privilege()], security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def grant_privileges(cluster, role_name, privileges, opts \\ [])
      when is_binary(role_name) and is_list(privileges) and is_list(opts) do
    with {:ok, privileges} <- validate_privileges(privileges),
         {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.grant_privileges(cluster, role_name, privileges, opts)
    end
  end

  @doc """
  Revokes privileges from a security role.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec revoke_privileges(cluster(), String.t(), [privilege()], security_admin_opts()) ::
          :ok | {:error, Aerospike.Error.t()}
  def revoke_privileges(cluster, role_name, privileges, opts \\ [])
      when is_binary(role_name) and is_list(privileges) and is_list(opts) do
    with {:ok, privileges} <- validate_privileges(privileges),
         {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.revoke_privileges(cluster, role_name, privileges, opts)
    end
  end

  @doc """
  Queries one security role.

  Returns `{:ok, nil}` when the named role does not exist.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec query_role(cluster(), String.t(), security_admin_opts()) ::
          {:ok, role_info() | nil} | {:error, Aerospike.Error.t()}
  def query_role(cluster, role_name, opts \\ [])
      when is_binary(role_name) and is_list(opts) do
    with {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.query_role(cluster, role_name, opts)
    end
  end

  @doc """
  Queries all security roles visible to the authenticated cluster user.

  This requires Aerospike Enterprise with security enabled.
  """
  @spec query_roles(cluster(), security_admin_opts()) ::
          {:ok, [role_info()]} | {:error, Aerospike.Error.t()}
  def query_roles(cluster, opts \\ []) when is_list(opts) do
    with {:ok, _policy} <- Policy.security_admin(opts) do
      Admin.query_roles(cluster, opts)
    end
  end

  @doc """
  Streams aggregate query values over the same node-buffered query
  runtime used by `query_stream/3`.

  The returned stream yields the aggregate values emitted by the server. It
  does not run local Lua finalization. Use `query_aggregate_result/6` when
  the caller wants one locally finalized aggregate result.
  """
  @spec query_aggregate(cluster(), Query.t(), String.t(), String.t(), list(), scan_query_opts()) ::
          {:ok, Enumerable.t()} | {:error, Aerospike.Error.t()}
  def query_aggregate(cluster, %Query{} = query, package, function, args, opts \\ [])
      when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do
    ScanOps.query_aggregate(cluster, query, package, function, args, opts)
  end

  @doc """
  Returns one finalized aggregate query result.

  This runs the same server aggregate query as `query_aggregate/6`, then
  locally executes the package's Lua stream finalization over the
  server-emitted aggregate values. Use `query_aggregate/6` when callers need
  the partial server values themselves.

  The local Lua package source must be supplied with exactly one of:

    * `:source` - inline Lua source as a binary
    * `:source_path` - readable local Lua source path

  Source loading is local-only and separate from server UDF registration. The
  client does not derive a local path from `package` and does not fetch source
  from the server. Source option errors, unsupported local argument values, and
  `node: node_name` fail with `{:error, %Aerospike.Error{code:
  :invalid_argument}}` before opening the server query stream.

  The local reducer runs in a fresh bounded Lua state. The existing `:timeout`
  option bounds local execution when it is a positive integer; otherwise a
  finite default is used. Supported local stream helpers are `map`, `filter`,
  `aggregate`, and `reduce`. Logging helpers are no-ops. Filesystem, OS,
  package loading, dynamic loading, debug access, `require`, `groupby`, `list`,
  `bytes`, and record/database mutation or lookup helpers fail explicitly with
  `%Aerospike.Error{code: :query_generic}`.

  Values crossing the local Lua boundary are limited to `nil`, booleans,
  integers, floats, binaries, lists, and maps with scalar keys. Blob, geo, raw,
  HLL, and other unsupported values fail instead of being coerced.

  Example:

      query =
        Aerospike.Query.new("test", "users")
        |> Aerospike.Query.where(Aerospike.Filter.range("age", 18, 65))

      Aerospike.query_aggregate_result(
        :aerospike,
        query,
        "user_stats",
        "sum_age",
        ["age"],
        source_path: "priv/udf/user_stats.lua",
        timeout: 10_000
      )

  Returns `{:ok, nil}` when local finalization produces no value. Returns an
  error when finalization produces multiple values or local Lua execution fails.
  """
  @spec query_aggregate_result(
          cluster(),
          Query.t(),
          String.t(),
          String.t(),
          list(),
          aggregate_result_opts()
        ) ::
          {:ok, term() | nil} | {:error, Aerospike.Error.t()}
  def query_aggregate_result(cluster, %Query{} = query, package, function, args, opts \\ [])
      when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do
    ScanOps.query_aggregate_result(cluster, query, package, function, args, opts)
  end

  @doc """
  Same as `query_aggregate_result/6` but returns the value or raises
  `Aerospike.Error`.
  """
  @spec query_aggregate_result!(
          cluster(),
          Query.t(),
          String.t(),
          String.t(),
          list(),
          aggregate_result_opts()
        ) ::
          term() | nil
  def query_aggregate_result!(cluster, %Query{} = query, package, function, args, opts \\ [])
      when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do
    case query_aggregate_result(cluster, query, package, function, args, opts) do
      {:ok, result} -> result
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Starts a background query write job that applies the given operations.

  This returns a pollable task handle, not a resumable record stream.
  Pass `node: node_name` in `opts` to start the job on one active node,
  using a name returned by `node_names/1` or `nodes/1`.
  """
  @spec query_execute(cluster(), Query.t(), list(), scan_query_opts()) ::
          {:ok, ExecuteTask.t()} | {:error, Aerospike.Error.t()}
  def query_execute(cluster, %Query{} = query, ops, opts \\ [])
      when is_list(ops) and is_list(opts) do
    ScanOps.query_execute(cluster, query, ops, opts)
  end

  @doc """
  Starts a background query UDF job.

  This returns a pollable task handle, not a resumable record stream.
  Pass `node: node_name` in `opts` to start the job on one active node,
  using a name returned by `node_names/1` or `nodes/1`.
  """
  @spec query_udf(cluster(), Query.t(), String.t(), String.t(), list(), scan_query_opts()) ::
          {:ok, ExecuteTask.t()} | {:error, Aerospike.Error.t()}
  def query_udf(cluster, %Query{} = query, package, function, args, opts \\ [])
      when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do
    ScanOps.query_udf(cluster, query, package, function, args, opts)
  end

  @doc """
  Executes one record UDF against `key`.

  This is a single-record command, not a background query job. It accepts the
  same write opts as the unary write family:

    * `:timeout`
    * `:socket_timeout`
    * `:max_retries`
    * `:sleep_between_retries_ms`
    * `:ttl`
    * `:generation`
    * `:generation_policy`
    * `:exists`
    * `:commit_level`
    * `:durable_delete`
    * `:respond_per_op`
    * `:send_key`
    * `:read_mode_ap`
    * `:read_mode_sc`
    * `:read_touch_ttl_percent`
    * `:use_compression`
    * `:filter`
    * `:txn`

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.

  Transport failures are not retried automatically once the request is on the
  wire, because record UDFs may already have produced server-side effects.
  Package lifecycle lives on `register_udf/*`, `remove_udf/*`, and `list_udfs/2`.
  """
  @spec apply_udf(cluster(), Key.key_input(), String.t(), String.t(), list(), write_opts()) ::
          {:ok, term()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def apply_udf(cluster, key, package, function, args, opts \\ [])
      when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do
    with {:ok, key} <- coerce_key(key) do
      ApplyUdf.execute(cluster, key, package, function, args, opts)
    end
  end

  @doc """
  Commits a transaction on the named cluster `cluster`.

  This only works for a transaction handle whose tracking row is already
  initialized on `cluster`. A fresh `%Aerospike.Txn{}` is not enough by itself.
  In the current driver, public code initializes that runtime state only when
  `transaction/2` or `transaction/3` enters its callback. Transaction
  tracking is keyed off the started cluster name, so this helper currently
  requires that registered atom.
  """
  @spec commit(named_cluster(), Txn.t()) ::
          {:ok, :committed | :already_committed} | {:error, Aerospike.Error.t()}
  def commit(cluster, %Txn{} = txn) when is_atom(cluster) do
    TxnRoll.commit(cluster, txn, [])
  end

  @doc """
  Aborts a transaction on the named cluster `cluster`.

  Like `commit/2`, this requires a handle with initialized runtime tracking on
  `cluster`. It is for an already-open transaction; it does not create one,
  and it currently requires the registered cluster atom.
  """
  @spec abort(named_cluster(), Txn.t()) ::
          {:ok, :aborted | :already_aborted} | {:error, Aerospike.Error.t()}
  def abort(cluster, %Txn{} = txn) when is_atom(cluster) do
    TxnRoll.abort(cluster, txn, [])
  end

  @doc """
  Returns the current state of a transaction on the named cluster `cluster`.

  This reflects only the in-flight states backed by the runtime tracking row.
  After commit or abort, the driver cleans that row up, so `txn_status/2`
  returns an error instead of a terminal `:committed` or `:aborted` state.
  Like the other transaction lifecycle helpers, this currently requires the
  registered cluster atom.
  """
  @spec txn_status(named_cluster(), Txn.t()) ::
          {:ok, :open | :verified | :committed | :aborted}
          | {:error, Aerospike.Error.t()}
  def txn_status(cluster, %Txn{} = txn) when is_atom(cluster) do
    TxnRoll.txn_status(cluster, txn)
  end

  @doc """
  Runs a function within a new transaction on the named cluster `cluster`.

  The callback owns the public transaction lifecycle. The driver initializes the
  runtime tracking row before invoking `fun`, then commits on success or aborts
  on any failure path. Do not call `commit/2` or `abort/2` from inside the
  callback.

  The `%Aerospike.Txn{}` passed to `fun` is safe only for sequential use within
  that transaction. Do not share it across concurrent processes, and do not use
  scans or queries with it; the current transaction proof covers only
  transaction-aware single-record commands. This helper currently requires the
  registered cluster atom.
  """
  @spec transaction(named_cluster(), (Txn.t() -> term())) ::
          {:ok, term()} | {:error, Aerospike.Error.t()}
  def transaction(cluster, fun) when is_atom(cluster) and is_function(fun, 1) do
    TxnRoll.transaction(cluster, [], fun)
  end

  @doc """
  Runs a function within a transaction on the named cluster `cluster` using a
  provided handle or options.

  When `txn_or_opts` is a `%Aerospike.Txn{}`, the driver initializes fresh
  runtime tracking for that handle on `cluster` at callback entry. Reusing the
  same handle concurrently or against another cluster is unsupported. This
  helper currently requires the registered cluster atom.
  """
  @spec transaction(named_cluster(), Txn.t() | transaction_opts(), (Txn.t() -> term())) ::
          {:ok, term()} | {:error, Aerospike.Error.t()}
  def transaction(cluster, txn_or_opts, fun)
      when is_atom(cluster) and is_function(fun, 1) do
    TxnRoll.transaction(cluster, txn_or_opts, fun)
  end

  @doc """
  Same as `scan_count/3` but returns the count or raises `Aerospike.Error`.
  """
  @spec scan_count!(cluster(), Scan.t(), scan_query_opts()) :: non_neg_integer()
  def scan_count!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    case scan_count(cluster, scan, opts) do
      {:ok, count} -> count
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Deprecated alias for `scan_count!/3`.
  """
  @deprecated "Use scan_count!/3 instead."
  @spec count!(cluster(), Scan.t(), scan_query_opts()) :: non_neg_integer()
  def count!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do
    scan_count!(cluster, scan, opts)
  end

  @doc """
  Writes `bins` for `key` to `cluster`.

  The driver accepts only a non-empty bin map and only scalar, list, map,
  bytes, geo, and HyperLogLog values supported by the command encoder.
  Supported write opts include:

    * `:timeout`
    * `:socket_timeout`
    * `:max_retries`
    * `:sleep_between_retries_ms`
    * `:ttl`
    * `:generation`
    * `:generation_policy` — `:none`, `:expect_equal`, or `:expect_gt`
    * `:exists` — one of `:update`, `:update_only`,
      `:create_or_replace`, `:replace_only`, or `:create_only`
    * `:commit_level` — `:all` or `:master`
    * `:durable_delete` — when `true`, write/delete commands that remove a
      record ask the server to leave a tombstone
    * `:respond_per_op`
    * `:send_key`
    * `:read_mode_ap`
    * `:read_mode_sc`
    * `:read_touch_ttl_percent`
    * `:use_compression`
    * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter
      expression, or `nil` for no filter

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec put(cluster(), Key.key_input(), Aerospike.Record.bins_input(), write_opts()) ::
          {:ok, Aerospike.Record.metadata()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def put(cluster, key, bins, opts \\ []) do
    with {:ok, key} <- coerce_key(key) do
      Put.execute(cluster, key, bins, opts)
    end
  end

  @doc """
  Sends a caller-built single-record write/delete frame for `key`.

  This is an advanced escape hatch for tooling, proxy, and replay scenarios.
  `payload` must be a complete Aerospike wire frame for one write-shaped
  command. The client uses `key` only to choose the write partition owner,
  forwards `payload` unchanged, and parses only the standard write response.

  The payload must already contain every server-visible write attribute, such
  as generation, TTL, send-key, delete flags, filters, and any transaction
  fields. Passing `:txn` validates the transaction option shape but does not
  register the key with the transaction monitor or add transaction fields.

  Supported write opts are validated for routing and I/O budgets:

    * `:timeout`
    * `:socket_timeout`
    * `:max_retries`
    * `:sleep_between_retries_ms`
    * `:ttl`
    * `:generation`
    * `:generation_policy`
    * `:exists`
    * `:commit_level`
    * `:durable_delete`
    * `:respond_per_op`
    * `:send_key`
    * `:use_compression`
    * `:filter`
    * `:txn`

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec put_payload(cluster(), Key.key_input(), binary(), write_opts()) ::
          :ok
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def put_payload(cluster, key, payload, opts \\ [])
      when is_binary(payload) and is_list(opts) do
    with {:ok, key} <- coerce_key(key) do
      PutPayload.execute(cluster, key, payload, opts)
    end
  end

  @doc """
  Same as `put_payload/4` but returns `:ok` or raises `Aerospike.Error`.
  """
  @spec put_payload!(cluster(), Key.key_input(), binary(), write_opts()) :: :ok
  def put_payload!(cluster, key, payload, opts \\ [])
      when is_binary(payload) and is_list(opts) do
    case put_payload(cluster, key, payload, opts) do
      :ok -> :ok
      {:error, %Aerospike.Error{} = err} -> raise err
    end
  end

  @doc """
  Atomically adds numeric deltas in `bins` for `key`.

  This is a thin unary write helper over the same routed write path as `put/4`.
  The return shape stays aligned with the write family and does not expose
  `operate/4` record results.

  Supported write opts include:

    * `:timeout`
    * `:socket_timeout`
    * `:max_retries`
    * `:sleep_between_retries_ms`
    * `:ttl`
    * `:generation`
    * `:generation_policy` — `:none`, `:expect_equal`, or `:expect_gt`
    * `:exists` — one of `:update`, `:update_only`,
      `:create_or_replace`, `:replace_only`, or `:create_only`
    * `:commit_level` — `:all` or `:master`
    * `:durable_delete` — when `true`, write/delete commands that remove a
      record ask the server to leave a tombstone
    * `:respond_per_op`
    * `:send_key`
    * `:read_mode_ap`
    * `:read_mode_sc`
    * `:read_touch_ttl_percent`
    * `:use_compression`
    * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter
      expression, or `nil` for no filter

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec add(cluster(), Key.key_input(), Aerospike.Record.bins_input(), write_opts()) ::
          {:ok, Aerospike.Record.metadata()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def add(cluster, key, bins, opts \\ []) do
    with {:ok, key} <- coerce_key(key) do
      WriteOp.execute(cluster, key, :add, bins, opts)
    end
  end

  @doc """
  Atomically appends string suffixes in `bins` for `key`.

  This stays on the unary write path and returns write metadata, not an
  `operate/4` record payload.

  Supported write opts include:

    * `:timeout`
    * `:socket_timeout`
    * `:max_retries`
    * `:sleep_between_retries_ms`
    * `:ttl`
    * `:generation`
    * `:generation_policy` — `:none`, `:expect_equal`, or `:expect_gt`
    * `:exists` — one of `:update`, `:update_only`,
      `:create_or_replace`, `:replace_only`, or `:create_only`
    * `:commit_level` — `:all` or `:master`
    * `:durable_delete` — when `true`, write/delete commands that remove a
      record ask the server to leave a tombstone
    * `:respond_per_op`
    * `:send_key`
    * `:read_mode_ap`
    * `:read_mode_sc`
    * `:read_touch_ttl_percent`
    * `:use_compression`
    * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter
      expression, or `nil` for no filter

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec append(cluster(), Key.key_input(), Aerospike.Record.bins_input(), write_opts()) ::
          {:ok, Aerospike.Record.metadata()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def append(cluster, key, bins, opts \\ []) do
    with {:ok, key} <- coerce_key(key) do
      WriteOp.execute(cluster, key, :append, bins, opts)
    end
  end

  @doc """
  Atomically prepends string prefixes in `bins` for `key`.

  This stays on the unary write path and returns write metadata, not an
  `operate/4` record payload.

  Supported write opts include:

    * `:timeout`
    * `:socket_timeout`
    * `:max_retries`
    * `:sleep_between_retries_ms`
    * `:ttl`
    * `:generation`
    * `:generation_policy` — `:none`, `:expect_equal`, or `:expect_gt`
    * `:exists` — one of `:update`, `:update_only`,
      `:create_or_replace`, `:replace_only`, or `:create_only`
    * `:commit_level` — `:all` or `:master`
    * `:durable_delete` — when `true`, write/delete commands that remove a
      record ask the server to leave a tombstone
    * `:respond_per_op`
    * `:send_key`
    * `:read_mode_ap`
    * `:read_mode_sc`
    * `:read_touch_ttl_percent`
    * `:use_compression`
    * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter
      expression, or `nil` for no filter

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec prepend(cluster(), Key.key_input(), Aerospike.Record.bins_input(), write_opts()) ::
          {:ok, Aerospike.Record.metadata()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def prepend(cluster, key, bins, opts \\ []) do
    with {:ok, key} <- coerce_key(key) do
      WriteOp.execute(cluster, key, :prepend, bins, opts)
    end
  end

  @doc """
  Returns whether `key` exists in `cluster` without reading bins.

  Supported read opts are `:timeout`, `:socket_timeout`, `:max_retries`,
  `:sleep_between_retries_ms`, `:replica_policy`, `:read_mode_ap`,
  `:read_mode_sc`, `:read_touch_ttl_percent`, `:send_key`,
  `:use_compression`, and `:filter`.

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec exists(cluster(), Key.key_input(), read_opts()) ::
          {:ok, boolean()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def exists(cluster, key, opts \\ []) do
    with {:ok, key} <- coerce_key(key) do
      Exists.execute(cluster, key, opts)
    end
  end

  @doc """
  Updates `key`'s header metadata in `cluster`.

  Supported write opts are `:timeout`, `:socket_timeout`, `:max_retries`,
  `:sleep_between_retries_ms`, `:ttl`, `:generation`, `:generation_policy`,
  `:exists`, `:commit_level`, `:durable_delete`, `:respond_per_op`,
  `:send_key`, `:read_mode_ap`, `:read_mode_sc`,
  `:read_touch_ttl_percent`, `:use_compression`, and `:filter`.

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec touch(cluster(), Key.key_input(), write_opts()) ::
          {:ok, Aerospike.Record.metadata()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def touch(cluster, key, opts \\ []) do
    with {:ok, key} <- coerce_key(key) do
      Touch.execute(cluster, key, opts)
    end
  end

  @doc """
  Deletes `key` from `cluster`.

  Returns `{:ok, true}` when a record was deleted and `{:ok, false}`
  when the key was already absent. Supported write opts are `:timeout`,
  `:socket_timeout`, `:max_retries`, `:sleep_between_retries_ms`, `:ttl`,
  `:generation`, `:generation_policy`, `:exists`, `:commit_level`,
  `:durable_delete`, `:respond_per_op`, `:send_key`, `:read_mode_ap`,
  `:read_mode_sc`, `:read_touch_ttl_percent`, `:use_compression`, and
  `:filter`.

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec delete(cluster(), Key.key_input(), write_opts()) ::
          {:ok, boolean()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def delete(cluster, key, opts \\ []) do
    with {:ok, key} <- coerce_key(key) do
      Delete.execute(cluster, key, opts)
    end
  end

  @doc """
  Runs a constrained unary operate list for `key`.

  Supported operations:

    * `{:write, bin, value}` — simple bin write
    * `{:read, bin}` — simple bin read
    * `{:add, bin, delta}` — numeric increment
    * `{:append, bin, suffix}` — string suffix mutation
    * `{:prepend, bin, prefix}` — string prefix mutation
    * `:touch` — refresh record metadata
    * `:delete` — remove the record

  The command routes per input batch: read-only lists use read routing;
  any list that includes a write uses write routing.

  Supported opts are `:timeout`, `:socket_timeout`, `:max_retries`,
  `:sleep_between_retries_ms`, `:ttl`, `:generation`, `:generation_policy`,
  `:exists`, `:commit_level`, `:durable_delete`, `:respond_per_op`,
  `:send_key`, `:read_mode_ap`, `:read_mode_sc`,
  `:read_touch_ttl_percent`, `:use_compression`, and `:filter`.

  Accepted operations include the simple tuple form plus the public
  `Aerospike.Op` helpers for primitive, CDT, bit, HyperLogLog, and expression
  operations. Returned operation values are accumulated into
  `%Aerospike.Record{bins: map}` by bin name.

      Aerospike.operate(cluster, key, [
        Aerospike.Op.Exp.read("projected", Aerospike.Exp.int_bin("count")),
        Aerospike.Op.Exp.write("computed", Aerospike.Exp.int(99))
      ])

      Aerospike.operate(cluster, key, [
        Aerospike.Op.List.append("profile", "signed-in",
          ctx: [Aerospike.Ctx.map_key("events")]
        )
      ])

      Aerospike.operate(cluster, key, [
        Aerospike.Op.Bit.count("flags", 0, 8),
        Aerospike.Op.HLL.get_count("visitors")
      ])

  Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`.
  """
  @spec operate(cluster(), Key.key_input(), [operate_operation()], write_opts()) ::
          {:ok, Aerospike.Record.t()}
          | {:error, Aerospike.Error.t()}
          | {:error, :cluster_not_ready | :no_master | :unknown_node}
  def operate(cluster, key, operations, opts \\ []) do
    with {:ok, key} <- coerce_key(key) do
      Operate.execute(cluster, key, operations, opts)
    end
  end

  defp coerce_key(key) do
    {:ok, Key.coerce!(key)}
  rescue
    err in ArgumentError ->
      {:error, Error.from_result_code(:invalid_argument, message: err.message)}
  end

  defp coerce_keys(keys) do
    {:ok, Enum.map(keys, &Key.coerce!/1)}
  rescue
    err in ArgumentError ->
      {:error, Error.from_result_code(:invalid_argument, message: err.message)}
  end

  defp batch_policy(cluster, opts) do
    cluster
    |> Cluster.retry_policy()
    |> Policy.batch(opts)
  end

  defp batch_command_entries(entries, %Policy.Batch{} = policy) do
    entries
    |> Enum.with_index()
    |> Enum.reduce_while({:ok, []}, &put_batch_command_entry(&1, &2, policy))
    |> case do
      {:ok, command_entries} -> {:ok, Enum.reverse(command_entries)}
      {:error, %Error{}} = err -> err
    end
  end

  defp put_batch_command_entry({entry, index}, {:ok, acc}, %Policy.Batch{} = policy) do
    case batch_command_entry(entry, index, policy) do
      {:ok, %Entry{} = command_entry} -> {:cont, {:ok, [command_entry | acc]}}
      {:error, %Error{}} = err -> {:halt, err}
    end
  end

  defp batch_command_entry(
         %Batch.Read{key: %Key{} = key, opts: opts},
         index,
         %Policy.Batch{} = policy
       ) do
    with {:ok, read_policy} <- Policy.batch_record_read(opts) do
      {:ok,
       %Entry{
         index: index,
         key: key,
         kind: :read,
         dispatch: {:read, policy.retry.replica_policy, 0},
         payload: batch_read_payload(read_policy)
       }}
    end
  end

  defp batch_command_entry(
         %Batch.Put{key: %Key{} = key, bins: bins, opts: opts},
         index,
         %Policy.Batch{}
       )
       when is_map(bins) do
    with {:ok, write_policy} <- Policy.batch_record_write(opts),
         {:ok, operations} <- batch_put_operations(bins) do
      {:ok,
       %Entry{
         index: index,
         key: key,
         kind: :put,
         dispatch: :write,
         payload: Map.put(batch_write_payload(write_policy), :operations, operations)
       }}
    end
  end

  defp batch_command_entry(%Batch.Delete{key: %Key{} = key, opts: opts}, index, %Policy.Batch{}) do
    with {:ok, write_policy} <- Policy.batch_record_write(opts) do
      {:ok,
       %Entry{
         index: index,
         key: key,
         kind: :delete,
         dispatch: :write,
         payload: batch_write_payload(write_policy)
       }}
    end
  end

  defp batch_command_entry(
         %Batch.Operate{key: %Key{} = key, operations: operations, opts: opts},
         index,
         %Policy.Batch{} = policy
       )
       when is_list(operations) do
    with :ok <- validate_batch_operations(operations) do
      flags = OperateFlags.scan_ops(operations)
      payload = batch_operate_payload(flags, opts)

      with {:ok, payload} <- payload do
        {:ok,
         %Entry{
           index: index,
           key: key,
           kind: :operate,
           dispatch: batch_operate_dispatch(flags, policy),
           payload: Map.merge(payload, %{operations: operations, flags: flags})
         }}
      end
    end
  end

  defp batch_command_entry(
         %Batch.UDF{
           key: %Key{} = key,
           package: package,
           function: function,
           args: args,
           opts: opts
         },
         index,
         %Policy.Batch{}
       )
       when is_binary(package) and is_binary(function) and is_list(args) do
    with {:ok, write_policy} <- Policy.batch_record_write(opts) do
      {:ok,
       %Entry{
         index: index,
         key: key,
         kind: :udf,
         dispatch: :write,
         payload:
           write_policy
           |> batch_write_payload()
           |> Map.merge(%{package: package, function: function, args: args})
       }}
    end
  end

  defp batch_command_entry(_entry, _index, %Policy.Batch{}) do
    {:error,
     Error.from_result_code(:invalid_argument,
       message: "Aerospike.batch_operate/3 expects entries built by Aerospike.Batch"
     )}
  end

  defp batch_read_payload(%Policy.BatchRead{} = policy) do
    Map.take(policy, [
      :filter,
      :read_mode_ap,
      :read_mode_sc,
      :read_touch_ttl_percent
    ])
  end

  defp batch_write_payload(%Policy.BatchWrite{} = policy) do
    Map.take(policy, [
      :ttl,
      :generation,
      :generation_policy,
      :filter,
      :exists,
      :commit_level,
      :durable_delete,
      :respond_per_op,
      :send_key,
      :read_mode_ap,
      :read_mode_sc,
      :read_touch_ttl_percent
    ])
  end

  defp batch_operate_payload(%{has_write?: true}, opts) do
    with {:ok, write_policy} <- Policy.batch_record_write(opts) do
      {:ok, batch_write_payload(write_policy)}
    end
  end

  defp batch_operate_payload(%{has_write?: false}, opts) do
    with {:ok, read_policy} <- Policy.batch_record_read(opts) do
      {:ok, batch_read_payload(read_policy)}
    end
  end

  defp batch_operate_dispatch(%{has_write?: true}, %Policy.Batch{}), do: :write

  defp batch_operate_dispatch(%{has_write?: false}, %Policy.Batch{} = policy) do
    {:read, policy.retry.replica_policy, 0}
  end

  defp batch_put_operations(bins) when map_size(bins) == 0 do
    {:error,
     Error.from_result_code(:invalid_argument,
       message: "batch put requires at least one bin write"
     )}
  end

  defp batch_put_operations(bins) do
    bins
    |> Enum.reduce_while({:ok, []}, fn {bin_name, value}, {:ok, acc} ->
      case Operation.write(normalize_batch_bin_name(bin_name), value) do
        {:ok, operation} -> {:cont, {:ok, [operation | acc]}}
        {:error, %Error{}} = err -> {:halt, err}
      end
    end)
    |> case do
      {:ok, operations} -> {:ok, Enum.reverse(operations)}
      {:error, %Error{}} = err -> err
    end
  end

  defp validate_batch_operations([]) do
    {:error,
     Error.from_result_code(:invalid_argument,
       message: "batch operate requires at least one operation"
     )}
  end

  defp validate_batch_operations(operations) do
    if Enum.all?(operations, &match?(%Operation{}, &1)) do
      :ok
    else
      {:error,
       Error.from_result_code(:invalid_argument,
         message: "batch operate expects a list of Aerospike.Op operations"
       )}
    end
  end

  defp normalize_batch_bin_name(bin_name) when is_atom(bin_name), do: Atom.to_string(bin_name)
  defp normalize_batch_bin_name(bin_name), do: bin_name

  defp validate_truncate_opts(opts, callsite) when is_list(opts) do
    with :ok <- validate_truncate_opt_keys(opts, callsite),
         {:ok, _policy} <- Policy.admin_info(Keyword.delete(opts, :before)),
         :ok <- validate_truncate_before(opts) do
      {:ok, opts}
    end
  end

  defp validate_truncate_opt_keys([], _callsite), do: :ok

  defp validate_truncate_opt_keys([{key, _value} | rest], callsite)
       when key in [:before, :pool_checkout_timeout] do
    validate_truncate_opt_keys(rest, callsite)
  end

  defp validate_truncate_opt_keys([{key, _value} | _rest], callsite) do
    {:error,
     Error.from_result_code(:invalid_argument,
       message:
         "#{callsite} supports only :before and :pool_checkout_timeout options, got #{inspect(key)}"
     )}
  end

  defp validate_truncate_before(opts) do
    case Keyword.get(opts, :before) do
      nil ->
        :ok

      %DateTime{} ->
        :ok

      other ->
        {:error,
         Error.from_result_code(:invalid_argument,
           message: ":before must be a DateTime, got: #{inspect(other)}"
         )}
    end
  end

  defp validate_role_names(roles) when is_list(roles) do
    if Enum.all?(roles, &is_binary/1) do
      {:ok, roles}
    else
      {:error,
       Error.from_result_code(:invalid_argument, message: "roles must be a list of strings")}
    end
  end

  defp validate_privileges(privileges) when is_list(privileges) do
    if Enum.all?(privileges, &match?(%Privilege{}, &1)) do
      {:ok, privileges}
    else
      {:error,
       Error.from_result_code(:invalid_argument,
         message: "privileges must be a list of %Aerospike.Privilege{} structs"
       )}
    end
  end

  defp validate_create_role_opts(opts) when is_list(opts) do
    supported = [:whitelist, :read_quota, :write_quota, :timeout, :pool_checkout_timeout]

    with :ok <- validate_supported_opts(opts, supported, "Aerospike.create_role/4"),
         {:ok, whitelist} <- validate_role_whitelist(opts),
         {:ok, read_quota} <- validate_role_quota(opts, :read_quota),
         {:ok, write_quota} <- validate_role_quota(opts, :write_quota) do
      call_opts = Keyword.drop(opts, [:whitelist, :read_quota, :write_quota])
      {:ok, {whitelist, read_quota, write_quota, call_opts}}
    end
  end

  defp validate_supported_opts(opts, supported, callsite) when is_list(opts) do
    case Enum.find(opts, fn {key, _value} -> key not in supported end) do
      nil ->
        :ok

      {key, _value} ->
        {:error,
         Error.from_result_code(:invalid_argument,
           message:
             "#{callsite} supports only #{format_supported_opts(supported)} options, got #{inspect(key)}"
         )}
    end
  end

  defp validate_role_whitelist(opts) do
    Keyword.get(opts, :whitelist, [])
    |> validate_role_whitelist_value()
  end

  defp validate_role_whitelist_value(whitelist) when is_list(whitelist) do
    if Enum.all?(whitelist, &is_binary/1) do
      {:ok, whitelist}
    else
      invalid_role_whitelist(whitelist)
    end
  end

  defp validate_role_whitelist_value(other), do: invalid_role_whitelist(other)

  defp validate_role_quota(opts, key) when key in [:read_quota, :write_quota] do
    opts
    |> Keyword.get(key, 0)
    |> validate_role_quota_value(key)
  end

  defp validate_role_quota_value(quota, _key) when is_integer(quota) and quota >= 0 do
    {:ok, quota}
  end

  defp validate_role_quota_value(other, key) when key in [:read_quota, :write_quota] do
    {:error,
     Error.from_result_code(:invalid_argument,
       message: ":#{key} must be a non-negative integer, got: #{inspect(other)}"
     )}
  end

  defp format_supported_opts([opt]), do: inspect(opt)
  defp format_supported_opts([left, right]), do: "#{inspect(left)} and #{inspect(right)}"

  defp format_supported_opts(opts) when is_list(opts) do
    {init, [last]} = Enum.split(opts, -1)
    Enum.map_join(init, ", ", &inspect/1) <> ", and " <> inspect(last)
  end

  defp invalid_role_whitelist(value) do
    {:error,
     Error.from_result_code(:invalid_argument,
       message: ":whitelist must be a list of strings, got: #{inspect(value)}"
     )}
  end

  defp validate_enable_metrics_opts(opts) when is_list(opts) do
    case Enum.find(opts, fn {key, _value} -> key != :reset end) do
      nil ->
        validate_enable_metrics_reset(opts)

      {key, _value} ->
        {:error,
         Error.from_result_code(:invalid_argument,
           message: "Aerospike.enable_metrics/2 supports only :reset option, got #{inspect(key)}"
         )}
    end
  end

  defp validate_enable_metrics_reset(opts) do
    case Keyword.get(opts, :reset, false) do
      reset when is_boolean(reset) ->
        :ok

      other ->
        {:error,
         Error.from_result_code(:invalid_argument,
           message: ":reset must be a boolean, got: #{inspect(other)}"
         )}
    end
  end

  defp validate_warm_up_opts(opts) when is_list(opts) do
    with :ok <-
           validate_supported_opts(opts, [:count, :pool_checkout_timeout], "Aerospike.warm_up/2"),
         :ok <- validate_non_neg_opt(opts, :count) do
      validate_non_neg_opt(opts, :pool_checkout_timeout)
    end
  end

  defp validate_non_neg_opt(opts, key) do
    case Keyword.fetch(opts, key) do
      :error ->
        :ok

      {:ok, value} when is_integer(value) and value >= 0 ->
        :ok

      {:ok, value} ->
        {:error,
         Error.from_result_code(:invalid_argument,
           message: ":#{key} must be a non-negative integer, got: #{inspect(value)}"
         )}
    end
  end
end