defmodule Aerospike do @moduledoc """ Public API for the Aerospike Elixir client. Start a named connection (supervision tree) with `start_link/1`, then pass the same `name` atom as the first argument to CRUD functions. ## Example {:ok, _pid} = Aerospike.start_link( name: :aero, hosts: ["127.0.0.1:3000"], pool_size: 4 ) key = Aerospike.key("test", "users", "user:1") :ok = Aerospike.put!(:aero, key, %{"name" => "Ada"}) {:ok, record} = Aerospike.get(:aero, key) record.bins["name"] == "Ada" import Aerospike.Op {:ok, rec2} = Aerospike.operate(:aero, key, [put("status", "ok"), get("name")]) rec2.bins["name"] == "Ada" # Tuple key form (namespace, set, user_key) :ok = Aerospike.put!(:aero, {"test", "users", "user:2"}, %{"name" => "Grace"}) {:ok, rec3} = Aerospike.get(:aero, {"test", "users", "user:2"}) rec3.bins["name"] == "Grace" Aerospike.close(:aero) ## Key Input Forms Single-record APIs accept keys in either form: * `%Aerospike.Key{}` * `{namespace, set, user_key}` tuple Tuple keys are a convenience for user-key flows. For digest-only workflows, call `key_digest/3` explicitly. """ alias Aerospike.Admin alias Aerospike.Batch alias Aerospike.BatchOps alias Aerospike.CRUD alias Aerospike.Error alias Aerospike.IndexTask alias Aerospike.Key alias Aerospike.Page alias Aerospike.Policy alias Aerospike.Query alias Aerospike.RegisterTask alias Aerospike.Scan alias Aerospike.ScanOps alias Aerospike.Supervisor, as: AeroSupervisor alias Aerospike.Txn alias Aerospike.TxnRoll @typedoc """ Named connection handle: currently an atom (`:name` from `start_link/1`). Internal routing and ETS table names assume atom registration today. Broader `GenServer.server()` support may be added in a future phase. """ @type conn :: atom() @typedoc """ One cluster node as returned by `nodes/1`: display `name`, TCP `host`, and `port`. """ @type node_info :: %{name: String.t(), host: String.t(), port: non_neg_integer()} @doc """ Returns a child specification for supervision trees. Options are validated with `NimbleOptions`; invalid options raise `ArgumentError`. ## Options See `start_link/1`. ## Example children = [ {Aerospike, name: :aero, hosts: ["127.0.0.1:3000"]} ] """ @spec child_spec(keyword()) :: Supervisor.child_spec() def child_spec(opts) when is_list(opts), do: AeroSupervisor.child_spec(opts) @doc """ Starts the client supervision tree under a registered supervisor name. Returns `{:ok, pid}` on success. If no seed is reachable at startup, the cluster retries on each tend cycle until a seed becomes available. Poll the `:cluster_ready` ETS flag to detect when the cluster is operational. ## Options * `:name` — required atom used as `conn` for all operations. * `:hosts` — required non-empty list of host strings (`"host:port"` or `"host"` for port 3000). * `:pool_size` — connections per node (default `10`). * `:pool_checkout_timeout` — pool checkout timeout in ms (default `5000`). * `:connect_timeout` — TCP connect timeout in ms (default `5000`). * `:tend_interval` — periodic cluster tend interval in ms (default `1000`). * `:recv_timeout` — receive timeout for protocol reads in ms (default `5000`). * `:max_error_rate` — per-node circuit-breaker threshold within the configured tend window (default `100`). Set to `0` to disable breaker logic. * `:error_rate_window` — circuit-breaker window size in tend ticks (default `1`). * `:auth_opts` — optional authentication keyword list. * `:tls` — when `true`, upgrades each node connection with TLS after TCP connect (default `false`). * `:tls_opts` — keyword list passed to `:ssl.connect/3` (certificates, verify, SNI, etc.; default `[]`). For non-IP hosts, `:server_name_indication` defaults to the hostname unless set in `:tls_opts`. * `:defaults` — policy defaults per command (`:write`, `:read`, `:delete`, `:exists`, `:touch`, `:operate`, `:batch`, `:scan`, `:query`). ## TLS example Aerospike.start_link( name: :aero_tls, hosts: ["aerospike.example.com:4333"], tls: true, tls_opts: [ verify: :verify_peer, cacertfile: "/etc/ssl/certs/ca-certificates.crt" ] ) ## Circuit breaker example Aerospike.start_link( name: :aero_cb, hosts: ["127.0.0.1:3000"], max_error_rate: 50, error_rate_window: 5 ) The breaker uses fixed-threshold semantics per node: requests are rejected with `:max_error_rate` once a node reaches `max_error_rate` errors inside the window, and counters reset every `error_rate_window` tend ticks. Telemetry events: * `[:aerospike, :circuit_breaker, :increment]` * `[:aerospike, :circuit_breaker, :reject]` * `[:aerospike, :circuit_breaker, :reset]` """ @spec start_link(keyword()) :: Supervisor.on_start() def start_link(opts) when is_list(opts), do: AeroSupervisor.start_link(opts) @doc """ Stops the supervision tree for `conn` (the `:name` given at startup). `timeout` is passed to `Supervisor.stop/3` (default `15_000` ms). """ @spec close(conn, timeout :: non_neg_integer()) :: :ok def close(conn, timeout \\ 15_000) when is_atom(conn) and is_integer(timeout) and timeout >= 0 do _ = Supervisor.stop(AeroSupervisor.sup_name(conn), :normal, timeout) :ok end @doc """ Builds a key from namespace, set, and user key (string or int64 integer). Delegates to `Aerospike.Key.new/3`. """ @spec key(String.t(), String.t(), String.t() | integer()) :: Key.t() defdelegate key(namespace, set, user_key), to: Key, as: :new @doc """ Builds a key from namespace, set, and an existing 20-byte digest. Delegates to `Aerospike.Key.from_digest/3`. """ @spec key_digest(String.t(), String.t(), <<_::160>>) :: Key.t() defdelegate key_digest(namespace, set, digest), to: Key, as: :from_digest @doc """ Writes bins for the given key. Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}` tuple keys. Per-call `opts` are merged over connection `defaults` (`Keyword.merge/2`). ## Options Write policy options: `:ttl`, `:timeout`, `:generation`, `:gen_policy`, `:exists`, `:send_key`, `:durable_delete`, `:pool_checkout_timeout`, `:replica`. ## Example key = Aerospike.key("test", "users", "user:1") :ok = Aerospike.put(:aero, key, %{"name" => "Ada"}) :ok = Aerospike.put(:aero, {"test", "users", "user:2"}, %{"name" => "Grace"}) """ @spec put(conn, Key.key_input(), Aerospike.Record.bins_input(), keyword()) :: :ok | {:error, Error.t()} def put(conn, key, bins, opts \\ []) when is_atom(conn) and is_map(bins) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_write(opts) do CRUD.put(conn, key, bins, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `put/4` but returns `:ok` or raises `Aerospike.Error`. """ @spec put!(conn, Key.key_input(), Aerospike.Record.bins_input(), keyword()) :: :ok def put!(conn, key, bins, opts \\ []) when is_atom(conn) and is_map(bins) and is_list(opts) do case put(conn, key, bins, opts) do :ok -> :ok {:error, %Error{} = e} -> raise e end end @doc """ Reads a record for the key. Accepts tuple keys; see `put/4`. ## Options Read policy: `:timeout`, `:bins`, `:header_only`, `:pool_checkout_timeout`, `:replica`. """ @spec get(conn, Key.key_input(), keyword()) :: {:ok, Aerospike.Record.t()} | {:error, Error.t()} def get(conn, key, opts \\ []) when is_atom(conn) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_read(opts) do CRUD.get(conn, key, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `get/3` but returns the record or raises `Aerospike.Error`. """ @spec get!(conn, Key.key_input(), keyword()) :: Aerospike.Record.t() def get!(conn, key, opts \\ []) when is_atom(conn) and is_list(opts) do case get(conn, key, opts) do {:ok, record} -> record {:error, %Error{} = e} -> raise e end end @doc """ Deletes the record. Returns `{:ok, true}` if a record was removed, `{:ok, false}` if absent. Accepts tuple keys; see `put/4`. ## Options `:timeout`, `:durable_delete`, `:pool_checkout_timeout`, `:replica`. """ @spec delete(conn, Key.key_input(), keyword()) :: {:ok, boolean()} | {:error, Error.t()} def delete(conn, key, opts \\ []) when is_atom(conn) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_delete(opts) do CRUD.delete(conn, key, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `delete/3` but returns the boolean or raises `Aerospike.Error`. """ @spec delete!(conn, Key.key_input(), keyword()) :: boolean() def delete!(conn, key, opts \\ []) when is_atom(conn) and is_list(opts) do case delete(conn, key, opts) do {:ok, deleted?} -> deleted? {:error, %Error{} = e} -> raise e end end @doc """ Returns whether a record exists for the key. Accepts tuple keys; see `put/4`. ## Options `:timeout`, `:pool_checkout_timeout`, `:replica`. """ @spec exists(conn, Key.key_input(), keyword()) :: {:ok, boolean()} | {:error, Error.t()} def exists(conn, key, opts \\ []) when is_atom(conn) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_exists(opts) do CRUD.exists(conn, key, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `exists/3` but returns the boolean or raises `Aerospike.Error`. """ @spec exists!(conn, Key.key_input(), keyword()) :: boolean() def exists!(conn, key, opts \\ []) when is_atom(conn) and is_list(opts) do case exists(conn, key, opts) do {:ok, exists?} -> exists? {:error, %Error{} = e} -> raise e end end @doc """ Refreshes TTL without changing bins. Accepts tuple keys; see `put/4`. ## Options `:ttl`, `:timeout`, `:pool_checkout_timeout`, `:replica`. """ @spec touch(conn, Key.key_input(), keyword()) :: :ok | {:error, Error.t()} def touch(conn, key, opts \\ []) when is_atom(conn) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_touch(opts) do CRUD.touch(conn, key, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `touch/3` but returns `:ok` or raises `Aerospike.Error`. """ @spec touch!(conn, Key.key_input(), keyword()) :: :ok def touch!(conn, key, opts \\ []) when is_atom(conn) and is_list(opts) do case touch(conn, key, opts) do :ok -> :ok {:error, %Error{} = e} -> raise e end end @doc """ Runs multiple read/write operations on one record in a single atomic round-trip. Accepts tuple keys; see `put/4`. Pass a list of operations from `Aerospike.Op`, `Aerospike.Op.List`, `Aerospike.Op.Map`, etc. ## Options Merges **read** and **write** policy keys: `:timeout`, `:ttl`, `:generation`, `:gen_policy`, `:exists`, `:send_key`, `:durable_delete`, `:respond_per_each_op`, `:pool_checkout_timeout`, `:replica`. """ @spec operate(conn, Key.key_input(), [Aerospike.Op.t()], keyword()) :: {:ok, Aerospike.Record.t()} | {:error, Error.t()} def operate(conn, key, ops, opts \\ []) when is_atom(conn) and is_list(ops) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_operate(opts) do CRUD.operate(conn, key, ops, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `operate/4` but returns `%Aerospike.Record{}` or raises `Aerospike.Error`. """ @spec operate!(conn, Key.key_input(), [Aerospike.Op.t()], keyword()) :: Aerospike.Record.t() def operate!(conn, key, ops, opts \\ []) when is_atom(conn) and is_list(ops) and is_list(opts) do case operate(conn, key, ops, opts) do {:ok, record} -> record {:error, %Error{} = e} -> raise e end end @doc """ Atomically adds integer deltas to bins. Accepts tuple keys; see `put/4`. If the record does not exist, Aerospike implicitly creates it — bins start at the added value. This makes `add` the idiomatic way to implement counters. ## Options Write policy options: `:ttl`, `:timeout`, `:generation`, `:gen_policy`, `:exists`, `:send_key`, `:durable_delete`, `:pool_checkout_timeout`, `:replica`. ## Example :ok = Aerospike.add(:aero, key, %{"login_count" => 1, "bytes_used" => 256}) """ @spec add(conn, Key.key_input(), Aerospike.Record.bins_input(), keyword()) :: :ok | {:error, Error.t()} def add(conn, key, bins, opts \\ []) when is_atom(conn) and is_map(bins) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_write(opts) do CRUD.add(conn, key, bins, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `add/4` but returns `:ok` or raises `Aerospike.Error`. """ @spec add!(conn, Key.key_input(), Aerospike.Record.bins_input(), keyword()) :: :ok def add!(conn, key, bins, opts \\ []) when is_atom(conn) and is_map(bins) and is_list(opts) do case add(conn, key, bins, opts) do :ok -> :ok {:error, %Error{} = e} -> raise e end end @doc """ Atomically appends string suffixes to bins. Accepts tuple keys; see `put/4`. If the record does not exist, Aerospike implicitly creates it — the bin value becomes the appended string (not appended to an empty string). ## Options Write policy options: `:ttl`, `:timeout`, `:generation`, `:gen_policy`, `:exists`, `:send_key`, `:durable_delete`, `:pool_checkout_timeout`, `:replica`. ## Example :ok = Aerospike.append(:aero, key, %{"greeting" => " world"}) """ @spec append(conn, Key.key_input(), Aerospike.Record.bins_input(), keyword()) :: :ok | {:error, Error.t()} def append(conn, key, bins, opts \\ []) when is_atom(conn) and is_map(bins) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_write(opts) do CRUD.append(conn, key, bins, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `append/4` but returns `:ok` or raises `Aerospike.Error`. """ @spec append!(conn, Key.key_input(), Aerospike.Record.bins_input(), keyword()) :: :ok def append!(conn, key, bins, opts \\ []) when is_atom(conn) and is_map(bins) and is_list(opts) do case append(conn, key, bins, opts) do :ok -> :ok {:error, %Error{} = e} -> raise e end end @doc """ Atomically prepends string prefixes to bins. Accepts tuple keys; see `put/4`. If the record does not exist, Aerospike implicitly creates it — the bin value becomes the prepended string (not prepended to an empty string). ## Options Write policy options: `:ttl`, `:timeout`, `:generation`, `:gen_policy`, `:exists`, `:send_key`, `:durable_delete`, `:pool_checkout_timeout`, `:replica`. ## Example :ok = Aerospike.prepend(:aero, key, %{"greeting" => "hello "}) """ @spec prepend(conn, Key.key_input(), Aerospike.Record.bins_input(), keyword()) :: :ok | {:error, Error.t()} def prepend(conn, key, bins, opts \\ []) when is_atom(conn) and is_map(bins) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, call_opts} <- Policy.validate_write(opts) do CRUD.prepend(conn, key, bins, call_opts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `prepend/4` but returns `:ok` or raises `Aerospike.Error`. """ @spec prepend!(conn, Key.key_input(), Aerospike.Record.bins_input(), keyword()) :: :ok def prepend!(conn, key, bins, opts \\ []) when is_atom(conn) and is_map(bins) and is_list(opts) do case prepend(conn, key, bins, opts) do :ok -> :ok {:error, %Error{} = e} -> raise e end end @batch_read_opts [:bins, :header_only, :read_touch_ttl_percent] @doc """ Reads multiple keys in one round-trip per server node. Returns `{:ok, records}` in **key order**; missing keys appear as `nil`. ## Options * Batch: `:timeout`, `:pool_checkout_timeout`, `:replica` (`:master`, `:sequence`, `:any`, or a non-negative replica index), `:respond_all_keys`, `:filter` (`Aerospike.Exp.from_wire/1`) * Read: `:bins`, `:header_only`, `:read_touch_ttl_percent` (omit `:bins` to read all bins; `:bins` must not be an empty list) ## Examples keys = Enum.map(1..3, &Aerospike.key("test", "users", "user:\#{&1}")) {:ok, records} = Aerospike.batch_get(:aero, keys) # records: [%Aerospike.Record{} | nil, ...] — aligned with keys # Project specific bins {:ok, records} = Aerospike.batch_get(:aero, keys, bins: ["name", "age"]) """ @spec batch_get(conn, [Key.key_input()], keyword()) :: {:ok, [Aerospike.Record.t() | nil]} | {:error, Error.t()} def batch_get(conn, keys, opts \\ []) when is_atom(conn) and is_list(keys) and is_list(opts) do # `Keyword.split/2` returns `{taken_for_these_keys, rest}`. {read_kw, batch_kw} = Keyword.split(opts, @batch_read_opts) with {:ok, keys} <- coerce_keys(keys), {:ok, bopts} <- Policy.validate_batch(batch_kw), {:ok, ropts} <- Policy.validate_read(read_kw) do BatchOps.batch_get(conn, keys, Keyword.merge(bopts, ropts)) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `batch_get/3` but returns the list or raises `Aerospike.Error`. ## Example records = Aerospike.batch_get!(:aero, keys, bins: ["name"]) """ @spec batch_get!(conn, [Key.key_input()], keyword()) :: [Aerospike.Record.t() | nil] def batch_get!(conn, keys, opts \\ []) when is_atom(conn) and is_list(keys) and is_list(opts) do case batch_get(conn, keys, opts) do {:ok, recs} -> recs {:error, %Error{} = e} -> raise e end end @doc """ Checks existence for multiple keys in one round-trip per node. Returns `{:ok, booleans}` aligned with `keys`. ## Options `:timeout`, `:pool_checkout_timeout`, `:replica` (`:master`, `:sequence`, `:any`, or index), `:respond_all_keys`, `:filter` (`Aerospike.Exp.from_wire/1`) ## Example keys = [key1, key2, key3] {:ok, [true, false, true]} = Aerospike.batch_exists(:aero, keys) """ @spec batch_exists(conn, [Key.key_input()], keyword()) :: {:ok, [boolean()]} | {:error, Error.t()} def batch_exists(conn, keys, opts \\ []) when is_atom(conn) and is_list(keys) and is_list(opts) do with {:ok, keys} <- coerce_keys(keys), {:ok, bopts} <- Policy.validate_batch(opts) do BatchOps.batch_exists(conn, keys, bopts) else {:error, %Error{} = e} -> {:error, e} {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `batch_exists/3` but returns the list or raises `Aerospike.Error`. ## Example [true, false] = Aerospike.batch_exists!(:aero, [key1, key2]) """ @spec batch_exists!(conn, [Key.key_input()], keyword()) :: [boolean()] def batch_exists!(conn, keys, opts \\ []) when is_atom(conn) and is_list(keys) and is_list(opts) do case batch_exists(conn, keys, opts) do {:ok, xs} -> xs {:error, %Error{} = e} -> raise e end end @doc """ Runs a heterogeneous batch built with `Aerospike.Batch`. Each input operation produces one `Aerospike.BatchResult` in the same position. ## Options `:timeout`, `:pool_checkout_timeout`, `:replica` (`:master`, `:sequence`, `:any`, or index), `:respond_all_keys`, `:filter` (`Aerospike.Exp.from_wire/1`) ## Example alias Aerospike.Batch {:ok, results} = Aerospike.batch_operate(:aero, [ Batch.read(key1, bins: ["name"]), Batch.put(key2, %{"score" => 100}), Batch.delete(key3), Batch.operate(key4, [Aerospike.Op.add("hits", 1)]) ]) Enum.each(results, fn %Aerospike.BatchResult{status: :ok, record: rec} -> IO.inspect(rec, label: "success") %Aerospike.BatchResult{status: :error, error: err} -> IO.inspect(err, label: "failed") end) """ @spec batch_operate(conn, [Batch.t()], keyword()) :: {:ok, [Aerospike.BatchResult.t()]} | {:error, Error.t()} def batch_operate(conn, ops, opts \\ []) when is_atom(conn) and is_list(ops) and is_list(opts) do case Policy.validate_batch(opts) do {:ok, bopts} -> BatchOps.batch_operate(conn, ops, bopts) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `batch_operate/3` but returns results or raises `Aerospike.Error`. ## Example results = Aerospike.batch_operate!(:aero, [Batch.read(key1)]) """ @spec batch_operate!(conn, [Batch.t()], keyword()) :: [Aerospike.BatchResult.t()] def batch_operate!(conn, ops, opts \\ []) when is_atom(conn) and is_list(ops) and is_list(opts) do case batch_operate(conn, ops, opts) do {:ok, rs} -> rs {:error, %Error{} = e} -> raise e end end @doc """ Returns a lazy `Stream` of records from a scan or query. The stream yields bare `%Aerospike.Record{}` structs. If a network error or server error occurs mid-stream, `Aerospike.Error` is raised. The stream holds one pool connection per node for its entire lifetime. On early termination (e.g. `Enum.take/2`), the connection is closed rather than returned to the pool. In multi-node clusters, both `stream!/3` and `all/3` fan out across nodes concurrently by default. ## Options Scan/query policy: `:timeout`, `:pool_checkout_timeout`, `:replica`, `:max_concurrent_nodes` (0 = all nodes). ## Examples alias Aerospike.Scan Aerospike.stream!(:aero, Scan.new("test", "users")) |> Stream.filter(fn r -> r.bins["age"] > 21 end) |> Enum.take(100) """ @spec stream!(conn(), Scan.t() | Query.t(), keyword()) :: Enumerable.t() def stream!(conn, scannable, opts \\ []) when is_atom(conn) and is_list(opts) do case validate_scan_query_opts(scannable, opts) do {:ok, call_opts} -> ScanOps.stream(conn, scannable, call_opts) {:error, %NimbleOptions.ValidationError{} = e} -> raise Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e) ) end end @doc """ Eagerly collects all records from a scan or query into a list. Requires `max_records` on the scan/query builder. Returns `{:error, %Aerospike.Error{code: :max_records_required}}` if missing. ## Options Scan/query policy: `:timeout`, `:pool_checkout_timeout`, `:replica`. ## Examples alias Aerospike.Scan {:ok, records} = Aerospike.all(:aero, Scan.new("test", "users") |> Scan.max_records(1_000)) """ @spec all(conn(), Scan.t() | Query.t(), keyword()) :: {:ok, [Aerospike.Record.t()]} | {:error, Error.t()} def all(conn, scannable, opts \\ []) when is_atom(conn) and is_list(opts) do case validate_scan_query_opts(scannable, opts) do {:ok, call_opts} -> ScanOps.all(conn, scannable, call_opts) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `all/3` but returns the list or raises `Aerospike.Error`. """ @spec all!(conn(), Scan.t() | Query.t(), keyword()) :: [Aerospike.Record.t()] def all!(conn, scannable, opts \\ []) when is_atom(conn) and is_list(opts) do case all(conn, scannable, opts) do {:ok, records} -> records {:error, %Error{} = e} -> raise e end end @doc """ Counts records matching a scan or query using a server-side NOBINDATA scan. **Cost:** This performs a full scan or query with bin payloads omitted. The server sends one record header per matching record, and the client counts them. For large datasets, prefer the Aerospike info protocol (`sets//`) via a raw info command for unfiltered per-set counts. ## Options Scan/query policy: `:timeout`, `:pool_checkout_timeout`, `:replica`. ## Examples alias Aerospike.Scan {:ok, n} = Aerospike.count(:aero, Scan.new("test", "users")) """ @spec count(conn(), Scan.t() | Query.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, Error.t()} def count(conn, scannable, opts \\ []) when is_atom(conn) and is_list(opts) do case validate_scan_query_opts(scannable, opts) do {:ok, call_opts} -> ScanOps.count(conn, scannable, call_opts) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `count/3` but returns the count or raises `Aerospike.Error`. """ @spec count!(conn(), Scan.t() | Query.t(), keyword()) :: non_neg_integer() def count!(conn, scannable, opts \\ []) when is_atom(conn) and is_list(opts) do case count(conn, scannable, opts) do {:ok, n} -> n {:error, %Error{} = e} -> raise e end end @doc """ Returns one page of records from a scan or query with cursor-based pagination. Pass `cursor: page.cursor` from a previous `page/3` call to resume. ## Options * Scan/query policy: `:timeout`, `:pool_checkout_timeout`, `:replica` * `:cursor` — `%Aerospike.Cursor{}` or encoded cursor binary from `Cursor.encode/1` ## Examples alias Aerospike.Scan scan = Scan.new("test", "users") |> Scan.max_records(100) {:ok, page} = Aerospike.page(:aero, scan) {:ok, page2} = Aerospike.page(:aero, scan, cursor: page.cursor) """ @spec page(conn(), Scan.t() | Query.t(), keyword()) :: {:ok, Page.t()} | {:error, Error.t()} def page(conn, scannable, opts \\ []) when is_atom(conn) and is_list(opts) do {cursor, opts2} = Keyword.pop(opts, :cursor) case validate_scan_query_opts(scannable, opts2) do {:ok, call_opts} -> opts3 = if cursor != nil, do: Keyword.put(call_opts, :cursor, cursor), else: call_opts ScanOps.page(conn, scannable, opts3) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Same as `page/3` but returns `%Aerospike.Page{}` or raises `Aerospike.Error`. """ @spec page!(conn(), Scan.t() | Query.t(), keyword()) :: Page.t() def page!(conn, scannable, opts \\ []) when is_atom(conn) and is_list(opts) do case page(conn, scannable, opts) do {:ok, p} -> p {:error, %Error{} = e} -> raise e end end @doc """ Sends a raw info command to a random cluster node and returns the response string. ## Options `:timeout`, `:pool_checkout_timeout` ## Example {:ok, "test"} = Aerospike.info(:aero, "namespaces") """ @spec info(conn(), String.t(), keyword()) :: {:ok, String.t()} | {:error, Error.t()} def info(conn, command, opts \\ []) when is_atom(conn) and is_binary(command) and is_list(opts) do case Policy.validate_info(opts) do {:ok, call_opts} -> Admin.info(conn, command, call_opts) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Sends a raw info command to the named cluster node and returns the response string. ## Options `:timeout`, `:pool_checkout_timeout` ## Example {:ok, [%{name: name}]} = Aerospike.nodes(:aero) {:ok, response} = Aerospike.info_node(:aero, name, "statistics") """ @spec info_node(conn(), String.t(), String.t(), keyword()) :: {:ok, String.t()} | {:error, Error.t()} def info_node(conn, node_name, command, opts \\ []) when is_atom(conn) and is_binary(node_name) and is_binary(command) and is_list(opts) do case Policy.validate_info(opts) do {:ok, call_opts} -> Admin.info_node(conn, node_name, command, call_opts) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Returns the list of cluster nodes with their name, host, and port. Each element matches `t:Aerospike.node_info/0`. ## Example {:ok, [%{name: "BB9...", host: "127.0.0.1", port: 3000}]} = Aerospike.nodes(:aero) """ @spec nodes(conn()) :: {:ok, [node_info()]} | {:error, Error.t()} def nodes(conn) when is_atom(conn), do: Admin.nodes(conn) @doc """ Returns the list of cluster node name strings. ## Example {:ok, ["BB9..."]} = Aerospike.node_names(:aero) """ @spec node_names(conn()) :: {:ok, [String.t()]} | {:error, Error.t()} def node_names(conn) when is_atom(conn), do: Admin.node_names(conn) @doc """ Truncates all records in `namespace`, optionally only those written before `before:`. ## Options * `:before` — `%DateTime{}` — truncate only records written before this timestamp. * `:pool_checkout_timeout` — pool checkout timeout in ms. ## Example :ok = Aerospike.truncate(:aero, "test") :ok = Aerospike.truncate(:aero, "test", before: DateTime.utc_now()) """ @spec truncate(conn(), String.t(), keyword()) :: :ok | {:error, Error.t()} def truncate(conn, namespace, opts \\ []) def truncate(conn, namespace, set) when is_atom(conn) and is_binary(namespace) and is_binary(set) do truncate(conn, namespace, set, []) end def truncate(conn, namespace, opts) when is_atom(conn) and is_binary(namespace) and is_list(opts) do case Policy.validate_info(Keyword.delete(opts, :before)) do {:ok, call_opts} -> Admin.truncate(conn, namespace, Keyword.merge(call_opts, Keyword.take(opts, [:before]))) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Truncates all records in `namespace` and `set`, optionally those written before `before:`. ## Options * `:before` — `%DateTime{}` — truncate only records written before this timestamp. * `:pool_checkout_timeout` — pool checkout timeout in ms. ## Example :ok = Aerospike.truncate(:aero, "test", "users") :ok = Aerospike.truncate(:aero, "test", "users", before: DateTime.utc_now()) """ @spec truncate(conn(), String.t(), String.t(), keyword()) :: :ok | {:error, Error.t()} def truncate(conn, namespace, set, opts) when is_atom(conn) and is_binary(namespace) and is_binary(set) and is_list(opts) do case Policy.validate_info(Keyword.delete(opts, :before)) do {:ok, call_opts} -> Admin.truncate( conn, namespace, set, Keyword.merge(call_opts, Keyword.take(opts, [:before])) ) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Creates a secondary index on `set` in `namespace`. The server builds the index in the background and returns an `IndexTask` that you can poll with `IndexTask.status/1` or block on with `IndexTask.wait/2`. ## Options * `:bin` — required string; the bin name to index. * `:name` — required string; the index name. * `:type` — required atom; `:numeric`, `:string`, or `:geo2dsphere`. * `:collection` — optional atom; `:list`, `:mapkeys`, or `:mapvalues` for CDT bins. * `:pool_checkout_timeout` — pool checkout timeout in ms. ## Example {:ok, task} = Aerospike.create_index(:aero, "test", "demo", bin: "age", name: "age_idx", type: :numeric ) :ok = Aerospike.IndexTask.wait(task, timeout: 30_000) """ @spec create_index(conn(), String.t(), String.t(), keyword()) :: {:ok, IndexTask.t()} | {:error, Error.t()} def create_index(conn, namespace, set, opts) when is_atom(conn) and is_binary(namespace) and is_binary(set) and is_list(opts) do case Policy.validate_index_create(opts) do {:ok, call_opts} -> Admin.create_index(conn, namespace, set, call_opts) {:error, %NimbleOptions.ValidationError{} = e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Drops a secondary index by name. Returns `:ok` if the index was dropped or did not exist. ## Example :ok = Aerospike.drop_index(:aero, "test", "age_idx") """ @spec drop_index(conn(), String.t(), String.t()) :: :ok | {:error, Error.t()} def drop_index(conn, namespace, index_name) when is_atom(conn) and is_binary(namespace) and is_binary(index_name) do Admin.drop_index(conn, namespace, index_name, []) end @doc """ Registers a UDF package on the cluster. `path_or_content` can be either a filesystem path to a `.lua` file or the raw Lua source as a string. `server_name` is the package name used on the server (typically the filename, e.g. `"my_module.lua"`). Returns an `RegisterTask` that you can poll with `RegisterTask.status/1` or block on with `RegisterTask.wait/2`. ## Example {:ok, task} = Aerospike.register_udf(:aero, "/path/to/my_module.lua", "my_module.lua") :ok = Aerospike.RegisterTask.wait(task, timeout: 10_000) """ @spec register_udf(conn(), String.t(), String.t()) :: {:ok, RegisterTask.t()} | {:error, Error.t()} def register_udf(conn, path_or_content, server_name) when is_atom(conn) and is_binary(path_or_content) and is_binary(server_name) do Admin.register_udf(conn, path_or_content, server_name, []) end @doc """ Registers a UDF package on the cluster with options. See `register_udf/3` for details. ## Options * `:pool_checkout_timeout` — pool checkout timeout in ms. """ @spec register_udf(conn(), String.t(), String.t(), keyword()) :: {:ok, RegisterTask.t()} | {:error, Error.t()} def register_udf(conn, path_or_content, server_name, opts) when is_atom(conn) and is_binary(path_or_content) and is_binary(server_name) and is_list(opts) do Admin.register_udf(conn, path_or_content, server_name, opts) end @doc """ Removes a UDF package from the cluster. Returns `:ok` whether or not the package was registered. ## Example :ok = Aerospike.remove_udf(:aero, "my_module.lua") """ @spec remove_udf(conn(), String.t()) :: :ok | {:error, Error.t()} def remove_udf(conn, udf_name) when is_atom(conn) and is_binary(udf_name) do Admin.remove_udf(conn, udf_name, []) end @doc """ Removes a UDF package from the cluster with options. See `remove_udf/2` for details. ## Options * `:pool_checkout_timeout` — pool checkout timeout in ms. """ @spec remove_udf(conn(), String.t(), keyword()) :: :ok | {:error, Error.t()} def remove_udf(conn, udf_name, opts) when is_atom(conn) and is_binary(udf_name) and is_list(opts) do Admin.remove_udf(conn, udf_name, opts) end @doc """ Executes a UDF (User Defined Function) on a single record. Accepts tuple keys; see `put/4`. `package` is the Lua module name as registered on the server (without the `.lua` extension). `function` is the Lua function name. `args` is the list of arguments passed to the function. Returns `{:ok, value}` where `value` is the Lua function's return value, or `{:error, %Error{code: :udf_bad_response}}` on UDF runtime error. ## Example {:ok, task} = Aerospike.register_udf(:aero, "/path/my.lua", "my.lua") :ok = Aerospike.RegisterTask.wait(task) key = Aerospike.key("test", "demo", "k1") :ok = Aerospike.put(:aero, key, %{"n" => 42}) {:ok, result} = Aerospike.apply_udf(:aero, key, "my", "double_n", []) """ @spec apply_udf(conn(), Key.key_input(), String.t(), String.t(), list()) :: {:ok, term()} | {:error, Error.t()} def apply_udf(conn, key, package, function, args) when is_atom(conn) and is_binary(package) and is_binary(function) and is_list(args) do with {:ok, key} <- coerce_key(key) do CRUD.apply_udf(conn, key, package, function, args, []) end end @doc """ Executes a UDF on a single record with options. See `apply_udf/5` for details. ## Options * `:timeout` — socket timeout in milliseconds. * `:filter` — server-side expression filter (`%Aerospike.Exp{}`). * `:pool_checkout_timeout` — pool checkout timeout in milliseconds. * `:replica` — replica routing: `:master`, `:sequence`, or `:any`. """ @spec apply_udf(conn(), Key.key_input(), String.t(), String.t(), list(), keyword()) :: {:ok, term()} | {:error, Error.t()} def apply_udf(conn, key, package, function, args, opts) when is_atom(conn) and is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do with {:ok, key} <- coerce_key(key), {:ok, validated} <- Policy.validate_udf(opts) do CRUD.apply_udf(conn, key, package, function, args, validated) else {:error, %Error{} = e} -> {:error, e} {:error, e} -> {:error, Error.from_result_code(:parameter_error, message: Policy.validation_error_message(e))} end end @doc """ Commits a transaction. Runs the multi-phase commit protocol: verifies all tracked reads, marks the server-side monitor record for roll-forward, and finalizes each write. Returns `{:ok, :committed}` on success, `{:ok, :already_committed}` if the transaction was already committed, or `{:error, %Aerospike.Error{}}` on failure. Most callers should use `transaction/2` or `transaction/3`, which handle initialization, commit, and abort automatically. Use `commit/2` directly only when managing the transaction lifecycle manually (without the `transaction` wrapper). ## Example txn = Txn.new() TxnOps.init_tracking(:conn, txn) Aerospike.put(:conn, key1, %{"x" => 1}, txn: txn) Aerospike.put(:conn, key2, %{"x" => 2}, txn: txn) {:ok, :committed} = Aerospike.commit(:conn, txn) """ @spec commit(conn(), Txn.t()) :: {:ok, :committed | :already_committed} | {:error, Error.t()} def commit(conn, %Txn{} = txn) when is_atom(conn) do TxnRoll.commit(conn, txn, []) end @doc """ Aborts a transaction, rolling back all writes. Returns `{:ok, :aborted}` on success, `{:ok, :already_aborted}` if the transaction was already aborted, or `{:error, %Aerospike.Error{}}` if the transaction was already committed. Roll-back writes are best-effort: if some fail (e.g., during a network partition), the server releases remaining locks when the MRT timeout expires. Most callers should use `transaction/2` or `transaction/3`, which abort automatically on failure. Use `abort/2` directly only when managing the transaction lifecycle manually (without the `transaction` wrapper). ## Example txn = Txn.new() TxnOps.init_tracking(:conn, txn) Aerospike.put(:conn, key, %{"x" => 1}, txn: txn) {:ok, :aborted} = Aerospike.abort(:conn, txn) """ @spec abort(conn(), Txn.t()) :: {:ok, :aborted | :already_aborted} | {:error, Error.t()} def abort(conn, %Txn{} = txn) when is_atom(conn) do TxnRoll.abort(conn, txn, []) end @doc """ Returns the current state of a transaction. Primarily useful **inside** a `transaction/2` or `transaction/3` callback to check whether the transaction is still open. After a successful commit or abort, the transaction's ETS tracking is cleaned up, so this function returns `{:error, %Aerospike.Error{}}` rather than `{:ok, :committed}` or `{:ok, :aborted}`. ## States - `:open` — transaction is active and accepting operations - `:verified` — verify phase completed (intermediate state during commit) ## Example Aerospike.transaction(:conn, fn txn -> {:ok, :open} = Aerospike.txn_status(:conn, txn) Aerospike.put(:conn, key, %{"x" => 1}, txn: txn) end) """ @spec txn_status(conn(), Txn.t()) :: {:ok, :open | :verified | :committed | :aborted} | {:error, Error.t()} def txn_status(conn, %Txn{} = txn) when is_atom(conn) do TxnRoll.txn_status(conn, txn) end @doc """ Runs a function within a new transaction and commits or aborts automatically. Creates a new transaction, calls `fun.(txn)` with the transaction handle, then: - Commits on success (non-exception return). - Aborts and returns `{:error, e}` if `fun` raises `%Aerospike.Error{}`. - Aborts and re-raises if `fun` raises any other exception, throws, or exits. Abort runs on **all** failure paths (not just `%Aerospike.Error{}`), so server-side write locks are released immediately instead of waiting for the MRT timeout to expire. Returns `{:ok, fun_result}` on successful commit. Do not call `commit/2` or `abort/2` directly inside `fun` — the wrapper manages both automatically. If you call `abort/2` inside the callback and then return normally, the auto-commit will fail because the transaction's tracking state has already been cleaned up. Use `commit/2` and `abort/2` only when managing the transaction lifecycle manually (without this wrapper). ## Example {:ok, _} = Aerospike.transaction(:conn, fn txn -> Aerospike.put(:conn, key1, %{"x" => 1}, txn: txn) Aerospike.put(:conn, key2, %{"x" => 2}, txn: txn) end) """ @spec transaction(conn(), (Txn.t() -> term())) :: {:ok, term()} | {:error, Error.t()} def transaction(conn, fun) when is_atom(conn) and is_function(fun, 1) do TxnRoll.transaction(conn, [], fun) end @doc """ Runs a function within a transaction using a provided handle or options, committing or aborting automatically. When `txn_or_opts` is a `%Aerospike.Txn{}`, the existing transaction handle is used and tracking is initialized fresh. When it is a keyword list, a new transaction is created with those options (e.g., `timeout: 5_000`). See `transaction/2` for commit/abort behavior. ## Example {:ok, _} = Aerospike.transaction(:conn, [timeout: 5_000], fn txn -> Aerospike.put(:conn, key, %{"x" => 1}, txn: txn) end) """ @spec transaction(conn(), Txn.t() | keyword(), (Txn.t() -> term())) :: {:ok, term()} | {:error, Error.t()} def transaction(conn, txn_or_opts, fun) when is_atom(conn) and is_function(fun, 1) do TxnRoll.transaction(conn, txn_or_opts, fun) end defp coerce_key(key) do {:ok, Key.coerce!(key)} rescue e in ArgumentError -> {:error, Error.from_result_code(:parameter_error, message: e.message)} end defp coerce_keys(keys) do {:ok, Enum.map(keys, &Key.coerce!/1)} rescue e in ArgumentError -> {:error, Error.from_result_code(:parameter_error, message: e.message)} end defp validate_scan_query_opts(%Scan{} = _scannable, opts) when is_list(opts), do: Policy.validate_scan(opts) defp validate_scan_query_opts(%Query{} = _scannable, opts) when is_list(opts), do: Policy.validate_query(opts) end