@spec append(FerricStore.Instance.t(), binary(), binary()) :: {:ok, non_neg_integer()}

Atomically appends suffix to the value of key.

If the key does not exist, it is created with value suffix. Returns {:ok, new_byte_length}.

@spec batch_async_put(FerricStore.Instance.t(), [{binary(), binary()}]) ::
  :ok | {:error, binary()}

Batch async PUT for pipelined SET commands without options.

Takes a list of {key, value} tuples. All keys must target async durability namespaces. Groups by shard, does batch ETS inserts per shard, fires BitcaskWriter casts and Raft submissions individually (they batch internally). Returns :ok or {:error, reason}.

Caller must validate key/value sizes and check pressure flags before calling. This skips all per-key validation for speed.

@spec batch_quorum_put(FerricStore.Instance.t(), [{binary(), binary()}]) :: [
  :ok | {:error, binary()}
]

Batch quorum PUT for pipelined SET commands.

Groups commands by shard, submits each group as a single batch to its Batcher with ONE reply ref per shard, then waits for all shard replies. Returns a list of results in input order.

Uses one ref per shard (not per command) so the connection process's selective receive scans at most shard_count refs instead of N*pipeline refs — critical for high concurrency where TCP messages flood the mailbox.

@spec bitmap_op(FerricStore.Instance.t(), binary(), [term()]) :: term()

Runs a Bitmap RMW command (BITOP / BITFIELD) atomically via Raft. SETBIT has its own setbit/4 path. GETBIT/BITCOUNT/BITPOS are read-only and do not go through this path.

@spec dbsize(FerricStore.Instance.t()) :: non_neg_integer()

Returns the count of all live keys across every shard.

@spec delete(FerricStore.Instance.t(), binary()) :: :ok

Deletes key. Returns :ok whether or not the key existed.

@spec durability_for_key_public(FerricStore.Instance.t(), binary()) ::
  :quorum | :async

Public wrapper for durability_for_key, used by batch SET fast path.

@spec execute_list_op_inline(FerricStore.Instance.t(), non_neg_integer(), tuple()) ::
  term()

Executes a list_op inline under a held latch. Called from Router.async_list_op (fast path) and RmwCoordinator.handle_call (contended path). The latch guarantees exclusive access to the list's compound keys.

Uses ListOps.execute/3 with an origin-local compound store that writes ETS + casts BitcaskWriter for each mutation. Then submits the original {:list_op, key, op} command to Raft via Batcher.async_submit so replicas re-execute against their own state in Raft log order (deterministic convergence).

@spec execute_rmw_inline(FerricStore.Instance.t(), non_neg_integer(), tuple()) ::
  term()

Executes an RMW command inline against local ETS + BitcaskWriter and submits the delta to Raft via Batcher.async_submit.

Called with the per-key latch held. The latch guarantees exclusive access to key's ETS row among RMW paths. Router.async_rmw/4 (latch path) and Ferricstore.Store.RmwCoordinator (worker path) both call this after winning the latch.

Returns the command's natural result shape (e.g. {:ok, new_int} for INCR, old_value_or_nil for GETSET/GETDEL).

@spec exists?(FerricStore.Instance.t(), binary()) :: boolean()

Returns true if key exists and is not expired.

Uses direct ETS lookup (no GenServer roundtrip) for hot and cold keys. A key is considered existing if it is in the keydir and not expired, regardless of whether its value is hot (in ETS) or cold (on disk only).

@spec exists_fast?(FerricStore.Instance.t(), binary()) :: boolean()

Fast ETS-direct existence check for a key.

Returns true if the key exists in ETS and is not expired, false otherwise. This bypasses the GenServer entirely, saving ~1-3us per call. Used in the hot write path (check_keydir_full/2) where we only need a boolean answer and can tolerate the fact that cold keys (value=nil but still in keydir) are correctly detected as existing.

@spec extract_hash_tag(binary()) :: binary() | nil

Extracts the hash tag from a key, following Redis hash tag semantics.

If the key contains a substring enclosed in {...} where the content between the first { and the next } is non-empty, that substring is used for hashing instead of the full key. This allows related keys to be routed to the same shard.

Examples

iex> Ferricstore.Store.Router.extract_hash_tag("{user:42}:session")
"user:42"

iex> Ferricstore.Store.Router.extract_hash_tag("no_tag")
nil

iex> Ferricstore.Store.Router.extract_hash_tag("{}empty")
nil

@spec geo_op(FerricStore.Instance.t(), binary(), [term()]) :: term()

Runs a Geo RMW command (GEOADD, GEOSEARCHSTORE) atomically via Raft. GEOSEARCHSTORE routes by the destination key; GEOADD routes by the key. Read-only ops (GEOPOS, GEODIST, GEOHASH, GEOSEARCH) don't go through here.

@spec get(FerricStore.Instance.t(), binary()) :: binary() | nil

Retrieves the value for key, or nil if the key does not exist or is expired.

Hot path: reads directly from ETS (no GenServer roundtrip for cached keys). Falls back to a GenServer call for cache misses or when the ETS table is temporarily unavailable (e.g. during a shard restart).

Each successful read is recorded as either hot (ETS hit) or cold (Bitcask fallback) in Ferricstore.Stats for the FERRICSTORE.HOTNESS command and the INFO stats hot/cold fields.

@spec get_file_ref(FerricStore.Instance.t(), binary()) ::
  {binary(), non_neg_integer(), non_neg_integer()} | nil

Returns the on-disk file reference for a key's value, or nil.

Used by the sendfile optimisation in standalone TCP mode. Returns {file_path, value_byte_offset, value_size} for cold (on-disk) keys. Returns nil for hot keys (ETS), expired keys, or missing keys -- the caller should fall back to the normal read path.

Only cold keys benefit from sendfile: hot keys are already in BEAM memory and would need a normal get + transport.send.

@spec get_keydir_file_ref(FerricStore.Instance.t(), binary()) ::
  {:ok, {non_neg_integer(), non_neg_integer(), non_neg_integer()}} | :miss

Returns the keydir disk location for a key, or :miss.

Reads the {file_id, offset, value_size} fields directly from the keydir ETS table without a GenServer roundtrip. Returns {:ok, {fid, off, vsize}} for live keys, or :miss if the key is not in the keydir or is expired.

Used by sendfile zero-copy and STRLEN on cold keys.

@spec get_meta(FerricStore.Instance.t(), binary()) ::
  {binary(), non_neg_integer()} | nil

Returns {value, expire_at_ms} for a live key, or nil if the key does not exist or is expired.

Hot path: reads directly from ETS for cached keys. Each read is recorded as hot or cold in Ferricstore.Stats.

@spec get_version(FerricStore.Instance.t(), binary()) :: non_neg_integer()

Returns the current write version of the shard that owns key.

Used by the WATCH/EXEC transaction mechanism to detect concurrent modifications.

@spec get_with_file_ref(FerricStore.Instance.t(), binary()) ::
  {:hot, binary()}
  | {:cold_ref, binary(), non_neg_integer(), non_neg_integer()}
  | {:cold_value, binary()}
  | :miss

Unified GET that returns everything from a single ETS lookup.

Returns:

• {:hot, value} — value is in ETS, ready to return
• {:cold_ref, path, offset, size} — value is on disk, file ref for sendfile
• {:cold_value, value} — value was on disk, GenServer fetched it
• :miss — key doesn't exist

@spec getdel(FerricStore.Instance.t(), binary()) :: binary() | nil

Atomically gets and deletes key.

Returns the value, or nil if the key did not exist.

@spec getex(FerricStore.Instance.t(), binary(), non_neg_integer()) :: binary() | nil

Atomically gets the value and updates the expiry of key.

expire_at_ms is an absolute Unix-epoch timestamp in milliseconds; pass 0 to persist (remove expiry). Returns the value, or nil if the key did not exist.

@spec getset(FerricStore.Instance.t(), binary(), binary()) :: binary() | nil

Atomically gets the old value and sets a new value for key.

Returns the old value, or nil if the key did not exist.

@spec hincrby(FerricStore.Instance.t(), binary(), binary(), integer()) ::
  integer() | {:error, binary()}

Atomically increments the integer value of hash field field in key by delta. Returns {:ok, new_int} or {:error, reason}. Shares ordering with the parent hash's shard (routes by the hash's redis_key).

@spec hincrbyfloat(FerricStore.Instance.t(), binary(), binary(), float()) ::
  binary() | {:error, binary()}

Atomically increments the float value of hash field field in key by delta. Returns the new value as a string, or {:error, reason}.

@spec hll_op(FerricStore.Instance.t(), binary(), [term()]) :: term()

Runs a HyperLogLog RMW command (PFADD / PFMERGE) atomically via Raft. PFCOUNT is read-only and should not go through this path.

@spec incr(FerricStore.Instance.t(), binary(), integer()) ::
  {:ok, integer()} | {:error, binary()}

Atomically increments the integer value of key by delta.

If the key does not exist, it is set to delta. Returns {:ok, new_integer} on success or {:error, reason} if the value is not a valid integer.

@spec incr_float(FerricStore.Instance.t(), binary(), float()) ::
  {:ok, binary()} | {:error, binary()}

Atomically increments the float value of key by delta.

If the key does not exist, it is set to delta. Returns {:ok, new_float_string} on success or {:error, reason} if the value is not a valid float.

@spec json_op(FerricStore.Instance.t(), binary(), [term()]) :: term()

Runs a JSON RMW command atomically via Raft. cmd is the Redis command name (e.g. "JSON.SET"), args is the argument list starting with the key. The state machine dispatches to Ferricstore.Commands.Json.handle/3 with a state-machine-scoped store, so concurrent callers serialize through the Raft log — no lost updates.

@spec keydir_name(FerricStore.Instance.t(), non_neg_integer()) :: atom() | reference()

Returns the keydir ETS table ref for the shard at index.

Uses the pre-computed tuple from the instance context for O(1) lookup.

@spec keys(FerricStore.Instance.t()) :: [binary()]

Returns all live (non-expired, non-deleted) keys across every shard.

@spec max_value_size() :: pos_integer()

Returns the maximum allowed value size in bytes.

@spec prob_write(FerricStore.Instance.t(), tuple()) :: term()

Routes a probabilistic data structure write command through Raft.

@spec put(FerricStore.Instance.t(), binary(), binary(), non_neg_integer()) ::
  :ok | {:error, binary()}

Stores key with value. expire_at_ms is an absolute Unix-epoch timestamp in milliseconds; pass 0 for no expiry.

@spec server_command(FerricStore.Instance.t(), term()) :: term()

Submits a server command through Raft for replication to all nodes.

Server commands are opaque to the library — the state machine dispatches them via the raft_apply_hook callback on the Instance struct. Routed through shard 0 for consistent ordering.

@spec setbit(FerricStore.Instance.t(), binary(), non_neg_integer(), 0 | 1) :: 0 | 1

Atomically sets the bit at offset to bit_val (0 or 1). Returns the previous bit value (0 or 1). Extends the bitmap with zero bytes if necessary. Goes through Raft so concurrent SETBITs on the same key never lose updates — the state machine is the sole mutator.

@spec setrange(FerricStore.Instance.t(), binary(), non_neg_integer(), binary()) ::
  {:ok, non_neg_integer()}

Atomically overwrites part of the string at key starting at offset.

Zero-pads if the key doesn't exist or the string is shorter than offset. Returns {:ok, new_byte_length}.

@spec shard_for(FerricStore.Instance.t(), binary()) :: non_neg_integer()

Returns the shard index (0-based) that owns key.

Routes through the 1,024-slot indirection layer: key -> phash2(key) & 0x3FF -> slot -> slot_map[slot] -> shard_index

Supports Redis hash tags: if the key contains {tag} (non-empty content between the first { and the next }), the tag is used for hashing instead of the full key.

@spec shard_name(FerricStore.Instance.t(), non_neg_integer()) :: atom()

Returns the registered process name for the shard at index.

Uses the pre-computed tuple from the instance context for O(1) lookup.

@spec slot_for(FerricStore.Instance.t(), binary()) :: non_neg_integer()

Returns the slot (0-1023) for a key, respecting hash tags.

@spec tdigest_op(FerricStore.Instance.t(), binary(), [term()]) :: term()

Runs a TDigest RMW command (TDIGEST.ADD / RESET / MERGE / CREATE) atomically via Raft. Read-only ops (QUANTILE, CDF, INFO, RANK, etc.) stay in the caller process.

@spec zincrby(FerricStore.Instance.t(), binary(), number(), binary()) ::
  binary() | {:error, binary()}

Atomically increments the score of member in the sorted set at key by increment. Returns the new score as a string.