defmodule FerricStore do
  @moduledoc """
  Direct Elixir API for FerricStore in embedded mode.

  When FerricStore runs embedded inside a host Elixir application, this module
  provides direct function calls that bypass TCP, RESP3 parsing, and ACL checks.
  These are thin wrappers around the same shard routing and group-commit pipeline
  that standalone mode uses, at ETS lookup latency (~1-5us for hot reads).

  ## Usage

      FerricStore.set("user:42:name", "alice", ttl: :timer.hours(1))
      {:ok, "alice"} = FerricStore.get("user:42:name")
      :ok = FerricStore.del("user:42:name")

  ## Sandbox support

  All API functions are sandbox-aware. When `FerricStore.Sandbox.checkout/1` has
  been called in the current process, a namespace prefix is transparently prepended
  to all keys. Production code runs with `Process.get(:ferricstore_sandbox) == nil`
  -- zero overhead, zero behaviour change. See `FerricStore.Sandbox` for details.

  ## Named caches

  Pass the `:cache` option to direct operations to a named cache instance:

      FerricStore.set("session:abc", data, cache: :sessions)
      FerricStore.get("session:abc", cache: :sessions)

  When `:cache` is not specified, the default cache is used.
  """

  alias Ferricstore.Store.Router

  # ---------------------------------------------------------------------------
  # Readiness
  # ---------------------------------------------------------------------------

  @doc """
  Blocks until FerricStore is fully ready to serve requests.

  Polls `Health.check/0` until all shards are alive and all Raft leaders
  are elected. Returns `:ok` when ready, raises on timeout.

  Call this in your application's `start/2` after FerricStore is in your
  supervision tree, or in test setup, to ensure writes won't fail.

  ## Options

    * `:timeout` - max milliseconds to wait (default: 30_000)
    * `:interval` - polling interval in ms (default: 100)

  ## Examples

      # In your Application.start/2:
      def start(_type, _args) do
        children = [
          {FerricStore, []},
          MyApp.Repo,
          MyAppWeb.Endpoint
        ]
        opts = [strategy: :one_for_one, name: MyApp.Supervisor]
        {:ok, pid} = Supervisor.start_link(children, opts)

        FerricStore.await_ready()
        {:ok, pid}
      end

      # With custom timeout:
      FerricStore.await_ready(timeout: 60_000)

  """
  @spec await_ready(keyword()) :: :ok
  def await_ready(opts \\ []) do
    timeout = Keyword.get(opts, :timeout, 30_000)
    interval = Keyword.get(opts, :interval, 100)
    deadline = System.monotonic_time(:millisecond) + timeout

    do_await_ready(deadline, interval)
  end

  defp do_await_ready(deadline, interval) do
    case Ferricstore.Health.check() do
      %{status: :ok} ->
        :ok

      _ ->
        if System.monotonic_time(:millisecond) > deadline do
          raise "FerricStore not ready within timeout. Check shard health with FerricStore.health()"
        end

        Process.sleep(interval)
        do_await_ready(deadline, interval)
    end
  end

  @doc """
  Returns the current health status without blocking.

  ## Examples

      iex> FerricStore.health()
      %{status: :ok, shard_count: 4, shards: [...], uptime_seconds: 120}

  """
  @spec health() :: Ferricstore.Health.health_result()
  def health do
    Ferricstore.Health.check()
  end

  @doc """
  Returns `true` if FerricStore is ready to serve requests.

  ## Examples

      iex> FerricStore.ready?()
      true

  """
  @spec ready?() :: boolean()
  def ready? do
    Ferricstore.Health.ready?()
  end

  @doc """
  Gracefully shuts down FerricStore, flushing all pending data to disk.

  Flushes Raft batchers, BitcaskWriters, shard pending writes, and
  triggers a WAL rollover. Call before stopping the application to
  ensure zero data loss.

  ## Examples

      FerricStore.shutdown()
      Application.stop(:ferricstore)

  """
  @spec shutdown() :: :ok
  def shutdown do
    Ferricstore.Application.prep_stop(nil)
    :ok
  end

  # ---------------------------------------------------------------------------
  # Types
  # ---------------------------------------------------------------------------

  @type key :: binary()
  @type value :: binary()
  @type set_opts :: [
          ttl: non_neg_integer(),
          exat: pos_integer(),
          pxat: pos_integer(),
          nx: boolean(),
          xx: boolean(),
          get: boolean(),
          keepttl: boolean(),
          cache: atom()
        ]
  @type get_opts :: [cache: atom()]
  @type cas_opts :: [ttl: non_neg_integer()]
  @type fetch_or_compute_opts :: [ttl: pos_integer(), hint: binary()]
  @type zrange_opts :: [withscores: boolean()]

  # ---------------------------------------------------------------------------
  # Strings
  # ---------------------------------------------------------------------------

  @doc """
  Sets `key` to `value` with optional TTL and condition flags.

  ## Options

    * `:ttl` - Time-to-live in milliseconds (relative). When omitted or `0`,
      the key never expires. Mutually exclusive with `:exat`, `:pxat`, `:keepttl`.
    * `:exat` - Absolute Unix timestamp in seconds at which the key expires.
      Mutually exclusive with `:ttl`, `:pxat`, `:keepttl`.
    * `:pxat` - Absolute Unix timestamp in milliseconds at which the key expires.
      Mutually exclusive with `:ttl`, `:exat`, `:keepttl`.
    * `:nx` - Only set the key if it does not already exist.
    * `:xx` - Only set the key if it already exists.
    * `:get` - Return the old value stored at the key before overwriting.
      When set, the return value changes to `{:ok, old_value}` (or `{:ok, nil}`
      if the key did not exist).
    * `:keepttl` - Retain the existing TTL associated with the key instead of
      clearing it. Mutually exclusive with `:ttl`, `:exat`, `:pxat`.
    * `:cache` - Named cache instance (default: the default cache).

  ## Examples

      iex> FerricStore.set("user:42:name", "alice")
      :ok

      iex> FerricStore.set("session:abc", "token_data", ttl: :timer.hours(1))
      :ok

      iex> FerricStore.set("cache:page:home", "html", exat: 1711234567)
      :ok

      iex> FerricStore.set("lock:order:99", "owner_1", nx: true)
      :ok

      iex> FerricStore.set("lock:order:99", "owner_2", nx: true)
      nil

      iex> FerricStore.set("counter", "0")
      :ok
      iex> FerricStore.set("counter", "100", get: true)
      {:ok, "0"}

      iex> FerricStore.set("missing", "val", get: true)
      {:ok, nil}

      iex> FerricStore.set("session:abc", "refreshed", keepttl: true)
      :ok

  Returns `{:error, reason}` if the value exceeds the configured
  `max_value_size`.
  """
  @spec set(key(), value(), set_opts()) :: :ok | {:ok, value() | nil} | nil | {:error, binary()}
  def set(key, value, opts \\ []) do
    max_value_size =
      Application.get_env(:ferricstore, :max_value_size, Ferricstore.Resp.Parser.default_max_value_size())

    if is_binary(value) and byte_size(value) > max_value_size do
      {:error, "ERR value too large (#{byte_size(value)} bytes, max #{max_value_size} bytes)"}
    else
      set_inner(key, value, opts)
    end
  end

  defp set_inner(key, value, opts) do
    resolved_key = sandbox_key(key)
    ttl = Keyword.get(opts, :ttl, 0)
    exat = Keyword.get(opts, :exat)
    pxat = Keyword.get(opts, :pxat)
    nx? = Keyword.get(opts, :nx, false)
    xx? = Keyword.get(opts, :xx, false)
    get? = Keyword.get(opts, :get, false)
    keepttl? = Keyword.get(opts, :keepttl, false)

    # Determine expire_at_ms from the expiry options (mutually exclusive)
    {expire_at_ms, from_keepttl?} =
      cond do
        keepttl? -> {0, true}
        exat != nil -> {exat * 1000, false}
        pxat != nil -> {pxat, false}
        ttl > 0 -> {System.os_time(:millisecond) + ttl, false}
        true -> {0, false}
      end

    # Read old metadata when GET or KEEPTTL is needed
    {old_value, effective_expire} =
      if get? or from_keepttl? do
        case Router.get_meta(resolved_key) do
          nil ->
            {nil, expire_at_ms}

          {old_val, old_exp} ->
            eff_exp = if from_keepttl?, do: old_exp, else: expire_at_ms
            {old_val, eff_exp}
        end
      else
        {nil, expire_at_ms}
      end

    # Condition check
    skip? =
      cond do
        nx? and Router.exists?(resolved_key) -> true
        xx? and not Router.exists?(resolved_key) -> true
        true -> false
      end

    if skip? do
      if get?, do: {:ok, old_value}, else: nil
    else
      Router.put(resolved_key, value, effective_expire)
      if get?, do: {:ok, old_value}, else: :ok
    end
  end

  @doc """
  Gets the value stored at `key`.

  Returns `{:ok, value}` if the key exists and has not expired, or `{:ok, nil}`
  if the key does not exist or has expired.

  ## Options

    * `:cache` - Named cache instance (default: the default cache).

  ## Examples

      iex> FerricStore.set("user:42:name", "alice")
      :ok
      iex> FerricStore.get("user:42:name")
      {:ok, "alice"}

      iex> FerricStore.get("nonexistent:key")
      {:ok, nil}

  """
  @spec get(key(), get_opts()) :: {:ok, value() | nil}
  def get(key, opts \\ []) do
    _cache = Keyword.get(opts, :cache)

    result =
      key
      |> sandbox_key()
      |> Router.get()

    {:ok, result}
  end

  @doc """
  Deletes `key` from the store.

  Returns `:ok` regardless of whether the key existed.

  ## Examples

      iex> FerricStore.set("session:temp", "data")
      :ok
      iex> FerricStore.del("session:temp")
      :ok

      iex> FerricStore.del("nonexistent:key")
      :ok

  """
  @spec del(key()) :: :ok
  def del(key) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    Ferricstore.Commands.Strings.handle("DEL", [resolved_key], store)
    :ok
  end

  @doc """
  Increments the integer value stored at `key` by 1.

  If the key does not exist, it is initialized to `0` before incrementing,
  resulting in a value of `1`. Returns `{:error, reason}` if the stored value
  cannot be parsed as an integer.

  ## Examples

      iex> FerricStore.incr("page:views:home")
      {:ok, 1}

      iex> FerricStore.incr("page:views:home")
      {:ok, 2}

      iex> FerricStore.set("name", "alice")
      :ok
      iex> FerricStore.incr("name")
      {:error, "ERR value is not an integer or out of range"}

  """
  @spec incr(key()) :: {:ok, integer()} | {:error, binary()}
  def incr(key) do
    incr_by(key, 1)
  end

  @doc """
  Decrements the integer value stored at `key` by 1.

  If the key does not exist, it is initialized to `0` before decrementing,
  resulting in a value of `-1`. Returns `{:error, reason}` if the stored value
  cannot be parsed as an integer.

  ## Examples

      iex> FerricStore.decr("rate_limit:user:42")
      {:ok, -1}

      iex> FerricStore.set("stock:item:99", "10")
      :ok
      iex> FerricStore.decr("stock:item:99")
      {:ok, 9}

  """
  @spec decr(key()) :: {:ok, integer()} | {:error, binary()}
  def decr(key) do
    incr_by(key, -1)
  end

  @doc """
  Decrements the integer value stored at `key` by `amount`.

  If the key does not exist, it is initialized to `0` before decrementing.
  Returns `{:error, reason}` if the stored value is not a valid integer.

  ## Examples

      iex> FerricStore.set("stock:item:99", "100")
      :ok
      iex> FerricStore.decr_by("stock:item:99", 10)
      {:ok, 90}

      iex> FerricStore.decr_by("new_counter", 5)
      {:ok, -5}

  """
  @spec decr_by(key(), integer()) :: {:ok, integer()} | {:error, binary()}
  def decr_by(key, amount) when is_integer(amount) do
    incr_by(key, -amount)
  end

  @doc """
  Increments the integer value stored at `key` by `amount`.

  If the key does not exist, it is initialized to `0` before incrementing.
  Returns `{:error, reason}` if the stored value is not a valid integer.

  ## Examples

      iex> FerricStore.incr_by("page:views:home", 10)
      {:ok, 10}

      iex> FerricStore.incr_by("page:views:home", 5)
      {:ok, 15}

      iex> FerricStore.set("name", "alice")
      :ok
      iex> FerricStore.incr_by("name", 1)
      {:error, "ERR value is not an integer or out of range"}

  """
  @spec incr_by(key(), integer()) :: {:ok, integer()} | {:error, binary()}
  def incr_by(key, amount) when is_integer(amount) do
    resolved_key = sandbox_key(key)

    current = Router.get(resolved_key)

    case parse_int_value(current) do
      {:ok, int_val} ->
        new_val = int_val + amount
        Router.put(resolved_key, Integer.to_string(new_val), 0)
        {:ok, new_val}

      {:error, _} = err ->
        err
    end
  end

  @doc """
  Increments the numeric value stored at `key` by a floating-point `amount`.

  If the key does not exist, it is initialized to `0.0` before incrementing.
  The new value is returned as a string representation. Returns
  `{:error, reason}` if the stored value is not a valid number.

  ## Examples

      iex> FerricStore.incr_by_float("price:item:99", 3.14)
      {:ok, "3.14"}

      iex> FerricStore.set("balance:user:42", "100.50")
      :ok
      iex> FerricStore.incr_by_float("balance:user:42", -20.25)
      {:ok, "80.25"}

  """
  @spec incr_by_float(key(), float()) :: {:ok, binary()} | {:error, binary()}
  def incr_by_float(key, amount) when is_number(amount) do
    resolved_key = sandbox_key(key)

    case Router.incr_float(resolved_key, amount * 1.0) do
      {:ok, result} -> {:ok, result}
      {:error, _} = err -> err
    end
  end

  @doc """
  Gets values for multiple keys in a single call.

  Returns `{:ok, values}` where `values` is a list in the same order as the
  input keys. Missing or expired keys appear as `nil` in the result list.

  ## Examples

      iex> FerricStore.set("user:1:name", "alice")
      :ok
      iex> FerricStore.set("user:2:name", "bob")
      :ok
      iex> FerricStore.mget(["user:1:name", "user:2:name", "user:3:name"])
      {:ok, ["alice", "bob", nil]}

  """
  @spec mget([key()]) :: {:ok, [value() | nil]}
  def mget(keys) when is_list(keys) do
    values = Enum.map(keys, fn key ->
      key |> sandbox_key() |> Router.get()
    end)
    {:ok, values}
  end

  @doc """
  Sets multiple key-value pairs in a single call.

  All pairs are written without expiry. Use `set/3` with `:ttl` if individual
  keys need time-to-live.

  ## Examples

      iex> FerricStore.mset(%{"user:1:name" => "alice", "user:2:name" => "bob"})
      :ok

      iex> FerricStore.get("user:1:name")
      {:ok, "alice"}

  """
  @spec mset(%{key() => value()}) :: :ok
  def mset(pairs) when is_map(pairs) do
    Enum.each(pairs, fn {key, value} ->
      key |> sandbox_key() |> Router.put(value, 0)
    end)
    :ok
  end

  @doc """
  Appends `suffix` to the string value stored at `key`.

  If the key does not exist, it is created with `suffix` as its value.
  Returns the byte length of the string after the append.

  ## Examples

      iex> FerricStore.set("log:request:42", "GET /api")
      :ok
      iex> FerricStore.append("log:request:42", " 200 OK")
      {:ok, 15}

      iex> FerricStore.append("new:key", "hello")
      {:ok, 5}

  """
  @spec append(key(), binary()) :: {:ok, non_neg_integer()}
  def append(key, suffix) do
    resolved_key = sandbox_key(key)
    case Router.append(resolved_key, suffix) do
      {:ok, len} -> {:ok, len}
      len when is_integer(len) -> {:ok, len}
    end
  end

  @doc """
  Returns the byte length of the string value stored at `key`.

  Returns `{:ok, 0}` if the key does not exist.

  ## Examples

      iex> FerricStore.set("user:42:name", "alice")
      :ok
      iex> FerricStore.strlen("user:42:name")
      {:ok, 5}

      iex> FerricStore.strlen("nonexistent:key")
      {:ok, 0}

  """
  @spec strlen(key()) :: {:ok, non_neg_integer()}
  def strlen(key) do
    resolved_key = sandbox_key(key)
    case Router.get(resolved_key) do
      nil -> {:ok, 0}
      value -> {:ok, byte_size(value)}
    end
  end

  @doc """
  Atomically sets `key` to `value` and returns the previous value.

  Returns `{:ok, old_value}` or `{:ok, nil}` if the key did not previously
  exist. Useful for atomic swap patterns like rotating session tokens.

  ## Examples

      iex> FerricStore.set("session:token", "tok_abc")
      :ok
      iex> FerricStore.getset("session:token", "tok_xyz")
      {:ok, "tok_abc"}

      iex> FerricStore.getset("fresh:key", "first_value")
      {:ok, nil}

  """
  @spec getset(key(), value()) :: {:ok, value() | nil}
  def getset(key, value) do
    resolved_key = sandbox_key(key)
    result = Router.getset(resolved_key, value)
    {:ok, result}
  end

  @doc """
  Atomically gets the value of `key` and deletes it.

  Returns `{:ok, value}` or `{:ok, nil}` if the key did not exist. Useful
  for consuming one-time tokens or dequeuing single values.

  ## Examples

      iex> FerricStore.set("otp:user:42", "839201")
      :ok
      iex> FerricStore.getdel("otp:user:42")
      {:ok, "839201"}
      iex> FerricStore.getdel("otp:user:42")
      {:ok, nil}

  """
  @spec getdel(key()) :: {:ok, value() | nil}
  def getdel(key) do
    resolved_key = sandbox_key(key)
    result = Router.getdel(resolved_key)
    {:ok, result}
  end

  @doc """
  Gets the value of `key` and optionally updates its expiry.

  When called without options, behaves identically to `get/2`. Pass `:ttl`
  to refresh the expiry on access, or `:persist` to remove it.

  ## Options

    * `:ttl` - New TTL in milliseconds to set on the key.
    * `:persist` - When `true`, removes any existing TTL, making the key persistent.

  ## Examples

      iex> FerricStore.set("session:abc", "data", ttl: 10_000)
      :ok
      iex> FerricStore.getex("session:abc", ttl: 60_000)
      {:ok, "data"}

      iex> FerricStore.getex("session:abc", persist: true)
      {:ok, "data"}

      iex> FerricStore.getex("nonexistent:key")
      {:ok, nil}

  """
  @spec getex(key(), keyword()) :: {:ok, value() | nil}
  def getex(key, opts \\ []) do
    resolved_key = sandbox_key(key)

    expire_at_ms =
      cond do
        Keyword.get(opts, :persist, false) ->
          0

        ttl = Keyword.get(opts, :ttl) ->
          System.os_time(:millisecond) + ttl

        true ->
          nil
      end

    case expire_at_ms do
      nil ->
        {:ok, Router.get(resolved_key)}

      ms ->
        result = Router.getex(resolved_key, ms)
        {:ok, result}
    end
  end

  @doc """
  Sets `key` to `value` only if the key does not already exist.

  Returns `{:ok, true}` if the key was created, or `{:ok, false}` if the key
  already existed and the write was skipped.

  ## Examples

      iex> FerricStore.setnx("lock:job:import", "worker_1")
      {:ok, true}

      iex> FerricStore.setnx("lock:job:import", "worker_2")
      {:ok, false}

  """
  @spec setnx(key(), value()) :: {:ok, boolean()}
  def setnx(key, value) do
    resolved_key = sandbox_key(key)
    if Router.exists?(resolved_key) do
      {:ok, false}
    else
      Router.put(resolved_key, value, 0)
      {:ok, true}
    end
  end

  @doc """
  Sets `key` to `value` with a TTL in seconds.

  This is a convenience wrapper equivalent to
  `set(key, value, ttl: seconds * 1_000)`.

  ## Examples

      iex> FerricStore.setex("session:abc", 3600, "token_data")
      :ok

      iex> FerricStore.setex("cache:query:recent", 60, "[\"row1\",\"row2\"]")
      :ok

  """
  @spec setex(key(), pos_integer(), value()) :: :ok
  def setex(key, seconds, value) do
    resolved_key = sandbox_key(key)
    expire_at_ms = System.os_time(:millisecond) + seconds * 1_000
    Router.put(resolved_key, value, expire_at_ms)
  end

  @doc """
  Sets `key` to `value` with a TTL in milliseconds.

  This is a convenience wrapper equivalent to
  `set(key, value, ttl: milliseconds)`.

  ## Examples

      iex> FerricStore.psetex("rate_limit:user:42", 500, "1")
      :ok

      iex> FerricStore.psetex("debounce:click", 200, "pending")
      :ok

  """
  @spec psetex(key(), pos_integer(), value()) :: :ok
  def psetex(key, milliseconds, value) do
    resolved_key = sandbox_key(key)
    expire_at_ms = System.os_time(:millisecond) + milliseconds
    Router.put(resolved_key, value, expire_at_ms)
  end

  @doc """
  Returns a substring of the string stored at `key` between byte offsets `start` and `stop` (inclusive).

  Negative offsets count from the end of the string (`-1` is the last byte).
  Returns `{:ok, ""}` if the key does not exist or the range is empty.

  ## Examples

      iex> FerricStore.set("greeting", "Hello, World!")
      :ok
      iex> FerricStore.getrange("greeting", 7, 11)
      {:ok, "World"}

      iex> FerricStore.getrange("greeting", -6, -1)
      {:ok, "orld!"}

      iex> FerricStore.getrange("nonexistent", 0, 10)
      {:ok, ""}

  """
  @spec getrange(key(), integer(), integer()) :: {:ok, binary()}
  def getrange(key, start, stop) do
    resolved_key = sandbox_key(key)
    value = Router.get(resolved_key) || ""
    len = byte_size(value)

    if len == 0 do
      {:ok, ""}
    else
      s = if start < 0, do: max(0, len + start), else: min(start, len - 1)
      e = if stop < 0, do: max(0, len + stop), else: min(stop, len - 1)

      if s > e do
        {:ok, ""}
      else
        {:ok, binary_part(value, s, e - s + 1)}
      end
    end
  end

  @doc """
  Overwrites part of the string stored at `key` starting at byte `offset`.

  If the key does not exist, or the existing string is shorter than `offset`,
  the value is zero-padded to reach the offset before writing. Returns
  `{:ok, new_byte_length}` with the total length after the write.

  ## Examples

      iex> FerricStore.set("greeting", "Hello World")
      :ok
      iex> FerricStore.setrange("greeting", 6, "Redis")
      {:ok, 11}
      iex> FerricStore.get("greeting")
      {:ok, "Hello Redis"}

      iex> FerricStore.setrange("padded:key", 5, "!")
      {:ok, 6}

  """
  @spec setrange(key(), non_neg_integer(), binary()) :: {:ok, non_neg_integer()}
  def setrange(key, offset, value) do
    resolved_key = sandbox_key(key)
    case Router.setrange(resolved_key, offset, value) do
      {:ok, len} -> {:ok, len}
      len when is_integer(len) -> {:ok, len}
    end
  end

  @doc """
  Sets multiple key-value pairs only if none of the given keys already exist.

  This is atomic: either all keys are set, or none are. If any key in the
  map already exists, the entire operation is skipped and `{:ok, false}` is
  returned.

  ## Examples

      iex> FerricStore.msetnx(%{"user:1:email" => "a@test.com", "user:2:email" => "b@test.com"})
      {:ok, true}

      iex> FerricStore.msetnx(%{"user:1:email" => "new@test.com", "user:3:email" => "c@test.com"})
      {:ok, false}

  """
  @spec msetnx(%{key() => value()}) :: {:ok, boolean()}
  def msetnx(pairs) when is_map(pairs) do
    resolved_pairs = Enum.map(pairs, fn {k, v} -> {sandbox_key(k), v} end)

    any_exists = Enum.any?(resolved_pairs, fn {k, _v} -> Router.exists?(k) end)

    if any_exists do
      {:ok, false}
    else
      Enum.each(resolved_pairs, fn {k, v} -> Router.put(k, v, 0) end)
      {:ok, true}
    end
  end

  # ---------------------------------------------------------------------------
  # Hash
  # ---------------------------------------------------------------------------

  @doc """
  Sets one or more fields in the hash stored at `key`.

  `fields` is a map of `%{field_name => value}`. Field names and values are
  stored as binaries. If the hash does not exist, a new one is created.
  Existing fields are overwritten.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30"})
      :ok

      iex> FerricStore.hset("user:42", %{"name" => "bob"})
      :ok

  """
  @spec hset(key(), %{binary() => binary()}) :: :ok
  def hset(key, fields) when is_map(fields) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    args =
      Enum.flat_map(fields, fn {k, v} -> [to_string(k), to_string(v)] end)

    case Ferricstore.Commands.Hash.handle("HSET", [resolved_key | args], store) do
      {:error, _} = err -> err
      _count -> :ok
    end
  end

  @doc """
  Gets the value of a single field from the hash stored at `key`.

  Returns `{:ok, value}` if the field exists, or `{:ok, nil}` if the field
  or the hash does not exist.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30"})
      iex> FerricStore.hget("user:42", "name")
      {:ok, "alice"}

      iex> FerricStore.hget("user:42", "nonexistent_field")
      {:ok, nil}

      iex> FerricStore.hget("no_such_hash", "field")
      {:ok, nil}

  """
  @spec hget(key(), binary()) :: {:ok, binary() | nil}
  def hget(key, field) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle("HGET", [resolved_key, to_string(field)], store) do
      {:error, _} = err -> err
      result -> {:ok, result}
    end
  end

  @doc """
  Gets all fields and values from the hash stored at `key`.

  Returns `{:ok, map}` where `map` is a `%{field => value}` map. If the key
  does not exist, returns `{:ok, %{}}`.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30"})
      iex> FerricStore.hgetall("user:42")
      {:ok, %{"name" => "alice", "age" => "30"}}

      iex> FerricStore.hgetall("no_such_hash")
      {:ok, %{}}

  """
  @spec hgetall(key()) :: {:ok, %{binary() => binary()}}
  def hgetall(key) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle("HGETALL", [resolved_key], store) do
      {:error, _} = err -> err
      flat_list ->
        map =
          flat_list
          |> Enum.chunk_every(2)
          |> Map.new(fn [f, v] -> {f, v} end)

        {:ok, map}
    end
  end

  # ---------------------------------------------------------------------------
  # Lists
  # ---------------------------------------------------------------------------

  @doc """
  Pushes one or more elements to the left (head) of the list stored at `key`.

  If the key does not exist, a new list is created. Elements are inserted
  left-to-right, so the last element in the list ends up as the leftmost
  element (matching Redis LPUSH semantics).

  ## Examples

      iex> FerricStore.lpush("tasks:queue", ["send_email"])
      {:ok, 1}

      iex> FerricStore.lpush("tasks:queue", ["generate_report", "resize_image"])
      {:ok, 3}

  """
  @spec lpush(key(), [binary()]) :: {:ok, non_neg_integer()}
  def lpush(key, elements) when is_list(elements) do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:lpush, elements})

    {:ok, result}
  end

  @doc """
  Pushes one or more elements to the right (tail) of the list stored at `key`.

  If the key does not exist, a new list is created.

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["send_email"])
      {:ok, 1}

      iex> FerricStore.rpush("tasks:queue", ["generate_report", "resize_image"])
      {:ok, 3}

  """
  @spec rpush(key(), [binary()]) :: {:ok, non_neg_integer()}
  def rpush(key, elements) when is_list(elements) do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:rpush, elements})

    {:ok, result}
  end

  @doc """
  Pops one or more elements from the left (head) of the list stored at `key`.

  When `count` is 1 (the default), returns a single element. When `count` is
  greater than 1, returns a list of elements. Returns `{:ok, nil}` if the key
  does not exist or the list is empty.

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["task_a", "task_b", "task_c"])
      iex> FerricStore.lpop("tasks:queue")
      {:ok, "task_a"}

      iex> FerricStore.lpop("tasks:queue", 2)
      {:ok, ["task_b", "task_c"]}

      iex> FerricStore.lpop("empty_queue")
      {:ok, nil}

  """
  @spec lpop(key(), pos_integer()) :: {:ok, binary() | [binary()] | nil}
  def lpop(key, count \\ 1) when is_integer(count) and count >= 1 do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:lpop, count})

    {:ok, result}
  end

  @doc """
  Pops one or more elements from the right (tail) of the list stored at `key`.

  When `count` is 1 (the default), returns a single element. When `count` is
  greater than 1, returns a list of elements. Returns `{:ok, nil}` if the key
  does not exist or the list is empty.

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["task_a", "task_b", "task_c"])
      iex> FerricStore.rpop("tasks:queue")
      {:ok, "task_c"}

      iex> FerricStore.rpop("tasks:queue", 2)
      {:ok, ["task_b", "task_a"]}

      iex> FerricStore.rpop("empty_queue")
      {:ok, nil}

  """
  @spec rpop(key(), pos_integer()) :: {:ok, binary() | [binary()] | nil}
  def rpop(key, count \\ 1) when is_integer(count) and count >= 1 do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:rpop, count})

    {:ok, result}
  end

  @doc """
  Returns elements from the list stored at `key` within the range `start..stop`.

  Both `start` and `stop` are zero-based, inclusive indices. Negative indices
  count from the end of the list (-1 is the last element).

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["task_a", "task_b", "task_c"])
      iex> FerricStore.lrange("tasks:queue", 0, -1)
      {:ok, ["task_a", "task_b", "task_c"]}

      iex> FerricStore.lrange("tasks:queue", 1, 1)
      {:ok, ["task_b"]}

      iex> FerricStore.lrange("nonexistent", 0, -1)
      {:ok, []}

  """
  @spec lrange(key(), integer(), integer()) :: {:ok, [binary()]}
  def lrange(key, start, stop) do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:lrange, start, stop})

    {:ok, result}
  end

  @doc """
  Returns the length of the list stored at `key`.

  Returns `{:ok, 0}` if the key does not exist.

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["task_a", "task_b", "task_c"])
      iex> FerricStore.llen("tasks:queue")
      {:ok, 3}

      iex> FerricStore.llen("nonexistent")
      {:ok, 0}

  """
  @spec llen(key()) :: {:ok, non_neg_integer()}
  def llen(key) do
    result =
      key
      |> sandbox_key()
      |> Router.list_op(:llen)

    {:ok, result}
  end

  # ---------------------------------------------------------------------------
  # Sets
  # ---------------------------------------------------------------------------

  @doc """
  Adds one or more members to the set stored at `key`.

  If the key does not exist, a new set is created. Members that already exist
  in the set are ignored. Returns the count of members actually added.

  ## Examples

      iex> FerricStore.sadd("article:42:tags", ["elixir", "rust", "database"])
      {:ok, 3}

      iex> FerricStore.sadd("article:42:tags", ["rust", "performance"])
      {:ok, 1}

  """
  @spec sadd(key(), [binary()]) :: {:ok, non_neg_integer()}
  def sadd(key, members) when is_list(members) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.Set.handle("SADD", [resolved_key | members], store)
    {:ok, result}
  end

  @doc """
  Removes one or more members from the set stored at `key`.

  Members that do not exist in the set are ignored. Returns the count of
  members actually removed.

  ## Examples

      iex> FerricStore.sadd("article:42:tags", ["elixir", "rust", "database"])
      iex> FerricStore.srem("article:42:tags", ["rust"])
      {:ok, 1}

      iex> FerricStore.srem("article:42:tags", ["nonexistent"])
      {:ok, 0}

  """
  @spec srem(key(), [binary()]) :: {:ok, non_neg_integer()}
  def srem(key, members) when is_list(members) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.Set.handle("SREM", [resolved_key | members], store)
    {:ok, result}
  end

  @doc """
  Returns all members of the set stored at `key`.

  Returns `{:ok, []}` if the key does not exist. The order of returned
  members is not guaranteed.

  ## Examples

      iex> FerricStore.sadd("article:42:tags", ["elixir", "rust"])
      iex> {:ok, members} = FerricStore.smembers("article:42:tags")
      iex> Enum.sort(members)
      ["elixir", "rust"]

      iex> FerricStore.smembers("nonexistent")
      {:ok, []}

  """
  @spec smembers(key()) :: {:ok, [binary()]}
  def smembers(key) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.Set.handle("SMEMBERS", [resolved_key], store)
    {:ok, result}
  end

  @doc """
  Checks whether `member` is a member of the set stored at `key`.

  Returns `true` if the member exists in the set, `false` otherwise.
  Returns `false` if the key does not exist.

  ## Examples

      iex> FerricStore.sadd("article:42:tags", ["elixir", "rust"])
      iex> FerricStore.sismember("article:42:tags", "elixir")
      true

      iex> FerricStore.sismember("article:42:tags", "python")
      false

      iex> FerricStore.sismember("nonexistent", "member")
      false

  """
  @spec sismember(key(), binary()) :: boolean()
  def sismember(key, member) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.Set.handle("SISMEMBER", [resolved_key, member], store)
    result == 1
  end

  @doc """
  Returns the number of members in the set stored at `key` (the set cardinality).

  Returns `{:ok, 0}` if the key does not exist.

  ## Examples

      iex> FerricStore.sadd("article:42:tags", ["elixir", "rust", "database"])
      iex> FerricStore.scard("article:42:tags")
      {:ok, 3}

      iex> FerricStore.scard("nonexistent")
      {:ok, 0}

  """
  @spec scard(key()) :: {:ok, non_neg_integer()}
  def scard(key) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.Set.handle("SCARD", [resolved_key], store)
    {:ok, result}
  end

  # ---------------------------------------------------------------------------
  # Sorted Sets
  # ---------------------------------------------------------------------------

  @doc """
  Adds members with scores to the sorted set stored at `key`.

  `score_member_pairs` is a list of `{score, member}` tuples where `score` is
  a number and `member` is a binary string. If a member already exists, its
  score is updated. Returns the count of new members added (not counting
  score updates).

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}])
      {:ok, 2}

      iex> FerricStore.zadd("leaderboard", [{150.0, "alice"}, {300.0, "charlie"}])
      {:ok, 1}

  """
  @spec zadd(key(), [{number(), binary()}]) :: {:ok, non_neg_integer()}
  def zadd(key, score_member_pairs) when is_list(score_member_pairs) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    # Build the ZADD args: key score1 member1 score2 member2 ...
    args =
      Enum.flat_map(score_member_pairs, fn {score, member} ->
        [to_string(score * 1.0), member]
      end)

    result = Ferricstore.Commands.SortedSet.handle("ZADD", [resolved_key | args], store)
    {:ok, result}
  end

  @doc """
  Returns members in the sorted set stored at `key` within the rank range `start..stop`.

  Indices are zero-based and inclusive. Negative indices count from the end
  (-1 is the last element). Members are ordered by score ascending.

  ## Options

    * `:withscores` - When `true`, returns `{member, score}` tuples instead
      of bare member strings. Defaults to `false`.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}, {300.0, "charlie"}])
      iex> FerricStore.zrange("leaderboard", 0, -1)
      {:ok, ["alice", "bob", "charlie"]}

      iex> FerricStore.zrange("leaderboard", 0, 1, withscores: true)
      {:ok, [{"alice", 100.0}, {"bob", 200.0}]}

      iex> FerricStore.zrange("nonexistent", 0, -1)
      {:ok, []}

  """
  @spec zrange(key(), integer(), integer(), zrange_opts()) :: {:ok, [binary() | {binary(), float()}]}
  def zrange(key, start, stop, opts \\ []) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    with_scores = Keyword.get(opts, :withscores, false)

    args = [resolved_key, to_string(start), to_string(stop)]
    args = if with_scores, do: args ++ ["WITHSCORES"], else: args

    result = Ferricstore.Commands.SortedSet.handle("ZRANGE", args, store)

    if with_scores and is_list(result) and result != [] do
      # ZRANGE WITHSCORES returns flat list: [member, score_str, member, score_str, ...]
      pairs =
        result
        |> Enum.chunk_every(2)
        |> Enum.map(fn [member, score_str] ->
          {score, ""} = Float.parse(score_str)
          {member, score}
        end)

      {:ok, pairs}
    else
      {:ok, result}
    end
  end

  @doc """
  Returns the score of `member` in the sorted set stored at `key`.

  Returns `{:ok, score}` if the member exists, or `{:ok, nil}` if the member
  or the key does not exist.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}])
      iex> FerricStore.zscore("leaderboard", "alice")
      {:ok, 100.0}

      iex> FerricStore.zscore("leaderboard", "unknown")
      {:ok, nil}

  """
  @spec zscore(key(), binary()) :: {:ok, float() | nil}
  def zscore(key, member) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.SortedSet.handle("ZSCORE", [resolved_key, member], store) do
      nil ->
        {:ok, nil}

      score_str when is_binary(score_str) ->
        {score, ""} = Float.parse(score_str)
        {:ok, score}
    end
  end

  @doc """
  Returns the number of members in the sorted set stored at `key`.

  Returns `{:ok, 0}` if the key does not exist.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}])
      iex> FerricStore.zcard("leaderboard")
      {:ok, 2}

      iex> FerricStore.zcard("nonexistent")
      {:ok, 0}

  """
  @spec zcard(key()) :: {:ok, non_neg_integer()}
  def zcard(key) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZCARD", [resolved_key], store)
    {:ok, result}
  end

  @doc """
  Removes one or more members from the sorted set stored at `key`.

  Members that do not exist are ignored. Returns the count of members
  actually removed.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}])
      iex> FerricStore.zrem("leaderboard", ["alice"])
      {:ok, 1}

      iex> FerricStore.zrem("leaderboard", ["nonexistent"])
      {:ok, 0}

  """
  @spec zrem(key(), [binary()]) :: {:ok, non_neg_integer()}
  def zrem(key, members) when is_list(members) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZREM", [resolved_key | members], store)
    {:ok, result}
  end

  # ---------------------------------------------------------------------------
  # Native Commands
  # ---------------------------------------------------------------------------

  @doc """
  Performs an atomic compare-and-swap (optimistic locking) on `key`.

  If the current value of `key` equals `expected`, it is atomically replaced
  with `new_value`. This is the building block for lock-free concurrent updates --
  read the current value, compute the new value, then CAS. If another writer
  changed the value in between, CAS returns `false` and you retry.

  ## Options

    * `:ttl` - Time-to-live in milliseconds for the new value. When omitted,
      the existing TTL is preserved.

  ## Returns

    * `{:ok, true}` if the swap was performed.
    * `{:ok, false}` if the current value did not match `expected` (retry needed).
    * `{:ok, nil}` if the key does not exist.

  ## Examples

      iex> FerricStore.set("inventory:item:99", "10")
      :ok
      iex> FerricStore.cas("inventory:item:99", "10", "9")
      {:ok, true}
      iex> FerricStore.cas("inventory:item:99", "10", "8")
      {:ok, false}

  """
  @spec cas(key(), binary(), binary(), cas_opts()) :: {:ok, true | false | nil}
  def cas(key, expected, new_value, opts \\ []) do
    resolved_key = sandbox_key(key)
    ttl_ms = Keyword.get(opts, :ttl)

    case Router.cas(resolved_key, expected, new_value, ttl_ms) do
      1 -> {:ok, true}
      0 -> {:ok, false}
      nil -> {:ok, nil}
    end
  end

  @doc """
  Cache-aside pattern with stampede (thundering herd) protection.

  Checks whether `key` has a cached value. If it does, returns
  `{:ok, {:hit, value}}`. If not, returns `{:ok, {:compute, hint}}` to
  indicate that the caller should compute the value and store it via
  `fetch_or_compute_result/3`.

  Only one caller at a time receives `{:compute, hint}` for a given key --
  all other concurrent callers block until the winner stores the computed
  value. This prevents N concurrent cache misses from triggering N
  identical expensive computations (the "stampede" problem).

  ## Options

    * `:ttl` (required) - TTL in milliseconds for the cached value.
    * `:hint` - An opaque string passed back in `{:compute, hint}`. Defaults
      to `""`.

  ## Returns

    * `{:ok, {:hit, value}}` if the value is cached.
    * `{:ok, {:compute, hint}}` if the caller should compute the value.
    * `{:error, reason}` on failure.

  ## Examples

      case FerricStore.fetch_or_compute("dashboard:stats:today", ttl: 30_000) do
        {:ok, {:hit, cached}} ->
          Jason.decode!(cached)

        {:ok, {:compute, _hint}} ->
          stats = DashboardService.compute_stats()
          encoded = Jason.encode!(stats)
          FerricStore.fetch_or_compute_result("dashboard:stats:today", encoded, ttl: 30_000)
          stats
      end

  """
  @spec fetch_or_compute(key(), fetch_or_compute_opts()) ::
          {:ok, {:hit, binary()} | {:compute, binary()}} | {:error, binary()}
  def fetch_or_compute(key, opts) do
    resolved_key = sandbox_key(key)
    ttl_ms = Keyword.fetch!(opts, :ttl)
    hint = Keyword.get(opts, :hint, "")

    case Ferricstore.FetchOrCompute.fetch_or_compute(resolved_key, ttl_ms, hint) do
      {:hit, value} -> {:ok, {:hit, value}}
      {:ok, value} -> {:ok, {:hit, value}}
      {:compute, compute_hint} -> {:ok, {:compute, compute_hint}}
      {:error, reason} -> {:error, reason}
    end
  end

  @doc """
  Stores the computed value for a `fetch_or_compute/2` cache miss and unblocks waiters.

  Must be called after receiving `{:ok, {:compute, hint}}` from `fetch_or_compute/2`.
  Stores the value in the cache and wakes all concurrent callers that were blocked
  waiting for the computation to complete.

  ## Options

    * `:ttl` (required) - TTL in milliseconds for the cached value.

  ## Returns

    * `:ok` on success.

  ## Examples

      iex> FerricStore.fetch_or_compute_result("dashboard:stats:today", encoded_json, ttl: 30_000)
      :ok

  """
  @spec fetch_or_compute_result(key(), binary(), keyword()) :: :ok | {:error, binary()}
  def fetch_or_compute_result(key, value, opts) do
    resolved_key = sandbox_key(key)
    ttl_ms = Keyword.fetch!(opts, :ttl)
    Ferricstore.FetchOrCompute.fetch_or_compute_result(resolved_key, value, ttl_ms)
  end

  # ---------------------------------------------------------------------------
  # Generic Key Operations
  # ---------------------------------------------------------------------------

  @doc """
  Checks whether `key` exists in the store and has not expired.

  ## Examples

      iex> FerricStore.set("user:42:name", "alice")
      :ok
      iex> FerricStore.exists("user:42:name")
      true

      iex> FerricStore.exists("nonexistent:key")
      false

  """
  @spec exists(key()) :: boolean()
  def exists(key) do
    key
    |> sandbox_key()
    |> Router.exists?()
  end

  @doc """
  Returns all keys matching `pattern` (glob-style).

  The pattern supports glob-style wildcards:

    * `*` - matches any sequence of characters
    * `?` - matches any single character

  In sandbox mode, the sandbox prefix is transparently stripped from returned
  keys, so patterns and results always use the logical key names.

  ## Examples

      iex> FerricStore.set("user:1:name", "alice")
      :ok
      iex> FerricStore.set("user:2:name", "bob")
      :ok
      iex> FerricStore.set("order:1", "pending")
      :ok
      iex> {:ok, user_keys} = FerricStore.keys("user:*")
      iex> Enum.sort(user_keys)
      ["user:1:name", "user:2:name"]

      iex> {:ok, all} = FerricStore.keys()
      iex> length(all) >= 3
      true

  """
  @spec keys(binary()) :: {:ok, [binary()]}
  def keys(pattern \\ "*") do
    alias Ferricstore.Store.CompoundKey

    namespace = Process.get(:ferricstore_sandbox)
    all_keys = Router.keys()
    match_all? = pattern == "*"

    # Compound keys store the sandbox namespace INSIDE the type prefix:
    #   "T:ns_prefix_user:42", "H:ns_prefix_user:42\0name"
    # So we must first convert raw keys to user-visible logical keys
    # (extracting from T: entries, rejecting H:/L:/S:/Z: etc.), then
    # strip the sandbox namespace from the resulting logical keys.
    visible = CompoundKey.user_visible_keys(all_keys)

    results =
      case namespace do
        %FerricStore.Sandbox{} ->
          # Struct sandbox: keys are already isolated (private shards), no prefix stripping
          if match_all? do
            visible
          else
            Enum.filter(visible, &Ferricstore.GlobMatcher.match?(&1, pattern))
          end

        ns when is_binary(ns) ->
          prefix_len = byte_size(ns)

          Enum.reduce(visible, [], fn key, acc ->
            if String.starts_with?(key, ns) do
              stripped = binary_part(key, prefix_len, byte_size(key) - prefix_len)

              if match_all? or Ferricstore.GlobMatcher.match?(stripped, pattern) do
                [stripped | acc]
              else
                acc
              end
            else
              acc
            end
          end)

        nil ->
          if match_all? do
            visible
          else
            Enum.filter(visible, &Ferricstore.GlobMatcher.match?(&1, pattern))
          end
      end

    {:ok, results}
  end

  @doc """
  Returns the total number of user-visible keys in the store.

  In sandbox mode, only keys belonging to the current sandbox namespace
  are counted. Internal compound keys (used by hashes, lists, sets, and
  sorted sets) are excluded from the count.

  ## Examples

      iex> FerricStore.set("key:a", "1")
      :ok
      iex> FerricStore.set("key:b", "2")
      :ok
      iex> {:ok, count} = FerricStore.dbsize()
      iex> count >= 2
      true

  """
  @spec dbsize() :: {:ok, non_neg_integer()}
  def dbsize do
    {:ok, matched_keys} = keys()
    {:ok, length(matched_keys)}
  end

  @doc """
  Deletes all keys from the store.

  In sandbox mode, only keys belonging to the current sandbox namespace
  are deleted.

  ## Returns

    * `:ok`

  ## Examples

      :ok = FerricStore.flushdb()

  """
  @spec flushdb() :: :ok
  def flushdb do
    sandbox = Process.get(:ferricstore_sandbox)

    case sandbox do
      %FerricStore.Sandbox{} ->
        # Struct sandbox: delete all keys via the private shards
        sc = Router.effective_shard_count()

        for i <- 0..(sc - 1) do
          shard = Router.resolve_shard(i)
          keydir = Router.resolve_keydir(i)

          raw_keys =
            try do
              :ets.foldl(fn {key, _, _, _, _, _, _}, acc -> [key | acc] end, [], keydir)
            rescue
              ArgumentError -> []
            end

          Enum.each(raw_keys, fn key ->
            try do
              GenServer.call(shard, {:delete, key}, 10_000)
            catch
              :exit, _ -> :ok
            end
          end)

          # Clear prefix index
          prefix_table = Router.resolve_prefix_table(i)
          try do :ets.delete_all_objects(prefix_table) catch :error, :badarg -> :ok end
        end

        :ok

      namespace when is_binary(namespace) ->
        # Old string namespace sandbox: flush namespace keys
        FerricStore.Sandbox.checkin(namespace)
        # Re-set the namespace since checkin clears it
        Process.put(:ferricstore_sandbox, namespace)
        :ok

      nil ->
        # No sandbox: delete ALL keys
        shard_count = :persistent_term.get(:ferricstore_shard_count, 4)

        for i <- 0..(shard_count - 1) do
          shard = Router.shard_name(i)

          raw_keys =
            try do
              :ets.foldl(fn {key, _, _, _, _, _, _}, acc -> [key | acc] end, [], :"keydir_#{i}")
            rescue
              ArgumentError -> []
            end

          Enum.each(raw_keys, fn key ->
            try do
              GenServer.call(shard, {:delete, key}, 10_000)
            catch
              :exit, _ -> :ok
            end
          end)

          try do :ets.delete_all_objects(:"prefix_keys_#{i}") catch :error, :badarg -> :ok end
        end

        :ok
    end
  end

  # ---------------------------------------------------------------------------
  # TTL
  # ---------------------------------------------------------------------------

  @doc """
  Sets a TTL (in milliseconds) on an existing key.

  The key will be automatically deleted after `ttl_ms` milliseconds have
  elapsed. Returns `{:ok, false}` if the key does not exist.

  ## Examples

      iex> FerricStore.set("session:abc", "data")
      :ok
      iex> FerricStore.expire("session:abc", :timer.minutes(30))
      {:ok, true}

      iex> FerricStore.expire("nonexistent:key", 5_000)
      {:ok, false}

  """
  @spec expire(key(), non_neg_integer()) :: {:ok, boolean()}
  def expire(key, ttl_ms) when is_integer(ttl_ms) and ttl_ms > 0 do
    resolved_key = sandbox_key(key)

    case Router.get_meta(resolved_key) do
      nil ->
        {:ok, false}

      {value, _old_exp} ->
        expire_at_ms = System.os_time(:millisecond) + ttl_ms
        Router.put(resolved_key, value, expire_at_ms)
        {:ok, true}
    end
  end

  @doc """
  Returns the remaining time-to-live in milliseconds for `key`.

  Returns `{:ok, ms}` if the key has a TTL set, or `{:ok, nil}` if the key
  has no expiry or does not exist.

  ## Examples

      iex> FerricStore.set("session:abc", "data", ttl: 60_000)
      :ok
      iex> {:ok, ms} = FerricStore.ttl("session:abc")
      iex> ms > 0 and ms <= 60_000
      true

      iex> FerricStore.set("permanent:key", "data")
      :ok
      iex> FerricStore.ttl("permanent:key")
      {:ok, nil}

      iex> FerricStore.ttl("nonexistent:key")
      {:ok, nil}

  """
  @spec ttl(key()) :: {:ok, non_neg_integer() | nil}
  def ttl(key) do
    resolved_key = sandbox_key(key)

    case Router.get_meta(resolved_key) do
      nil -> {:ok, nil}
      {_value, 0} -> {:ok, nil}
      {_value, exp} -> {:ok, max(0, exp - System.os_time(:millisecond))}
    end
  end

  # ---------------------------------------------------------------------------
  # Key Operations (copy, rename, renamenx, type, randomkey)
  # ---------------------------------------------------------------------------

  @doc """
  Copies the value (and its TTL) from `source` to `destination`.

  By default, returns an error if the destination already exists. Pass
  `:replace` to overwrite.

  ## Options

    * `:replace` - When `true`, overwrites `destination` if it already exists.

  ## Examples

      iex> FerricStore.set("user:42:name", "alice")
      :ok
      iex> FerricStore.copy("user:42:name", "user:42:name:backup")
      {:ok, true}

      iex> FerricStore.copy("user:42:name", "user:42:name:backup")
      {:error, "ERR target key already exists"}

      iex> FerricStore.copy("user:42:name", "user:42:name:backup", replace: true)
      {:ok, true}

      iex> FerricStore.copy("nonexistent", "dst")
      {:error, "ERR no such key"}

  """
  @spec copy(key(), key(), keyword()) :: {:ok, true} | {:error, binary()}
  def copy(source, destination, opts \\ []) do
    src_key = sandbox_key(source)
    dst_key = sandbox_key(destination)
    replace = Keyword.get(opts, :replace, false)

    case Router.get_meta(src_key) do
      nil ->
        {:error, "ERR no such key"}

      {value, expire_at_ms} ->
        if Router.exists?(dst_key) and not replace do
          {:error, "ERR target key already exists"}
        else
          Router.put(dst_key, value, expire_at_ms)
          {:ok, true}
        end
    end
  end

  @doc """
  Renames `source` to `destination`, overwriting `destination` if it exists.

  The value and TTL are transferred to the new key name, and the source key
  is deleted. Returns `{:error, reason}` if the source does not exist.

  ## Examples

      iex> FerricStore.set("temp:upload:abc", "file_data")
      :ok
      iex> FerricStore.rename("temp:upload:abc", "file:abc")
      :ok
      iex> FerricStore.get("file:abc")
      {:ok, "file_data"}
      iex> FerricStore.exists("temp:upload:abc")
      false

      iex> FerricStore.rename("nonexistent", "dst")
      {:error, "ERR no such key"}

  """
  @spec rename(key(), key()) :: :ok | {:error, binary()}
  def rename(source, destination) do
    src_key = sandbox_key(source)
    dst_key = sandbox_key(destination)

    case Router.get_meta(src_key) do
      nil ->
        {:error, "ERR no such key"}

      {value, expire_at_ms} ->
        Router.put(dst_key, value, expire_at_ms)
        Router.delete(src_key)
        :ok
    end
  end

  @doc """
  Renames `source` to `destination` only if `destination` does not already exist.

  Unlike `rename/2`, this will not overwrite an existing destination key.
  The value and TTL are transferred on success.

  ## Examples

      iex> FerricStore.set("temp:import:1", "data")
      :ok
      iex> FerricStore.renamenx("temp:import:1", "import:1")
      {:ok, true}

      iex> FerricStore.set("import:2", "existing")
      :ok
      iex> FerricStore.set("temp:import:2", "new_data")
      :ok
      iex> FerricStore.renamenx("temp:import:2", "import:2")
      {:ok, false}

      iex> FerricStore.renamenx("nonexistent", "dst")
      {:error, "ERR no such key"}

  """
  @spec renamenx(key(), key()) :: {:ok, boolean()} | {:error, binary()}
  def renamenx(source, destination) do
    src_key = sandbox_key(source)
    dst_key = sandbox_key(destination)

    case Router.get_meta(src_key) do
      nil ->
        {:error, "ERR no such key"}

      {value, expire_at_ms} ->
        if Router.exists?(dst_key) do
          {:ok, false}
        else
          Router.put(dst_key, value, expire_at_ms)
          Router.delete(src_key)
          {:ok, true}
        end
    end
  end

  @doc """
  Returns the data type of the value stored at `key`.

  The returned type string reflects the underlying data structure: `"string"`,
  `"hash"`, `"list"`, `"set"`, `"zset"`, `"stream"`, or `"none"` if the key
  does not exist.

  ## Examples

      iex> FerricStore.set("user:42:name", "alice")
      :ok
      iex> FerricStore.type("user:42:name")
      {:ok, "string"}

      iex> FerricStore.hset("user:42", %{"name" => "alice"})
      :ok
      iex> FerricStore.type("user:42")
      {:ok, "hash"}

      iex> FerricStore.type("nonexistent:key")
      {:ok, "none"}

  """
  @spec type(key()) :: {:ok, binary()}
  def type(key) do
    resolved_key = sandbox_key(key)

    # Check compound key type registry first (for hash/set/zset stored via compound keys)
    store = build_compound_store(resolved_key)
    type_key = Ferricstore.Store.CompoundKey.type_key(resolved_key)
    compound_type = store.compound_get.(resolved_key, type_key)

    cond do
      compound_type != nil ->
        {:ok, compound_type}

      # Check for list metadata key (lists use compound keys, no type marker)
      store.compound_get.(resolved_key, Ferricstore.Store.CompoundKey.list_meta_key(resolved_key)) != nil ->
        {:ok, "list"}

      true ->
        case Router.get(resolved_key) do
          nil ->
            {:ok, "none"}

          value when is_binary(value) ->
            detected = try do
              case :erlang.binary_to_term(value) do
                {:list, _} -> "list"
                _ -> nil
              end
            rescue
              ArgumentError -> nil
            end

            if detected do
              {:ok, detected}
            else
              {:ok, "string"}
            end

          _ ->
            {:ok, "string"}
        end
    end
  end

  @doc """
  Returns a random key from the store, or `{:ok, nil}` if the store is empty.

  In sandbox mode, only keys within the current sandbox are candidates.

  ## Examples

      iex> FerricStore.set("key:a", "1")
      :ok
      iex> {:ok, key} = FerricStore.randomkey()
      iex> is_binary(key)
      true

      iex> # When the store is empty:
      iex> FerricStore.randomkey()
      {:ok, nil}

  """
  @spec randomkey() :: {:ok, key() | nil}
  def randomkey do
    {:ok, all_keys} = keys()
    case all_keys do
      [] -> {:ok, nil}
      _ -> {:ok, Enum.random(all_keys)}
    end
  end

  # ---------------------------------------------------------------------------
  # TTL extended: persist, pexpire, pexpireat, expireat, expiretime, pexpiretime, pttl
  # ---------------------------------------------------------------------------

  @doc """
  Removes the TTL from `key`, making it persist indefinitely.

  Returns `{:ok, true}` if an expiry was removed, or `{:ok, false}` if the
  key does not exist or already has no TTL.

  ## Examples

      iex> FerricStore.set("session:abc", "data", ttl: 60_000)
      :ok
      iex> FerricStore.persist("session:abc")
      {:ok, true}
      iex> FerricStore.ttl("session:abc")
      {:ok, nil}

      iex> FerricStore.persist("permanent:key")
      {:ok, false}

      iex> FerricStore.persist("nonexistent:key")
      {:ok, false}

  """
  @spec persist(key()) :: {:ok, boolean()}
  def persist(key) do
    resolved_key = sandbox_key(key)

    case Router.get_meta(resolved_key) do
      nil ->
        {:ok, false}

      {_value, 0} ->
        {:ok, false}

      {value, _exp} ->
        Router.put(resolved_key, value, 0)
        {:ok, true}
    end
  end

  @doc """
  Sets a TTL in milliseconds on an existing key.

  This is an alias for `expire/2` -- both accept milliseconds.

  ## Examples

      iex> FerricStore.set("rate_limit:user:42", "3")
      :ok
      iex> FerricStore.pexpire("rate_limit:user:42", 30_000)
      {:ok, true}

      iex> FerricStore.pexpire("nonexistent:key", 5_000)
      {:ok, false}

  """
  @spec pexpire(key(), non_neg_integer()) :: {:ok, boolean()}
  def pexpire(key, ttl_ms), do: expire(key, ttl_ms)

  @doc """
  Sets the key to expire at the given absolute Unix timestamp (in seconds).

  The key will be automatically deleted when the system clock reaches the
  specified timestamp. Returns `{:ok, false}` if the key does not exist.

  ## Examples

      iex> FerricStore.set("event:promo", "active")
      :ok
      iex> FerricStore.expireat("event:promo", 1_700_000_000)
      {:ok, true}

      iex> FerricStore.expireat("nonexistent:key", 1_700_000_000)
      {:ok, false}

  """
  @spec expireat(key(), non_neg_integer()) :: {:ok, boolean()}
  def expireat(key, unix_ts_seconds) do
    resolved_key = sandbox_key(key)

    case Router.get_meta(resolved_key) do
      nil ->
        {:ok, false}

      {value, _old_exp} ->
        expire_at_ms = unix_ts_seconds * 1_000
        Router.put(resolved_key, value, expire_at_ms)
        {:ok, true}
    end
  end

  @doc """
  Sets the key to expire at the given absolute Unix timestamp (in milliseconds).

  Like `expireat/2` but with millisecond precision. Returns `{:ok, false}`
  if the key does not exist.

  ## Examples

      iex> FerricStore.set("event:flash_sale", "active")
      :ok
      iex> FerricStore.pexpireat("event:flash_sale", 1_700_000_000_000)
      {:ok, true}

      iex> FerricStore.pexpireat("nonexistent:key", 1_700_000_000_000)
      {:ok, false}

  """
  @spec pexpireat(key(), non_neg_integer()) :: {:ok, boolean()}
  def pexpireat(key, unix_ts_ms) do
    resolved_key = sandbox_key(key)

    case Router.get_meta(resolved_key) do
      nil ->
        {:ok, false}

      {value, _old_exp} ->
        Router.put(resolved_key, value, unix_ts_ms)
        {:ok, true}
    end
  end

  @doc """
  Returns the absolute Unix timestamp (in seconds) at which `key` will expire.

  Returns `{:ok, -1}` if the key exists but has no associated expiry, and
  `{:ok, -2}` if the key does not exist.

  ## Examples

      iex> FerricStore.set("session:abc", "data", ttl: 60_000)
      :ok
      iex> {:ok, ts} = FerricStore.expiretime("session:abc")
      iex> ts > 0
      true

      iex> FerricStore.set("permanent:key", "data")
      :ok
      iex> FerricStore.expiretime("permanent:key")
      {:ok, -1}

      iex> FerricStore.expiretime("nonexistent:key")
      {:ok, -2}

  """
  @spec expiretime(key()) :: {:ok, integer()}
  def expiretime(key) do
    resolved_key = sandbox_key(key)

    case Router.get_meta(resolved_key) do
      nil -> {:ok, -2}
      {_value, 0} -> {:ok, -1}
      {_value, exp} -> {:ok, div(exp, 1_000)}
    end
  end

  @doc """
  Returns the absolute Unix timestamp (in milliseconds) at which `key` will expire.

  Like `expiretime/1` but with millisecond precision. Returns `{:ok, -1}`
  if the key has no expiry, and `{:ok, -2}` if it does not exist.

  ## Examples

      iex> FerricStore.set("session:abc", "data", ttl: 60_000)
      :ok
      iex> {:ok, ts_ms} = FerricStore.pexpiretime("session:abc")
      iex> ts_ms > 0
      true

      iex> FerricStore.set("permanent:key", "data")
      :ok
      iex> FerricStore.pexpiretime("permanent:key")
      {:ok, -1}

      iex> FerricStore.pexpiretime("nonexistent:key")
      {:ok, -2}

  """
  @spec pexpiretime(key()) :: {:ok, integer()}
  def pexpiretime(key) do
    resolved_key = sandbox_key(key)

    case Router.get_meta(resolved_key) do
      nil -> {:ok, -2}
      {_value, 0} -> {:ok, -1}
      {_value, exp} -> {:ok, exp}
    end
  end

  @doc """
  Returns the remaining time-to-live in milliseconds for `key`.

  This is an alias for `ttl/1` -- both return millisecond precision.
  Returns `{:ok, nil}` if the key has no expiry or does not exist.

  ## Examples

      iex> FerricStore.set("cache:result", "data", ttl: 30_000)
      :ok
      iex> {:ok, ms} = FerricStore.pttl("cache:result")
      iex> ms > 0 and ms <= 30_000
      true

      iex> FerricStore.pttl("nonexistent:key")
      {:ok, nil}

  """
  @spec pttl(key()) :: {:ok, non_neg_integer() | nil}
  def pttl(key), do: ttl(key)

  # ---------------------------------------------------------------------------
  # Bitmap operations
  # ---------------------------------------------------------------------------

  @doc """
  Sets or clears the bit at `offset` in the string value stored at `key`.

  Returns the original bit value at that position.

  ## Examples

      {:ok, 0} = FerricStore.setbit("key", 7, 1)

  """
  @spec setbit(key(), non_neg_integer(), 0 | 1) :: {:ok, 0 | 1}
  def setbit(key, offset, bit_value) when bit_value in [0, 1] do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Bitmap.handle("SETBIT", [resolved_key, to_string(offset), to_string(bit_value)], store)
    wrap_result(result)
  end

  @doc """
  Returns the bit value at `offset` in the string value stored at `key`.

  Returns `{:ok, 0}` for nonexistent keys or out-of-range offsets.

  ## Examples

      {:ok, 1} = FerricStore.getbit("key", 7)

  """
  @spec getbit(key(), non_neg_integer()) :: {:ok, 0 | 1}
  def getbit(key, offset) do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Bitmap.handle("GETBIT", [resolved_key, to_string(offset)], store)
    wrap_result(result)
  end

  @doc """
  Counts the number of set bits (1s) in the string value stored at `key`.

  ## Options

    * `:start` - Start byte offset (default: 0).
    * `:stop` - Stop byte offset (default: -1, meaning end of string).

  ## Returns

    * `{:ok, count}` on success.

  ## Examples

      {:ok, count} = FerricStore.bitcount("key")

  """
  @spec bitcount(key(), keyword()) :: {:ok, non_neg_integer()}
  def bitcount(key, opts \\ []) do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    start = Keyword.get(opts, :start)
    stop = Keyword.get(opts, :stop)

    args = if start != nil and stop != nil do
      [resolved_key, to_string(start), to_string(stop)]
    else
      [resolved_key]
    end

    result = Ferricstore.Commands.Bitmap.handle("BITCOUNT", args, store)
    wrap_result(result)
  end

  @doc """
  Performs a bitwise operation between strings stored at `source_keys` and
  stores the result in `dest_key`.

  ## Parameters

    * `op` - `:and`, `:or`, `:xor`, or `:not`
    * `dest_key` - Destination key.
    * `source_keys` - List of source keys.

  ## Returns

    * `{:ok, byte_length}` - Length of the result string.

  ## Examples

      {:ok, 3} = FerricStore.bitop(:and, "dest", ["key1", "key2"])

  """
  @spec bitop(atom(), key(), [key()]) :: {:ok, non_neg_integer()}
  def bitop(op, dest_key, source_keys) when is_atom(op) and is_list(source_keys) do
    resolved_dest = sandbox_key(dest_key)
    resolved_sources = Enum.map(source_keys, &sandbox_key/1)
    store = build_string_store(resolved_dest)
    op_str = op |> Atom.to_string() |> String.upcase()
    result = Ferricstore.Commands.Bitmap.handle("BITOP", [op_str, resolved_dest | resolved_sources], store)
    wrap_result(result)
  end

  @doc """
  Finds the first bit set to `bit_value` (0 or 1) in the string at `key`.

  ## Options

    * `:start` - Start byte offset.
    * `:stop` - Stop byte offset.

  ## Returns

    * `{:ok, position}` - Bit position, or -1 if not found within a bounded range.

  ## Examples

      {:ok, 8} = FerricStore.bitpos("key", 1)

  """
  @spec bitpos(key(), 0 | 1, keyword()) :: {:ok, integer()}
  def bitpos(key, bit_value, opts \\ []) when bit_value in [0, 1] do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    start = Keyword.get(opts, :start)
    stop = Keyword.get(opts, :stop)

    args = cond do
      start != nil and stop != nil ->
        [resolved_key, to_string(bit_value), to_string(start), to_string(stop)]
      start != nil ->
        [resolved_key, to_string(bit_value), to_string(start)]
      true ->
        [resolved_key, to_string(bit_value)]
    end

    result = Ferricstore.Commands.Bitmap.handle("BITPOS", args, store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # Hash extended operations
  # ---------------------------------------------------------------------------

  @doc """
  Deletes one or more fields from the hash stored at `key`.

  Fields that do not exist are ignored. Returns the count of fields actually
  removed.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30", "email" => "a@b.c"})
      iex> FerricStore.hdel("user:42", ["age", "email"])
      {:ok, 2}

      iex> FerricStore.hdel("user:42", ["nonexistent"])
      {:ok, 0}

  """
  @spec hdel(key(), [binary()]) :: {:ok, non_neg_integer()}
  def hdel(key, fields) when is_list(fields) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    str_fields = Enum.map(fields, &to_string/1)

    case Ferricstore.Commands.Hash.handle("HDEL", [resolved_key | str_fields], store) do
      {:error, _} = err -> err
      count -> {:ok, count}
    end
  end

  @doc """
  Returns whether `field` exists in the hash stored at `key`.

  Returns `true` if the field exists, `false` otherwise. Returns `false`
  if the key itself does not exist.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice"})
      iex> FerricStore.hexists("user:42", "name")
      true

      iex> FerricStore.hexists("user:42", "missing")
      false

      iex> FerricStore.hexists("no_such_hash", "field")
      false

  """
  @spec hexists(key(), binary()) :: boolean()
  def hexists(key, field) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle("HEXISTS", [resolved_key, to_string(field)], store) do
      {:error, _} = err -> err
      1 -> true
      0 -> false
    end
  end

  @doc """
  Returns the number of fields in the hash stored at `key`.

  Returns `{:ok, 0}` if the key does not exist.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30", "email" => "a@b.c"})
      iex> FerricStore.hlen("user:42")
      {:ok, 3}

      iex> FerricStore.hlen("no_such_hash")
      {:ok, 0}

  """
  @spec hlen(key()) :: {:ok, non_neg_integer()}
  def hlen(key) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle("HLEN", [resolved_key], store) do
      {:error, _} = err -> err
      count -> {:ok, count}
    end
  end

  @doc """
  Returns all field names from the hash stored at `key`.

  Returns `{:ok, []}` if the key does not exist. The order of returned field
  names is not guaranteed.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30"})
      iex> {:ok, fields} = FerricStore.hkeys("user:42")
      iex> Enum.sort(fields)
      ["age", "name"]

      iex> FerricStore.hkeys("no_such_hash")
      {:ok, []}

  """
  @spec hkeys(key()) :: {:ok, [binary()]}
  def hkeys(key) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle("HKEYS", [resolved_key], store) do
      {:error, _} = err -> err
      keys_list -> {:ok, keys_list}
    end
  end

  @doc """
  Returns all field values from the hash stored at `key`.

  Returns `{:ok, []}` if the key does not exist. The order of returned values
  corresponds to the order of fields (not guaranteed to be insertion order).

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30"})
      iex> {:ok, vals} = FerricStore.hvals("user:42")
      iex> Enum.sort(vals)
      ["30", "alice"]

      iex> FerricStore.hvals("no_such_hash")
      {:ok, []}

  """
  @spec hvals(key()) :: {:ok, [binary()]}
  def hvals(key) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle("HVALS", [resolved_key], store) do
      {:error, _} = err -> err
      vals_list -> {:ok, vals_list}
    end
  end

  @doc """
  Returns values for the specified `fields` from the hash at `key`.

  Returns `nil` for fields that do not exist. The order of returned values
  matches the order of the requested fields.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30"})
      iex> FerricStore.hmget("user:42", ["name", "missing", "age"])
      {:ok, ["alice", nil, "30"]}

      iex> FerricStore.hmget("no_such_hash", ["a", "b"])
      {:ok, [nil, nil]}

  """
  @spec hmget(key(), [binary()]) :: {:ok, [binary() | nil]}
  def hmget(key, fields) when is_list(fields) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    str_fields = Enum.map(fields, &to_string/1)

    case Ferricstore.Commands.Hash.handle("HMGET", [resolved_key | str_fields], store) do
      {:error, _} = err -> err
      values -> {:ok, values}
    end
  end

  @doc """
  Increments the integer value of `field` in the hash at `key` by `amount`.

  If the field does not exist, it is created with `0` before incrementing.
  Returns `{:error, reason}` if the field value is not a valid integer.

  ## Examples

      iex> FerricStore.hset("user:42", %{"login_count" => "10"})
      iex> FerricStore.hincrby("user:42", "login_count", 5)
      {:ok, 15}

      iex> FerricStore.hincrby("user:42", "new_counter", 1)
      {:ok, 1}

  """
  @spec hincrby(key(), binary(), integer()) :: {:ok, integer()} | {:error, binary()}
  def hincrby(key, field, amount) when is_integer(amount) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle(
           "HINCRBY",
           [resolved_key, to_string(field), Integer.to_string(amount)],
           store
         ) do
      {:error, _} = err -> err
      new_val -> {:ok, new_val}
    end
  end

  @doc """
  Increments the float value of `field` in the hash at `key` by `amount`.

  If the field does not exist, it is created with `0` before incrementing.
  Returns the new value as a string. Returns `{:error, reason}` if the field
  value is not a valid number.

  ## Examples

      iex> FerricStore.hset("product:99", %{"price" => "10.0"})
      iex> FerricStore.hincrbyfloat("product:99", "price", 2.5)
      {:ok, "12.5"}

      iex> FerricStore.hincrbyfloat("product:99", "discount", 0.15)
      {:ok, "0.15"}

  """
  @spec hincrbyfloat(key(), binary(), float()) :: {:ok, binary()} | {:error, binary()}
  def hincrbyfloat(key, field, amount) when is_number(amount) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    amount_str = :erlang.float_to_binary(amount * 1.0, [:compact, decimals: 17])

    case Ferricstore.Commands.Hash.handle(
           "HINCRBYFLOAT",
           [resolved_key, to_string(field), amount_str],
           store
         ) do
      {:error, _} = err -> err
      result_str -> {:ok, result_str}
    end
  end

  @doc """
  Sets `field` in the hash at `key` only if the field does not already exist.

  Returns `{:ok, true}` if the field was set, `{:ok, false}` if it already
  existed.

  ## Examples

      iex> FerricStore.hsetnx("user:42", "name", "alice")
      {:ok, true}

      iex> FerricStore.hsetnx("user:42", "name", "bob")
      {:ok, false}

  """
  @spec hsetnx(key(), binary(), binary()) :: {:ok, boolean()}
  def hsetnx(key, field, value) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle(
           "HSETNX",
           [resolved_key, to_string(field), to_string(value)],
           store
         ) do
      {:error, _} = err -> err
      1 -> {:ok, true}
      0 -> {:ok, false}
    end
  end

  @doc """
  Returns one or more random field names from the hash at `key`.

  Without `count`, returns a single field name or `nil` if the hash is empty.
  With positive `count`, returns up to `count` unique fields. With negative
  `count`, returns `abs(count)` fields with possible duplicates.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice", "age" => "30", "email" => "a@b.c"})
      iex> {:ok, field} = FerricStore.hrandfield("user:42")
      iex> field in ["name", "age", "email"]
      true

      iex> {:ok, fields} = FerricStore.hrandfield("user:42", 2)
      iex> length(fields)
      2

      iex> FerricStore.hrandfield("nonexistent")
      {:ok, nil}

  """
  @spec hrandfield(key(), integer() | nil) :: {:ok, binary() | [binary()] | nil}
  def hrandfield(key, count \\ nil) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case count do
      nil ->
        case Ferricstore.Commands.Hash.handle("HRANDFIELD", [resolved_key], store) do
          {:error, _} = err -> err
          result -> {:ok, result}
        end

      n when is_integer(n) ->
        case Ferricstore.Commands.Hash.handle(
               "HRANDFIELD",
               [resolved_key, Integer.to_string(n)],
               store
             ) do
          {:error, _} = err -> err
          result -> {:ok, result}
        end
    end
  end

  @doc """
  Returns the string length of the value for `field` in the hash at `key`.

  Returns `{:ok, 0}` if the field or the key does not exist.

  ## Examples

      iex> FerricStore.hset("user:42", %{"name" => "alice"})
      iex> FerricStore.hstrlen("user:42", "name")
      {:ok, 5}

      iex> FerricStore.hstrlen("user:42", "missing")
      {:ok, 0}

  """
  @spec hstrlen(key(), binary()) :: {:ok, non_neg_integer()}
  def hstrlen(key, field) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case Ferricstore.Commands.Hash.handle("HSTRLEN", [resolved_key, to_string(field)], store) do
      {:error, _} = err -> err
      len -> {:ok, len}
    end
  end

  # ---------------------------------------------------------------------------
  # List extended operations
  # ---------------------------------------------------------------------------

  @doc """
  Returns the element at `index` in the list stored at `key`.

  Negative indices count from the end (-1 is the last element). Returns
  `{:ok, nil}` for out-of-range indices or nonexistent keys.

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["task_a", "task_b", "task_c"])
      iex> FerricStore.lindex("tasks:queue", 0)
      {:ok, "task_a"}

      iex> FerricStore.lindex("tasks:queue", -1)
      {:ok, "task_c"}

      iex> FerricStore.lindex("tasks:queue", 99)
      {:ok, nil}

  """
  @spec lindex(key(), integer()) :: {:ok, binary() | nil}
  def lindex(key, index) do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:lindex, index})

    {:ok, result}
  end

  @doc """
  Sets the element at `index` in the list stored at `key`.

  Returns `:ok` on success, or `{:error, reason}` if the index is out of range
  or the key does not exist.

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["task_a", "task_b", "task_c"])
      iex> FerricStore.lset("tasks:queue", 1, "task_b_updated")
      :ok

      iex> FerricStore.lset("tasks:queue", 99, "value")
      {:error, "ERR index out of range"}

  """
  @spec lset(key(), integer(), binary()) :: :ok | {:error, binary()}
  def lset(key, index, element) do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:lset, index, element})

    case result do
      :ok -> :ok
      {:error, _} = err -> err
    end
  end

  @doc """
  Removes occurrences of `element` from the list at `key`.

  The `count` argument controls the direction and number of removals:

    * `count > 0` - Remove up to `count` occurrences scanning from head to tail.
    * `count < 0` - Remove up to `abs(count)` occurrences scanning from tail to head.
    * `count == 0` - Remove all occurrences.

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["retry", "send", "retry", "retry"])
      iex> FerricStore.lrem("tasks:queue", 0, "retry")
      {:ok, 3}

      iex> FerricStore.rpush("tasks:queue", ["a", "b", "a"])
      iex> FerricStore.lrem("tasks:queue", 1, "a")
      {:ok, 1}

  """
  @spec lrem(key(), integer(), binary()) :: {:ok, non_neg_integer()}
  def lrem(key, count, element) do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:lrem, count, element})

    {:ok, result}
  end

  @doc """
  Inserts `element` before or after `pivot` in the list at `key`.

  Returns `{:ok, new_length}` if the pivot was found, or `{:ok, -1}` if the
  pivot was not found. Returns `{:ok, 0}` if the key does not exist.

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["task_a", "task_b", "task_c"])
      iex> FerricStore.linsert("tasks:queue", :before, "task_b", "task_new")
      {:ok, 4}

      iex> FerricStore.linsert("tasks:queue", :after, "task_c", "task_last")
      {:ok, 5}

      iex> FerricStore.linsert("tasks:queue", :before, "nonexistent", "x")
      {:ok, -1}

  """
  @spec linsert(key(), :before | :after, binary(), binary()) :: {:ok, integer()}
  def linsert(key, direction, pivot, element) when direction in [:before, :after] do
    result =
      key
      |> sandbox_key()
      |> Router.list_op({:linsert, direction, pivot, element})

    {:ok, result}
  end

  @doc """
  Atomically moves an element from one list to another.

  Pops from `from_dir` of `source` and pushes to `to_dir` of `destination`.
  Returns `{:ok, nil}` if the source list is empty or does not exist.

  ## Examples

      iex> FerricStore.rpush("inbox", ["msg_a", "msg_b"])
      iex> FerricStore.lmove("inbox", "processing", :left, :right)
      {:ok, "msg_a"}

      iex> FerricStore.lmove("empty_list", "dst", :left, :right)
      {:ok, nil}

  """
  @spec lmove(key(), key(), :left | :right, :left | :right) :: {:ok, binary() | nil}
  def lmove(source, destination, from_dir, to_dir)
      when from_dir in [:left, :right] and to_dir in [:left, :right] do
    src_key = sandbox_key(source)
    dst_key = sandbox_key(destination)
    result = Router.list_op(src_key, {:lmove, dst_key, from_dir, to_dir})
    {:ok, result}
  end

  @doc """
  Finds the position of `element` in the list at `key`.

  Returns the zero-based index of the first match, or `{:ok, nil}` if not
  found. When `:count` is specified, returns a list of indices.

  ## Options

    * `:rank` - Skip the first N-1 matches and return starting from the Nth
      (default: 1). Negative rank searches from tail.
    * `:count` - Return up to N positions. 0 means all. When given, always
      returns a list.
    * `:maxlen` - Limit scan to the first N elements (default: 0, no limit).

  ## Examples

      iex> FerricStore.rpush("tasks:queue", ["retry", "send", "retry", "process"])
      iex> FerricStore.lpos("tasks:queue", "retry")
      {:ok, 0}

      iex> FerricStore.lpos("tasks:queue", "retry", count: 0)
      {:ok, [0, 2]}

      iex> FerricStore.lpos("tasks:queue", "missing")
      {:ok, nil}

  """
  @spec lpos(key(), binary(), keyword()) :: {:ok, integer() | [integer()] | nil}
  def lpos(key, element, opts \\ []) do
    resolved_key = sandbox_key(key)
    rank = Keyword.get(opts, :rank, 1)
    count = Keyword.get(opts, :count)
    maxlen = Keyword.get(opts, :maxlen, 0)

    result = Router.list_op(resolved_key, {:lpos, element, rank, count, maxlen})

    {:ok, result}
  end

  # ---------------------------------------------------------------------------
  # Set extended operations
  # ---------------------------------------------------------------------------

  @doc """
  Returns the membership status of multiple members in the set at `key`.

  Returns a list of 1s and 0s corresponding to each member, in the same order
  as the input list.

  ## Examples

      iex> FerricStore.sadd("article:42:tags", ["elixir", "rust", "database"])
      iex> FerricStore.smismember("article:42:tags", ["elixir", "python", "database"])
      {:ok, [1, 0, 1]}

      iex> FerricStore.smismember("nonexistent", ["a", "b"])
      {:ok, [0, 0]}

  """
  @spec smismember(key(), [binary()]) :: {:ok, [0 | 1]}
  def smismember(key, members) when is_list(members) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    results = Enum.map(members, fn member ->
      compound_key = Ferricstore.Store.CompoundKey.set_member(resolved_key, member)
      if store.compound_get.(resolved_key, compound_key) != nil, do: 1, else: 0
    end)

    {:ok, results}
  end

  @doc """
  Returns one or more random members from the set at `key` without removing them.

  Without `count`, returns a single member or `nil` for empty/nonexistent sets.
  With positive `count`, returns up to `count` unique members. With negative
  `count`, returns `abs(count)` members with possible duplicates.

  ## Examples

      iex> FerricStore.sadd("article:42:tags", ["elixir", "rust", "database"])
      iex> {:ok, member} = FerricStore.srandmember("article:42:tags")
      iex> member in ["elixir", "rust", "database"]
      true

      iex> {:ok, members} = FerricStore.srandmember("article:42:tags", 2)
      iex> length(members)
      2

      iex> FerricStore.srandmember("nonexistent")
      {:ok, nil}

  """
  @spec srandmember(key(), integer() | nil) :: {:ok, binary() | [binary()] | nil}
  def srandmember(key, count \\ nil) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case count do
      nil ->
        result = Ferricstore.Commands.Set.handle("SRANDMEMBER", [resolved_key], store)
        {:ok, result}

      n ->
        result = Ferricstore.Commands.Set.handle("SRANDMEMBER", [resolved_key, to_string(n)], store)
        {:ok, result}
    end
  end

  @doc """
  Removes and returns one or more random members from the set at `key`.

  Without `count`, returns a single member or `nil` for empty/nonexistent sets.
  With `count`, returns a list of up to `count` removed members.

  ## Examples

      iex> FerricStore.sadd("article:42:tags", ["elixir", "rust", "database"])
      iex> {:ok, tag} = FerricStore.spop("article:42:tags")
      iex> tag in ["elixir", "rust", "database"]
      true

      iex> {:ok, tags} = FerricStore.spop("article:42:tags", 2)
      iex> length(tags)
      2

      iex> FerricStore.spop("nonexistent")
      {:ok, nil}

  """
  @spec spop(key(), non_neg_integer() | nil) :: {:ok, binary() | [binary()] | nil}
  def spop(key, count \\ nil) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case count do
      nil ->
        result = Ferricstore.Commands.Set.handle("SPOP", [resolved_key], store)
        {:ok, result}

      n ->
        result = Ferricstore.Commands.Set.handle("SPOP", [resolved_key, to_string(n)], store)
        {:ok, result}
    end
  end

  @doc """
  Returns the set difference: members in the first set that are not in any of the other sets.

  Handles cross-shard keys transparently. Returns `{:ok, []}` if the first
  key does not exist.

  ## Examples

      iex> FerricStore.sadd("frontend:tags", ["elixir", "react", "tailwind"])
      iex> FerricStore.sadd("backend:tags", ["elixir", "postgres"])
      iex> {:ok, diff} = FerricStore.sdiff(["frontend:tags", "backend:tags"])
      iex> Enum.sort(diff)
      ["react", "tailwind"]

  """
  @spec sdiff([key()]) :: {:ok, [binary()]}
  def sdiff(keys) when is_list(keys) do
    resolved_keys = Enum.map(keys, &sandbox_key/1)
    sets = Enum.map(resolved_keys, &gather_set_members/1)

    result =
      case sets do
        [first | rest] ->
          Enum.reduce(rest, first, fn s, acc -> MapSet.difference(acc, s) end)
        [] ->
          MapSet.new()
      end

    {:ok, MapSet.to_list(result)}
  end

  @doc """
  Returns the set intersection: members common to all given sets.

  Handles cross-shard keys transparently. Returns `{:ok, []}` if any key
  does not exist.

  ## Examples

      iex> FerricStore.sadd("frontend:tags", ["elixir", "react", "tailwind"])
      iex> FerricStore.sadd("backend:tags", ["elixir", "postgres"])
      iex> FerricStore.sinter(["frontend:tags", "backend:tags"])
      {:ok, ["elixir"]}

  """
  @spec sinter([key()]) :: {:ok, [binary()]}
  def sinter(keys) when is_list(keys) do
    resolved_keys = Enum.map(keys, &sandbox_key/1)
    sets = Enum.map(resolved_keys, &gather_set_members/1)

    result =
      case sets do
        [first | rest] ->
          Enum.reduce(rest, first, fn s, acc -> MapSet.intersection(acc, s) end)
        [] ->
          MapSet.new()
      end

    {:ok, MapSet.to_list(result)}
  end

  @doc """
  Returns the set union: all unique members across all given sets.

  Handles cross-shard keys transparently.

  ## Examples

      iex> FerricStore.sadd("frontend:tags", ["elixir", "react"])
      iex> FerricStore.sadd("backend:tags", ["elixir", "postgres"])
      iex> {:ok, union} = FerricStore.sunion(["frontend:tags", "backend:tags"])
      iex> Enum.sort(union)
      ["elixir", "postgres", "react"]

  """
  @spec sunion([key()]) :: {:ok, [binary()]}
  def sunion(keys) when is_list(keys) do
    resolved_keys = Enum.map(keys, &sandbox_key/1)
    sets = Enum.map(resolved_keys, &gather_set_members/1)
    result = Enum.reduce(sets, MapSet.new(), fn s, acc -> MapSet.union(acc, s) end)
    {:ok, MapSet.to_list(result)}
  end

  @doc """
  Computes the set difference of the given keys and stores the result in `destination`.

  Any existing value at `destination` is overwritten. Returns the number of
  elements in the resulting set.

  ## Examples

      iex> FerricStore.sadd("frontend:tags", ["elixir", "react", "tailwind"])
      iex> FerricStore.sadd("backend:tags", ["elixir", "postgres"])
      iex> FerricStore.sdiffstore("frontend_only:tags", ["frontend:tags", "backend:tags"])
      {:ok, 2}

  """
  @spec sdiffstore(key(), [key()]) :: {:ok, non_neg_integer()}
  def sdiffstore(destination, keys) when is_list(keys) do
    resolved_dst = sandbox_key(destination)
    resolved_keys = Enum.map(keys, &sandbox_key/1)
    store = build_compound_store(resolved_dst)
    result = Ferricstore.Commands.Set.handle("SDIFFSTORE", [resolved_dst | resolved_keys], store)
    wrap_result(result)
  end

  @doc """
  Computes the set intersection of the given keys and stores the result in `destination`.

  Any existing value at `destination` is overwritten. Returns the number of
  elements in the resulting set.

  ## Examples

      iex> FerricStore.sadd("frontend:tags", ["elixir", "react"])
      iex> FerricStore.sadd("backend:tags", ["elixir", "postgres"])
      iex> FerricStore.sinterstore("shared:tags", ["frontend:tags", "backend:tags"])
      {:ok, 1}

  """
  @spec sinterstore(key(), [key()]) :: {:ok, non_neg_integer()}
  def sinterstore(destination, keys) when is_list(keys) do
    resolved_dst = sandbox_key(destination)
    resolved_keys = Enum.map(keys, &sandbox_key/1)
    store = build_compound_store(resolved_dst)
    result = Ferricstore.Commands.Set.handle("SINTERSTORE", [resolved_dst | resolved_keys], store)
    wrap_result(result)
  end

  @doc """
  Computes the set union of the given keys and stores the result in `destination`.

  Any existing value at `destination` is overwritten. Returns the number of
  elements in the resulting set.

  ## Examples

      iex> FerricStore.sadd("frontend:tags", ["elixir", "react"])
      iex> FerricStore.sadd("backend:tags", ["elixir", "postgres"])
      iex> FerricStore.sunionstore("all:tags", ["frontend:tags", "backend:tags"])
      {:ok, 3}

  """
  @spec sunionstore(key(), [key()]) :: {:ok, non_neg_integer()}
  def sunionstore(destination, keys) when is_list(keys) do
    resolved_dst = sandbox_key(destination)
    resolved_keys = Enum.map(keys, &sandbox_key/1)
    store = build_compound_store(resolved_dst)
    result = Ferricstore.Commands.Set.handle("SUNIONSTORE", [resolved_dst | resolved_keys], store)
    wrap_result(result)
  end

  @doc """
  Returns the cardinality of the intersection of all given sets.

  More efficient than `sinter/1` when you only need the count, not the
  actual members.

  ## Options

    * `:limit` - Stop counting after reaching this limit (0 means no limit,
      default: 0). Useful for early termination on large sets.

  ## Examples

      iex> FerricStore.sadd("frontend:tags", ["elixir", "react", "tailwind"])
      iex> FerricStore.sadd("backend:tags", ["elixir", "postgres", "tailwind"])
      iex> FerricStore.sintercard(["frontend:tags", "backend:tags"])
      {:ok, 2}

      iex> FerricStore.sintercard(["frontend:tags", "backend:tags"], limit: 1)
      {:ok, 1}

  """
  @spec sintercard([key()], keyword()) :: {:ok, non_neg_integer()}
  def sintercard(keys, opts \\ []) when is_list(keys) do
    resolved_keys = Enum.map(keys, &sandbox_key/1)
    numkeys = Integer.to_string(length(resolved_keys))
    limit = Keyword.get(opts, :limit, 0)

    args =
      [numkeys | resolved_keys] ++
        if limit > 0, do: ["LIMIT", Integer.to_string(limit)], else: []

    store =
      case resolved_keys do
        [first | _] -> build_compound_store(first)
        [] -> build_compound_store("")
      end

    result = Ferricstore.Commands.Set.handle("SINTERCARD", args, store)
    wrap_result(result)
  end

  # Gathers all set members for a resolved key, routing to the correct shard.
  defp gather_set_members(resolved_key) do
    shard = Router.resolve_shard(Router.shard_for(resolved_key))
    prefix = Ferricstore.Store.CompoundKey.set_prefix(resolved_key)
    pairs = GenServer.call(shard, {:scan_prefix, prefix})
    MapSet.new(pairs, fn {member, _} -> member end)
  end

  # ---------------------------------------------------------------------------
  # Sorted Set extended operations
  # ---------------------------------------------------------------------------

  @doc """
  Returns the rank of `member` in the sorted set at `key` (ascending score order).

  Rank is 0-based (the member with the lowest score has rank 0). Returns
  `{:ok, nil}` if the member or key does not exist.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}, {300.0, "charlie"}])
      iex> FerricStore.zrank("leaderboard", "alice")
      {:ok, 0}

      iex> FerricStore.zrank("leaderboard", "charlie")
      {:ok, 2}

      iex> FerricStore.zrank("leaderboard", "unknown")
      {:ok, nil}

  """
  @spec zrank(key(), binary()) :: {:ok, non_neg_integer() | nil}
  def zrank(key, member) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZRANK", [resolved_key, member], store)
    wrap_result(result)
  end

  @doc """
  Returns the reverse rank of `member` in the sorted set at `key` (descending score order).

  Rank is 0-based (the member with the highest score has rank 0). Returns
  `{:ok, nil}` if the member or key does not exist.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}, {300.0, "charlie"}])
      iex> FerricStore.zrevrank("leaderboard", "charlie")
      {:ok, 0}

      iex> FerricStore.zrevrank("leaderboard", "alice")
      {:ok, 2}

      iex> FerricStore.zrevrank("leaderboard", "unknown")
      {:ok, nil}

  """
  @spec zrevrank(key(), binary()) :: {:ok, non_neg_integer() | nil}
  def zrevrank(key, member) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZREVRANK", [resolved_key, member], store)
    wrap_result(result)
  end

  @doc """
  Returns members with scores between `min` and `max` (inclusive by default).

  Use "-inf" and "+inf" for unbounded ranges. Prefix a bound with "(" for
  exclusive (e.g., "(100" means score > 100).

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}, {300.0, "charlie"}])
      iex> FerricStore.zrangebyscore("leaderboard", "100", "200")
      {:ok, ["alice", "bob"]}

      iex> FerricStore.zrangebyscore("leaderboard", "-inf", "+inf")
      {:ok, ["alice", "bob", "charlie"]}

      iex> FerricStore.zrangebyscore("leaderboard", "(200", "+inf")
      {:ok, ["charlie"]}

  """
  @spec zrangebyscore(key(), binary(), binary(), keyword()) :: {:ok, [binary()]}
  def zrangebyscore(key, min, max, opts \\ []) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZRANGEBYSCORE", [resolved_key, min, max], store)
    wrap_result(result)
  end

  @doc """
  Counts members in the sorted set at `key` with scores between `min` and `max`.

  Use "-inf" and "+inf" for unbounded ranges. Prefix a bound with "(" for
  exclusive.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}, {300.0, "charlie"}])
      iex> FerricStore.zcount("leaderboard", "100", "200")
      {:ok, 2}

      iex> FerricStore.zcount("leaderboard", "-inf", "+inf")
      {:ok, 3}

  """
  @spec zcount(key(), binary(), binary()) :: {:ok, non_neg_integer()}
  def zcount(key, min, max) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZCOUNT", [resolved_key, min, max], store)
    wrap_result(result)
  end

  @doc """
  Increments the score of `member` in the sorted set at `key` by `increment`.

  Creates the member with the given increment as score if it does not exist.
  Returns the new score as a string.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}])
      iex> FerricStore.zincrby("leaderboard", 50.0, "alice")
      {:ok, "150.0"}

      iex> FerricStore.zincrby("leaderboard", 25.0, "newcomer")
      {:ok, "25.0"}

  """
  @spec zincrby(key(), number(), binary()) :: {:ok, binary()}
  def zincrby(key, increment, member) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZINCRBY", [resolved_key, to_string(increment * 1.0), member], store)
    wrap_result(result)
  end

  @doc """
  Returns one or more random members from the sorted set at `key`.

  Without `count`, returns a single member or `nil` for empty/nonexistent keys.
  With positive `count`, returns up to `count` unique members. With negative
  `count`, returns `abs(count)` members with possible duplicates.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}, {300.0, "charlie"}])
      iex> {:ok, member} = FerricStore.zrandmember("leaderboard")
      iex> member in ["alice", "bob", "charlie"]
      true

      iex> {:ok, members} = FerricStore.zrandmember("leaderboard", 2)
      iex> length(members)
      2

      iex> FerricStore.zrandmember("nonexistent")
      {:ok, nil}

  """
  @spec zrandmember(key(), integer() | nil) :: {:ok, binary() | [binary()] | nil}
  def zrandmember(key, count \\ nil) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)

    case count do
      nil ->
        result = Ferricstore.Commands.SortedSet.handle("ZRANDMEMBER", [resolved_key], store)
        wrap_result(result)

      n ->
        result = Ferricstore.Commands.SortedSet.handle("ZRANDMEMBER", [resolved_key, to_string(n)], store)
        wrap_result(result)
    end
  end

  @doc """
  Removes and returns up to `count` members with the lowest scores.

  Returns `{:ok, []}` if the key does not exist or the sorted set is empty.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}, {300.0, "charlie"}])
      iex> FerricStore.zpopmin("leaderboard", 1)
      {:ok, [{"alice", 100.0}]}

      iex> FerricStore.zpopmin("leaderboard", 2)
      {:ok, [{"bob", 200.0}, {"charlie", 300.0}]}

  """
  @spec zpopmin(key(), pos_integer()) :: {:ok, [{binary(), float()}]}
  def zpopmin(key, count \\ 1) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZPOPMIN", [resolved_key, to_string(count)], store)

    case result do
      {:error, _} = err -> err
      flat when is_list(flat) ->
        pairs =
          flat
          |> Enum.chunk_every(2)
          |> Enum.map(fn [member, score_str] ->
            {score, _} = Float.parse(score_str)
            {member, score}
          end)
        {:ok, pairs}
    end
  end

  @doc """
  Removes and returns up to `count` members with the highest scores.

  Returns `{:ok, []}` if the key does not exist or the sorted set is empty.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}, {300.0, "charlie"}])
      iex> FerricStore.zpopmax("leaderboard", 1)
      {:ok, [{"charlie", 300.0}]}

      iex> FerricStore.zpopmax("leaderboard", 2)
      {:ok, [{"bob", 200.0}, {"alice", 100.0}]}

  """
  @spec zpopmax(key(), pos_integer()) :: {:ok, [{binary(), float()}]}
  def zpopmax(key, count \\ 1) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZPOPMAX", [resolved_key, to_string(count)], store)

    case result do
      {:error, _} = err -> err
      flat when is_list(flat) ->
        pairs =
          flat
          |> Enum.chunk_every(2)
          |> Enum.map(fn [member, score_str] ->
            {score, _} = Float.parse(score_str)
            {member, score}
          end)
        {:ok, pairs}
    end
  end

  @doc """
  Returns scores for multiple members in the sorted set at `key`.

  Returns `nil` for members that do not exist. The order of returned scores
  matches the order of the input members.

  ## Examples

      iex> FerricStore.zadd("leaderboard", [{100.0, "alice"}, {200.0, "bob"}])
      iex> FerricStore.zmscore("leaderboard", ["alice", "unknown", "bob"])
      {:ok, [100.0, nil, 200.0]}

  """
  @spec zmscore(key(), [binary()]) :: {:ok, [float() | nil]}
  def zmscore(key, members) when is_list(members) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.SortedSet.handle("ZMSCORE", [resolved_key | members], store)

    case result do
      {:error, _} = err -> err
      scores when is_list(scores) ->
        parsed = Enum.map(scores, fn
          nil -> nil
          score_str when is_binary(score_str) ->
            {score, _} = Float.parse(score_str)
            score
        end)
        {:ok, parsed}
    end
  end

  # ---------------------------------------------------------------------------
  # Streams
  # ---------------------------------------------------------------------------

  @doc """
  Appends an entry to the stream at `key` with an auto-generated ID.

  `fields` is a flat list of field-value pairs: `["field1", "val1", "field2", "val2"]`.
  Streams are append-only logs ideal for event sourcing, activity feeds, and audit trails.

  ## Returns

    * `{:ok, entry_id}` where `entry_id` is a `"timestamp-seq"` string.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.xadd("events:user:42", ["action", "login", "ip", "10.0.0.1"])
      {:ok, "1711234567890-0"}

      iex> FerricStore.xadd("activity:feed", ["type", "comment", "body", "looks great!"])
      {:ok, "1711234567891-0"}

  """
  @spec xadd(key(), [binary()]) :: {:ok, binary()} | {:error, binary()}
  def xadd(key, fields) when is_list(fields) do
    resolved_key = sandbox_key(key)
    store = build_stream_store(resolved_key)
    result = Ferricstore.Commands.Stream.handle("XADD", [resolved_key, "*" | fields], store)
    wrap_result(result)
  end

  @doc """
  Returns the number of entries in the stream at `key`.

  ## Returns

    * `{:ok, length}` on success.

  ## Examples

      iex> FerricStore.xlen("events:user:42")
      {:ok, 5}

  """
  @spec xlen(key()) :: {:ok, non_neg_integer()}
  def xlen(key) do
    resolved_key = sandbox_key(key)
    store = build_stream_store(resolved_key)
    result = Ferricstore.Commands.Stream.handle("XLEN", [resolved_key], store)
    wrap_result(result)
  end

  @doc """
  Returns entries from the stream at `key` in forward (oldest-first) order between `start` and `stop`.

  Use `"-"` for the minimum and `"+"` for the maximum stream IDs.

  ## Options

    * `:count` - Maximum number of entries to return.

  ## Returns

    * `{:ok, entries}` where entries is a list of `{id, [field, value, ...]}` tuples.

  ## Examples

      iex> FerricStore.xrange("events:user:42", "-", "+", count: 10)
      {:ok, [{"1711234567890-0", ["action", "login", "ip", "10.0.0.1"]}]}

      iex> FerricStore.xrange("activity:feed", "-", "+")
      {:ok, [{"1711234567891-0", ["type", "comment", "body", "looks great!"]}]}

  """
  @spec xrange(key(), binary(), binary(), keyword()) :: {:ok, [tuple()]}
  def xrange(key, start, stop, opts \\ []) do
    resolved_key = sandbox_key(key)
    store = build_stream_store(resolved_key)
    count = Keyword.get(opts, :count)
    args = [resolved_key, start, stop]
    args = if count, do: args ++ ["COUNT", to_string(count)], else: args
    result = Ferricstore.Commands.Stream.handle("XRANGE", args, store)
    wrap_result(result)
  end

  @doc """
  Returns entries from the stream at `key` in reverse (newest-first) order between `stop` and `start`.

  ## Options

    * `:count` - Maximum number of entries to return.

  ## Returns

    * `{:ok, entries}` where entries is a list of `{id, [field, value, ...]}` tuples.

  ## Examples

      iex> FerricStore.xrevrange("events:user:42", "+", "-", count: 5)
      {:ok, [{"1711234567890-0", ["action", "login", "ip", "10.0.0.1"]}]}

  """
  @spec xrevrange(key(), binary(), binary(), keyword()) :: {:ok, [tuple()]}
  def xrevrange(key, stop, start, opts \\ []) do
    resolved_key = sandbox_key(key)
    store = build_stream_store(resolved_key)
    count = Keyword.get(opts, :count)
    args = [resolved_key, stop, start]
    args = if count, do: args ++ ["COUNT", to_string(count)], else: args
    result = Ferricstore.Commands.Stream.handle("XREVRANGE", args, store)
    wrap_result(result)
  end

  @doc """
  Trims the stream at `key` to a maximum number of entries, evicting the oldest.

  Useful for capping event logs and activity feeds to prevent unbounded growth.

  ## Options

    * `:maxlen` (required) - Maximum number of entries to keep.

  ## Returns

    * `{:ok, trimmed_count}` - the number of entries removed.

  ## Examples

      iex> FerricStore.xtrim("events:user:42", maxlen: 1000)
      {:ok, 5}

  """
  @spec xtrim(key(), keyword()) :: {:ok, non_neg_integer()}
  def xtrim(key, opts) do
    resolved_key = sandbox_key(key)
    store = build_stream_store(resolved_key)
    maxlen = Keyword.fetch!(opts, :maxlen)
    result = Ferricstore.Commands.Stream.handle("XTRIM", [resolved_key, "MAXLEN", to_string(maxlen)], store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # Bloom Filter operations
  # ---------------------------------------------------------------------------

  @doc """
  Creates a Bloom filter with specific error rate and capacity.

  ## Examples

      :ok = FerricStore.bf_reserve("filter", 0.01, 1000)

  """
  @spec bf_reserve(key(), float(), pos_integer()) :: :ok | {:error, binary()}
  def bf_reserve(key, error_rate, capacity) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    Ferricstore.Commands.Bloom.handle("BF.RESERVE", [resolved_key, to_string(error_rate), to_string(capacity)], store)
  end

  @doc """
  Adds an element to the Bloom filter at `key`, auto-creating if needed.

  ## Returns

    * `{:ok, 1}` if the element was added.
    * `{:ok, 0}` if the element was already present.

  ## Examples

      {:ok, 1} = FerricStore.bf_add("filter", "hello")

  """
  @spec bf_add(key(), binary()) :: {:ok, 0 | 1}
  def bf_add(key, element) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Bloom.handle("BF.ADD", [resolved_key, element], store)
    wrap_result(result)
  end

  @doc """
  Adds multiple elements to the Bloom filter at `key`.

  ## Returns

    * `{:ok, [0 | 1, ...]}` for each element.

  """
  @spec bf_madd(key(), [binary()]) :: {:ok, [0 | 1]}
  def bf_madd(key, elements) when is_list(elements) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Bloom.handle("BF.MADD", [resolved_key | elements], store)
    wrap_result(result)
  end

  @doc """
  Checks if an element may exist in the Bloom filter at `key`.

  ## Returns

    * `{:ok, 1}` if the element may exist.
    * `{:ok, 0}` if the element definitely does not exist.

  """
  @spec bf_exists(key(), binary()) :: {:ok, 0 | 1}
  def bf_exists(key, element) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Bloom.handle("BF.EXISTS", [resolved_key, element], store)
    wrap_result(result)
  end

  @doc """
  Checks if multiple elements may exist in the Bloom filter at `key`.

  ## Returns

    * `{:ok, [0 | 1, ...]}` for each element.

  """
  @spec bf_mexists(key(), [binary()]) :: {:ok, [0 | 1]}
  def bf_mexists(key, elements) when is_list(elements) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Bloom.handle("BF.MEXISTS", [resolved_key | elements], store)
    wrap_result(result)
  end

  @doc """
  Returns the approximate number of unique elements added to the Bloom filter at `key`.

  ## Returns

    * `{:ok, count}` on success.

  ## Examples

      iex> FerricStore.bf_card("emails:seen")
      {:ok, 42}

  """
  @spec bf_card(key()) :: {:ok, non_neg_integer()}
  def bf_card(key) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Bloom.handle("BF.CARD", [resolved_key], store)
    wrap_result(result)
  end

  @doc """
  Returns metadata about the Bloom filter at `key` (capacity, error rate, size, etc.).

  ## Returns

    * `{:ok, info_list}` - flat key-value list of filter properties.
    * `{:error, reason}` if the filter does not exist.

  ## Examples

      iex> FerricStore.bf_info("emails:seen")
      {:ok, ["Capacity", 100000, "Size", 120048, "Number of filters", 1, "Number of items inserted", 42, "Expansion rate", 2]}

  """
  @spec bf_info(key()) :: {:ok, list()} | {:error, binary()}
  def bf_info(key) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Bloom.handle("BF.INFO", [resolved_key], store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # Cuckoo Filter operations
  # ---------------------------------------------------------------------------

  @doc """
  Creates a Cuckoo filter with the specified capacity.

  A Cuckoo filter is similar to a Bloom filter but supports deletion and counting.
  Use it when you need probabilistic membership checks with the ability to remove
  items later (e.g., tracking active sessions that can be revoked).

  ## Parameters

    * `key` - the Cuckoo filter key
    * `capacity` - expected number of elements

  ## Examples

      iex> FerricStore.cf_reserve("sessions:active", 50_000)
      :ok

  """
  @spec cf_reserve(key(), pos_integer()) :: :ok | {:error, binary()}
  def cf_reserve(key, capacity) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    Ferricstore.Commands.Cuckoo.handle("CF.RESERVE", [resolved_key, to_string(capacity)], store)
  end

  @doc """
  Adds an element to the Cuckoo filter at `key`, auto-creating if needed.

  Unlike Bloom filters, duplicate insertions increase the count for the element.

  ## Returns

    * `{:ok, 1}` on success.
    * `{:error, reason}` if the filter is full and cannot accommodate the element.

  ## Examples

      iex> FerricStore.cf_add("sessions:active", "sess_abc123")
      {:ok, 1}

  """
  @spec cf_add(key(), binary()) :: {:ok, 0 | 1} | {:error, binary()}
  def cf_add(key, element) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Cuckoo.handle("CF.ADD", [resolved_key, element], store)
    wrap_result(result)
  end

  @doc """
  Adds an element to the Cuckoo filter only if it is not already present.

  ## Returns

    * `{:ok, 1}` if the element was newly added.
    * `{:ok, 0}` if the element already exists.
    * `{:error, reason}` if the filter is full.

  ## Examples

      iex> FerricStore.cf_addnx("sessions:active", "sess_abc123")
      {:ok, 1}

      iex> FerricStore.cf_addnx("sessions:active", "sess_abc123")
      {:ok, 0}

  """
  @spec cf_addnx(key(), binary()) :: {:ok, 0 | 1} | {:error, binary()}
  def cf_addnx(key, element) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Cuckoo.handle("CF.ADDNX", [resolved_key, element], store)
    wrap_result(result)
  end

  @doc """
  Deletes one occurrence of an element from the Cuckoo filter at `key`.

  This is the key advantage of Cuckoo filters over Bloom filters -- elements
  can be removed. Only deletes one occurrence if the element was added multiple times.

  ## Returns

    * `{:ok, 1}` if the element was deleted.
    * `{:ok, 0}` if the element was not found.

  ## Examples

      iex> FerricStore.cf_del("sessions:active", "sess_abc123")
      {:ok, 1}

  """
  @spec cf_del(key(), binary()) :: {:ok, 0 | 1}
  def cf_del(key, element) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Cuckoo.handle("CF.DEL", [resolved_key, element], store)
    wrap_result(result)
  end

  @doc """
  Checks if an element may exist in the Cuckoo filter at `key`.

  ## Returns

    * `{:ok, 1}` if the element probably exists.
    * `{:ok, 0}` if the element definitely does not exist.

  ## Examples

      iex> FerricStore.cf_exists("sessions:active", "sess_abc123")
      {:ok, 1}

  """
  @spec cf_exists(key(), binary()) :: {:ok, 0 | 1}
  def cf_exists(key, element) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Cuckoo.handle("CF.EXISTS", [resolved_key, element], store)
    wrap_result(result)
  end

  @doc """
  Checks multiple elements against the Cuckoo filter at `key` in a single call.

  ## Returns

    * `{:ok, [0 | 1, ...]}` - `1` for probably present, `0` for definitely absent,
      one per element.

  ## Examples

      iex> FerricStore.cf_mexists("sessions:active", ["sess_abc123", "sess_unknown"])
      {:ok, [1, 0]}

  """
  @spec cf_mexists(key(), [binary()]) :: {:ok, [0 | 1]}
  def cf_mexists(key, elements) when is_list(elements) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Cuckoo.handle("CF.MEXISTS", [resolved_key | elements], store)
    wrap_result(result)
  end

  @doc """
  Returns the approximate number of times an element was added to the Cuckoo filter.

  ## Returns

    * `{:ok, count}` - estimated insertion count for the element.

  ## Examples

      iex> FerricStore.cf_count("sessions:active", "sess_abc123")
      {:ok, 1}

  """
  @spec cf_count(key(), binary()) :: {:ok, non_neg_integer()}
  def cf_count(key, element) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Cuckoo.handle("CF.COUNT", [resolved_key, element], store)
    wrap_result(result)
  end

  @doc """
  Returns metadata about the Cuckoo filter at `key` (size, bucket count, etc.).

  ## Returns

    * `{:ok, info_list}` - flat key-value list of filter properties.
    * `{:error, reason}` if the filter does not exist.

  ## Examples

      iex> FerricStore.cf_info("sessions:active")
      {:ok, ["Size", 1024, "Number of buckets", 512, "Number of filters", 1, "Number of items inserted", 3, "Number of items deleted", 0, "Bucket size", 2, "Expansion rate", 1, "Max iterations", 20]}

  """
  @spec cf_info(key()) :: {:ok, list()} | {:error, binary()}
  def cf_info(key) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.Cuckoo.handle("CF.INFO", [resolved_key], store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # Count-Min Sketch operations
  # ---------------------------------------------------------------------------

  @doc """
  Creates a Count-Min Sketch with the given `width` and `depth` dimensions.

  A Count-Min Sketch is a probabilistic structure for approximate frequency counting.
  It uses sub-linear space and answers "how many times has X been seen?" with bounded
  over-estimation. Ideal for view counts, click tracking, and frequency analysis
  where exact counts are not required.

  ## Parameters

    * `key` - the CMS key
    * `width` - number of counters per hash function (larger = more accurate)
    * `depth` - number of hash functions (larger = lower error probability)

  ## Examples

      iex> FerricStore.cms_initbydim("page:views", 2000, 5)
      :ok

  """
  @spec cms_initbydim(key(), pos_integer(), pos_integer()) :: :ok | {:error, binary()}
  def cms_initbydim(key, width, depth) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    Ferricstore.Commands.CMS.handle("CMS.INITBYDIM", [resolved_key, to_string(width), to_string(depth)], store)
  end

  @doc """
  Creates a Count-Min Sketch with a target error rate and over-estimation probability.

  The sketch dimensions (width/depth) are computed automatically from the error bounds.

  ## Parameters

    * `key` - the CMS key
    * `error` - acceptable error rate as a fraction (e.g. `0.001` for 0.1%)
    * `probability` - probability of exceeding the error rate (e.g. `0.01` for 1%)

  ## Examples

      iex> FerricStore.cms_initbyprob("click:tracking", 0.001, 0.01)
      :ok

  """
  @spec cms_initbyprob(key(), float(), float()) :: :ok | {:error, binary()}
  def cms_initbyprob(key, error, probability) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    Ferricstore.Commands.CMS.handle("CMS.INITBYPROB", [resolved_key, to_string(error), to_string(probability)], store)
  end

  @doc """
  Increments the count for one or more elements in the Count-Min Sketch.

  ## Parameters

    * `key` - the CMS key
    * `pairs` - list of `{element, increment}` tuples

  ## Returns

    * `{:ok, [new_count, ...]}` - estimated count after increment, one per element.
    * `{:error, reason}` if the sketch does not exist.

  ## Examples

      iex> FerricStore.cms_incrby("page:views", [{"homepage", 1}, {"about", 3}])
      {:ok, [1, 3]}

  """
  @spec cms_incrby(key(), [{binary(), pos_integer()}]) :: {:ok, [non_neg_integer()]} | {:error, binary()}
  def cms_incrby(key, pairs) when is_list(pairs) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    args = Enum.flat_map(pairs, fn {element, count} -> [element, to_string(count)] end)
    result = Ferricstore.Commands.CMS.handle("CMS.INCRBY", [resolved_key | args], store)
    wrap_result(result)
  end

  @doc """
  Queries the estimated frequency count for one or more elements in the Count-Min Sketch.

  Counts may be over-estimated but never under-estimated.

  ## Returns

    * `{:ok, [count, ...]}` - estimated count per element.
    * `{:error, reason}` if the sketch does not exist.

  ## Examples

      iex> FerricStore.cms_query("page:views", ["homepage", "about", "unknown"])
      {:ok, [42, 7, 0]}

  """
  @spec cms_query(key(), [binary()]) :: {:ok, [non_neg_integer()]} | {:error, binary()}
  def cms_query(key, elements) when is_list(elements) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.CMS.handle("CMS.QUERY", [resolved_key | elements], store)
    wrap_result(result)
  end

  @doc """
  Returns metadata about the Count-Min Sketch at `key` (width, depth, total count).

  ## Returns

    * `{:ok, info_list}` - flat key-value list of sketch properties.
    * `{:error, reason}` if the sketch does not exist.

  ## Examples

      iex> FerricStore.cms_info("page:views")
      {:ok, ["width", 2000, "depth", 5, "count", 49]}

  """
  @spec cms_info(key()) :: {:ok, list()} | {:error, binary()}
  def cms_info(key) do
    resolved_key = sandbox_key(key)
    store = build_prob_store(resolved_key)
    result = Ferricstore.Commands.CMS.handle("CMS.INFO", [resolved_key], store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # TopK operations
  # ---------------------------------------------------------------------------

  @doc """
  Creates a Top-K tracker that maintains the `k` most frequent elements.

  A Top-K structure uses the Heavy Keeper algorithm to efficiently track the
  most popular items in a data stream with bounded memory. Ideal for trending
  topics, popular search queries, and hot product tracking.

  ## Parameters

    * `key` - the Top-K tracker key
    * `k` - number of top elements to track

  ## Examples

      iex> FerricStore.topk_reserve("trending:searches", 10)
      :ok

  """
  @spec topk_reserve(key(), pos_integer()) :: :ok | {:error, binary()}
  def topk_reserve(key, k) do
    resolved_key = sandbox_key(key)
    store = build_topk_store(resolved_key)
    Ferricstore.Commands.TopK.handle("TOPK.RESERVE", [resolved_key, to_string(k)], store)
  end

  @doc """
  Adds one or more elements to the Top-K tracker, updating frequency counts.

  If an element displaces another from the top-k, the displaced element is returned.

  ## Returns

    * `{:ok, [displaced | nil, ...]}` - `nil` if no element was displaced,
      or the name of the displaced element, one per input.

  ## Examples

      iex> FerricStore.topk_add("trending:searches", ["elixir", "rust", "golang"])
      {:ok, [nil, nil, nil]}

  """
  @spec topk_add(key(), [binary()]) :: {:ok, list()} | {:error, binary()}
  def topk_add(key, elements) when is_list(elements) do
    resolved_key = sandbox_key(key)
    store = build_topk_store(resolved_key)
    result = Ferricstore.Commands.TopK.handle("TOPK.ADD", [resolved_key | elements], store)
    wrap_result(result)
  end

  @doc """
  Checks whether elements are currently in the Top-K set.

  ## Returns

    * `{:ok, [0 | 1, ...]}` - `1` if the element is in the top-k, `0` otherwise.

  ## Examples

      iex> FerricStore.topk_query("trending:searches", ["elixir", "obscure-lang"])
      {:ok, [1, 0]}

  """
  @spec topk_query(key(), [binary()]) :: {:ok, list()} | {:error, binary()}
  def topk_query(key, elements) when is_list(elements) do
    resolved_key = sandbox_key(key)
    store = build_topk_store(resolved_key)
    result = Ferricstore.Commands.TopK.handle("TOPK.QUERY", [resolved_key | elements], store)
    wrap_result(result)
  end

  @doc """
  Returns the current Top-K elements, ordered by estimated frequency (descending).

  ## Returns

    * `{:ok, [element, ...]}` - the top-k element names.
    * `{:error, reason}` if the tracker does not exist.

  ## Examples

      iex> FerricStore.topk_list("trending:searches")
      {:ok, ["elixir", "rust", "golang"]}

  """
  @spec topk_list(key()) :: {:ok, [binary()]} | {:error, binary()}
  def topk_list(key) do
    resolved_key = sandbox_key(key)
    store = build_topk_store(resolved_key)
    result = Ferricstore.Commands.TopK.handle("TOPK.LIST", [resolved_key], store)
    wrap_result(result)
  end

  @doc """
  Returns metadata about the Top-K tracker at `key` (k, width, depth, decay).

  ## Returns

    * `{:ok, info_list}` - flat key-value list of tracker properties.
    * `{:error, reason}` if the tracker does not exist.

  ## Examples

      iex> FerricStore.topk_info("trending:searches")
      {:ok, ["k", 10, "width", 8, "depth", 7, "decay", "0.9"]}

  """
  @spec topk_info(key()) :: {:ok, list()} | {:error, binary()}
  def topk_info(key) do
    resolved_key = sandbox_key(key)
    store = build_topk_store(resolved_key)
    result = Ferricstore.Commands.TopK.handle("TOPK.INFO", [resolved_key], store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # T-Digest operations
  # ---------------------------------------------------------------------------

  @doc """
  Creates a T-Digest structure at `key` for estimating quantiles and percentiles.

  A T-Digest compactly summarizes a distribution of numeric values, enabling
  accurate estimation of percentiles (p50, p95, p99) with bounded memory.
  Ideal for latency monitoring, response time analysis, and SLA tracking.

  ## Examples

      iex> FerricStore.tdigest_create("api:latency:ms")
      :ok

  """
  @spec tdigest_create(key()) :: :ok | {:error, binary()}
  def tdigest_create(key) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    Ferricstore.Commands.TDigest.handle("TDIGEST.CREATE", [resolved_key], store)
  end

  @doc """
  Adds one or more numeric observations to the T-Digest at `key`.

  ## Parameters

    * `key` - the T-Digest key
    * `values` - list of numeric values to add

  ## Examples

      iex> FerricStore.tdigest_add("api:latency:ms", [12.5, 45.0, 3.2, 89.1, 150.0])
      :ok

  """
  @spec tdigest_add(key(), [number()]) :: :ok | {:error, binary()}
  def tdigest_add(key, values) when is_list(values) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    value_strs = Enum.map(values, &to_string(&1 * 1.0))
    Ferricstore.Commands.TDigest.handle("TDIGEST.ADD", [resolved_key | value_strs], store)
  end

  @doc """
  Estimates the values at the given quantile points (0.0 to 1.0).

  For example, quantile `0.5` is the median, `0.95` is the 95th percentile.

  ## Returns

    * `{:ok, [value, ...]}` - estimated value at each quantile.
    * `{:error, reason}` if the digest does not exist.

  ## Examples

      iex> FerricStore.tdigest_quantile("api:latency:ms", [0.5, 0.95, 0.99])
      {:ok, ["45.0", "150.0", "150.0"]}

  """
  @spec tdigest_quantile(key(), [float()]) :: {:ok, list()} | {:error, binary()}
  def tdigest_quantile(key, quantiles) when is_list(quantiles) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    qs = Enum.map(quantiles, &to_string(&1 * 1.0))
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.QUANTILE", [resolved_key | qs], store)
    wrap_result(result)
  end

  @doc """
  Estimates the cumulative distribution function (CDF) at the given values.

  Returns the fraction of observations less than or equal to each value.
  For example, a CDF of `0.95` at value `100` means 95% of observations
  were <= 100.

  ## Returns

    * `{:ok, [fraction, ...]}` - CDF value (0.0 to 1.0) at each input.

  ## Examples

      iex> FerricStore.tdigest_cdf("api:latency:ms", [50.0, 100.0])
      {:ok, ["0.6", "0.8"]}

  """
  @spec tdigest_cdf(key(), [number()]) :: {:ok, list()} | {:error, binary()}
  def tdigest_cdf(key, values) when is_list(values) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    vs = Enum.map(values, &to_string(&1 * 1.0))
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.CDF", [resolved_key | vs], store)
    wrap_result(result)
  end

  @doc """
  Returns the minimum value observed in the T-Digest at `key`.

  ## Examples

      iex> FerricStore.tdigest_min("api:latency:ms")
      {:ok, "3.2"}

  """
  @spec tdigest_min(key()) :: {:ok, binary()} | {:error, binary()}
  def tdigest_min(key) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.MIN", [resolved_key], store)
    wrap_result(result)
  end

  @doc """
  Returns the maximum value observed in the T-Digest at `key`.

  ## Examples

      iex> FerricStore.tdigest_max("api:latency:ms")
      {:ok, "150.0"}

  """
  @spec tdigest_max(key()) :: {:ok, binary()} | {:error, binary()}
  def tdigest_max(key) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.MAX", [resolved_key], store)
    wrap_result(result)
  end

  @doc """
  Returns metadata about the T-Digest at `key` (compression, total observations, etc.).

  ## Returns

    * `{:ok, info_list}` - flat key-value list of digest properties.
    * `{:error, reason}` if the digest does not exist.

  ## Examples

      iex> FerricStore.tdigest_info("api:latency:ms")
      {:ok, ["Compression", 100, "Capacity", 610, "Merged nodes", 5, "Unmerged nodes", 0, "Merged weight", "5.0", "Unmerged weight", "0.0", "Total compressions", 1]}

  """
  @spec tdigest_info(key()) :: {:ok, list()} | {:error, binary()}
  def tdigest_info(key) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.INFO", [resolved_key], store)
    wrap_result(result)
  end

  @doc """
  Resets the T-Digest at `key`, discarding all observations.

  ## Examples

      iex> FerricStore.tdigest_reset("api:latency:ms")
      :ok

  """
  @spec tdigest_reset(key()) :: :ok | {:error, binary()}
  def tdigest_reset(key) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    Ferricstore.Commands.TDigest.handle("TDIGEST.RESET", [resolved_key], store)
  end

  @doc """
  Computes the trimmed mean of values between quantile bounds `lo` and `hi`.

  A trimmed mean excludes outliers by only averaging values within the specified
  quantile range. For example, `tdigest_trimmed_mean(key, 0.1, 0.9)` averages
  the middle 80% of the distribution.

  ## Parameters

    * `key` - the T-Digest key
    * `lo` - lower quantile bound (0.0 to 1.0)
    * `hi` - upper quantile bound (0.0 to 1.0)

  ## Examples

      iex> FerricStore.tdigest_trimmed_mean("api:latency:ms", 0.1, 0.9)
      {:ok, "45.5"}

  """
  @spec tdigest_trimmed_mean(key(), float(), float()) :: {:ok, binary()} | {:error, binary()}
  def tdigest_trimmed_mean(key, lo, hi) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.TRIMMED_MEAN", [resolved_key, to_string(lo * 1.0), to_string(hi * 1.0)], store)
    wrap_result(result)
  end

  @doc """
  Estimates the rank (number of observations less than or equal to) for each value.

  ## Returns

    * `{:ok, [rank, ...]}` - estimated rank per value.

  ## Examples

      iex> FerricStore.tdigest_rank("api:latency:ms", [50.0, 100.0])
      {:ok, [3, 4]}

  """
  @spec tdigest_rank(key(), [number()]) :: {:ok, list()} | {:error, binary()}
  def tdigest_rank(key, values) when is_list(values) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    vs = Enum.map(values, &to_string(&1 * 1.0))
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.RANK", [resolved_key | vs], store)
    wrap_result(result)
  end

  @doc """
  Estimates the reverse rank (number of observations greater than) for each value.

  ## Returns

    * `{:ok, [reverse_rank, ...]}` - estimated reverse rank per value.

  ## Examples

      iex> FerricStore.tdigest_revrank("api:latency:ms", [50.0, 100.0])
      {:ok, [2, 1]}

  """
  @spec tdigest_revrank(key(), [number()]) :: {:ok, list()} | {:error, binary()}
  def tdigest_revrank(key, values) when is_list(values) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    vs = Enum.map(values, &to_string(&1 * 1.0))
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.REVRANK", [resolved_key | vs], store)
    wrap_result(result)
  end

  @doc """
  Estimates the value at each given rank (0-based position in sorted order).

  ## Returns

    * `{:ok, [value, ...]}` - estimated value at each rank.

  ## Examples

      iex> FerricStore.tdigest_byrank("api:latency:ms", [0, 2, 4])
      {:ok, ["3.2", "45.0", "150.0"]}

  """
  @spec tdigest_byrank(key(), [integer()]) :: {:ok, list()} | {:error, binary()}
  def tdigest_byrank(key, ranks) when is_list(ranks) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    rs = Enum.map(ranks, &to_string/1)
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.BYRANK", [resolved_key | rs], store)
    wrap_result(result)
  end

  @doc """
  Estimates the value at each given reverse rank (0 = largest, 1 = second largest, etc.).

  ## Returns

    * `{:ok, [value, ...]}` - estimated value at each reverse rank.

  ## Examples

      iex> FerricStore.tdigest_byrevrank("api:latency:ms", [0, 1])
      {:ok, ["150.0", "89.1"]}

  """
  @spec tdigest_byrevrank(key(), [integer()]) :: {:ok, list()} | {:error, binary()}
  def tdigest_byrevrank(key, ranks) when is_list(ranks) do
    resolved_key = sandbox_key(key)
    store = build_tdigest_store(resolved_key)
    rs = Enum.map(ranks, &to_string/1)
    result = Ferricstore.Commands.TDigest.handle("TDIGEST.BYREVRANK", [resolved_key | rs], store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # Vector operations
  # ---------------------------------------------------------------------------

  @doc """
  Creates a vector collection for storing and searching high-dimensional embeddings.

  Vector collections enable similarity search (k-nearest neighbors) for
  recommendation engines, semantic search, and content deduplication.

  ## Parameters

    * `collection` - name of the vector collection
    * `dims` - number of dimensions per vector (must match all inserted vectors)
    * `metric` - distance metric: `:cosine`, `:euclidean`, or `:ip` (inner product)

  ## Examples

      iex> FerricStore.vcreate("product:embeddings", 128, :cosine)
      :ok

  """
  @spec vcreate(binary(), pos_integer(), atom()) :: :ok | {:error, binary()}
  def vcreate(collection, dims, metric) do
    resolved = sandbox_key(collection)
    store = build_compound_store(resolved)
    metric_str = Atom.to_string(metric)
    result = Ferricstore.Commands.Vector.handle("VCREATE", [resolved, to_string(dims), metric_str], store)
    case result do
      :ok -> :ok
      {:error, _} = err -> err
    end
  end

  @doc """
  Adds or updates a named vector in the collection.

  ## Parameters

    * `collection` - the vector collection name
    * `key` - unique identifier for this vector
    * `components` - list of float components (length must match collection dims)

  ## Examples

      iex> FerricStore.vadd("product:embeddings", "prod_42", [0.1, 0.8, -0.3, 0.5])
      :ok

  """
  @spec vadd(binary(), binary(), [float()]) :: :ok | {:error, binary()}
  def vadd(collection, key, components) when is_list(components) do
    resolved = sandbox_key(collection)
    store = build_compound_store(resolved)
    comp_strs = Enum.map(components, &to_string(&1 * 1.0))
    result = Ferricstore.Commands.Vector.handle("VADD", [resolved, key | comp_strs], store)
    case result do
      :ok -> :ok
      {:error, _} = err -> err
    end
  end

  @doc """
  Deletes a vector from the collection by its key.

  ## Returns

    * `{:ok, 1}` if the vector was deleted.
    * `{:ok, 0}` if the vector did not exist.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.vdel("product:embeddings", "prod_42")
      {:ok, 1}

  """
  @spec vdel(binary(), binary()) :: {:ok, 0 | 1} | {:error, binary()}
  def vdel(collection, key) do
    resolved = sandbox_key(collection)
    store = build_compound_store(resolved)
    result = Ferricstore.Commands.Vector.handle("VDEL", [resolved, key], store)
    wrap_result(result)
  end

  @doc """
  Finds the `k` nearest neighbors to `query_vector` in the collection.

  Returns results ordered by similarity (closest first), using the distance
  metric configured when the collection was created.

  ## Parameters

    * `collection` - the vector collection name
    * `query_vector` - list of float components to search for
    * `k` - number of nearest neighbors to return

  ## Returns

    * `{:ok, results}` - list of `[key, distance]` pairs, closest first.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.vsearch("product:embeddings", [0.1, 0.7, -0.2, 0.6], 3)
      {:ok, [["prod_42", "0.05"], ["prod_99", "0.12"], ["prod_7", "0.34"]]}

  """
  @spec vsearch(binary(), [float()], pos_integer()) :: {:ok, list()} | {:error, binary()}
  def vsearch(collection, query_vector, k) when is_list(query_vector) do
    resolved = sandbox_key(collection)
    store = build_compound_store(resolved)
    comp_strs = Enum.map(query_vector, &to_string(&1 * 1.0))
    # VSEARCH format: collection f32... TOP k
    args = [resolved | comp_strs] ++ ["TOP", to_string(k)]
    result = Ferricstore.Commands.Vector.handle("VSEARCH", args, store)
    wrap_result(result)
  end

  @doc """
  Returns metadata about the vector collection (dimensions, metric, vector count).

  ## Returns

    * `{:ok, info_list}` - flat key-value list of collection properties.
    * `{:error, reason}` if the collection does not exist.

  ## Examples

      iex> FerricStore.vinfo("product:embeddings")
      {:ok, ["dimensions", 128, "metric", "cosine", "vector_count", 42]}

  """
  @spec vinfo(binary()) :: {:ok, list()} | {:error, binary()}
  def vinfo(collection) do
    resolved = sandbox_key(collection)
    store = build_compound_store(resolved)
    result = Ferricstore.Commands.Vector.handle("VINFO", [resolved], store)
    wrap_result(result)
  end

  @doc """
  Returns the number of vectors stored in the collection.

  ## Returns

    * `{:ok, count}` on success.
    * `{:error, reason}` if the collection does not exist.

  ## Examples

      iex> FerricStore.vcard("product:embeddings")
      {:ok, 42}

  """
  @spec vcard(binary()) :: {:ok, non_neg_integer()} | {:error, binary()}
  def vcard(collection) do
    resolved = sandbox_key(collection)
    store = build_compound_store(resolved)
    result = Ferricstore.Commands.Vector.handle("VINFO", [resolved], store)
    case result do
      {:error, _} = err -> err
      info when is_list(info) ->
        count_idx = Enum.find_index(info, &(&1 == "vector_count"))
        if count_idx, do: {:ok, Enum.at(info, count_idx + 1)}, else: {:ok, 0}
    end
  end

  # ---------------------------------------------------------------------------
  # Geo operations
  # ---------------------------------------------------------------------------

  @doc """
  Adds geospatial members (longitude, latitude, name) to the geo index at `key`.

  Members are stored in a sorted set using geohash-encoded scores,
  enabling radius queries and distance calculations for location-based features.

  ## Parameters

    * `key` - the geo index key
    * `members` - list of `{longitude, latitude, name}` tuples

  ## Returns

    * `{:ok, added_count}` - number of new members added.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.geoadd("stores:nyc", [
      ...>   {-73.935242, 40.730610, "brooklyn_store"},
      ...>   {-74.0060, 40.7128, "manhattan_store"}
      ...> ])
      {:ok, 2}

  """
  @spec geoadd(key(), [{number(), number(), binary()}]) :: {:ok, non_neg_integer()} | {:error, binary()}
  def geoadd(key, members) when is_list(members) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    args = Enum.flat_map(members, fn {lng, lat, member} ->
      [to_string(lng * 1.0), to_string(lat * 1.0), member]
    end)
    result = Ferricstore.Commands.Geo.handle("GEOADD", [resolved_key | args], store)
    wrap_result(result)
  end

  @doc """
  Returns the distance between two geo members.

  ## Parameters

    * `key` - the geo index key
    * `member1` - first member name
    * `member2` - second member name
    * `unit` - distance unit: `"m"` (meters, default), `"km"`, `"mi"`, or `"ft"`

  ## Returns

    * `{:ok, distance_string}` on success.
    * `{:ok, nil}` if either member does not exist.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.geodist("stores:nyc", "brooklyn_store", "manhattan_store", "km")
      {:ok, "8.4567"}

  """
  @spec geodist(key(), binary(), binary(), binary()) :: {:ok, binary()} | {:error, binary()}
  def geodist(key, member1, member2, unit \\ "m") do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.Geo.handle("GEODIST", [resolved_key, member1, member2, unit], store)
    wrap_result(result)
  end

  @doc """
  Returns geohash strings for the specified members.

  Geohashes are base-32 encoded strings representing a geographic area, useful
  for proximity grouping and prefix-based spatial queries.

  ## Returns

    * `{:ok, [geohash | nil, ...]}` - a geohash per member, or `nil` for missing members.

  ## Examples

      iex> FerricStore.geohash("stores:nyc", ["brooklyn_store", "manhattan_store"])
      {:ok, ["dr5regy3zc0", "dr5regw3pp0"]}

  """
  @spec geohash(key(), [binary()]) :: {:ok, list()} | {:error, binary()}
  def geohash(key, members) when is_list(members) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.Geo.handle("GEOHASH", [resolved_key | members], store)
    wrap_result(result)
  end

  @doc """
  Returns the longitude/latitude positions for the specified members.

  ## Returns

    * `{:ok, [[longitude, latitude] | nil, ...]}` - coordinates per member,
      or `nil` for missing members.

  ## Examples

      iex> FerricStore.geopos("stores:nyc", ["brooklyn_store"])
      {:ok, [["-73.935242", "40.730610"]]}

  """
  @spec geopos(key(), [binary()]) :: {:ok, list()} | {:error, binary()}
  def geopos(key, members) when is_list(members) do
    resolved_key = sandbox_key(key)
    store = build_compound_store(resolved_key)
    result = Ferricstore.Commands.Geo.handle("GEOPOS", [resolved_key | members], store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # JSON operations
  # ---------------------------------------------------------------------------

  @doc """
  Sets a JSON value at `path` in the document stored at `key`.

  Creates the document if it does not exist (when path is `"$"`). Uses
  JSONPath syntax for nested access. Ideal for storing user preferences,
  feature flags, and nested configuration.

  ## Parameters

    * `key` - the document key
    * `path` - JSONPath expression (e.g. `"$"`, `"$.settings.theme"`)
    * `value` - JSON-encoded string to store

  ## Returns

    * `:ok` on success.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.json_set("user:42:prefs", "$", ~s({"theme":"dark","lang":"en"}))
      :ok

      iex> FerricStore.json_set("user:42:prefs", "$.theme", ~s("light"))
      :ok

  """
  @spec json_set(key(), binary(), binary()) :: :ok | {:error, binary()}
  def json_set(key, path, value) do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.SET", [resolved_key, path, value], store)
    case result do
      :ok -> :ok
      {:error, _} = err -> err
    end
  end

  @doc """
  Gets the JSON value at `path` from the document stored at `key`.

  ## Parameters

    * `key` - the document key
    * `path` - JSONPath expression (default: `"$"` for the root)

  ## Returns

    * `{:ok, json_string}` on success.
    * `{:ok, nil}` if the key does not exist.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.json_get("user:42:prefs", "$.theme")
      {:ok, "[\"dark\"]"}

      iex> FerricStore.json_get("user:42:prefs")
      {:ok, "[{\"theme\":\"dark\",\"lang\":\"en\"}]"}

  """
  @spec json_get(key(), binary()) :: {:ok, binary()} | {:error, binary()}
  def json_get(key, path \\ "$") do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.GET", [resolved_key, path], store)
    wrap_result(result)
  end

  @doc """
  Deletes the value at `path` from the JSON document at `key`.

  When path is `"$"`, the entire document is deleted.

  ## Returns

    * `{:ok, deleted_count}` - number of paths deleted.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.json_del("user:42:prefs", "$.theme")
      {:ok, 1}

  """
  @spec json_del(key(), binary()) :: {:ok, term()} | {:error, binary()}
  def json_del(key, path \\ "$") do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.DEL", [resolved_key, path], store)
    wrap_result(result)
  end

  @doc """
  Returns the JSON type of the value at `path` in the document at `key`.

  ## Returns

    * `{:ok, type}` where type is one of `"object"`, `"array"`, `"string"`,
      `"number"`, `"boolean"`, `"null"`.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.json_type("user:42:prefs", "$.theme")
      {:ok, ["string"]}

  """
  @spec json_type(key(), binary()) :: {:ok, binary()} | {:error, binary()}
  def json_type(key, path \\ "$") do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.TYPE", [resolved_key, path], store)
    wrap_result(result)
  end

  @doc """
  Atomically increments a numeric value at `path` in the JSON document at `key`.

  ## Parameters

    * `key` - the document key
    * `path` - JSONPath to a numeric value
    * `increment` - the increment amount as a string (e.g. `"1"`, `"0.5"`)

  ## Returns

    * `{:ok, new_value_string}` on success.
    * `{:error, reason}` if the path is not a number.

  ## Examples

      iex> FerricStore.json_numincrby("config:app", "$.retry_count", "1")
      {:ok, "[4]"}

  """
  @spec json_numincrby(key(), binary(), binary()) :: {:ok, binary()} | {:error, binary()}
  def json_numincrby(key, path, increment) do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.NUMINCRBY", [resolved_key, path, increment], store)
    wrap_result(result)
  end

  @doc """
  Appends one or more JSON values to the array at `path` in the document at `key`.

  ## Parameters

    * `key` - the document key
    * `path` - JSONPath to an array
    * `values` - list of JSON-encoded strings to append

  ## Returns

    * `{:ok, new_array_length}` on success.
    * `{:error, reason}` if the path is not an array.

  ## Examples

      iex> FerricStore.json_arrappend("user:42:prefs", "$.tags", [~s("vip"), ~s("beta")])
      {:ok, [4]}

  """
  @spec json_arrappend(key(), binary(), [binary()]) :: {:ok, term()} | {:error, binary()}
  def json_arrappend(key, path, values) when is_list(values) do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.ARRAPPEND", [resolved_key, path | values], store)
    wrap_result(result)
  end

  @doc """
  Returns the length of the JSON array at `path` in the document at `key`.

  ## Examples

      iex> FerricStore.json_arrlen("user:42:prefs", "$.tags")
      {:ok, [4]}

  """
  @spec json_arrlen(key(), binary()) :: {:ok, integer()} | {:error, binary()}
  def json_arrlen(key, path \\ "$") do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.ARRLEN", [resolved_key, path], store)
    wrap_result(result)
  end

  @doc """
  Returns the length of the JSON string at `path` in the document at `key`.

  ## Examples

      iex> FerricStore.json_strlen("user:42:prefs", "$.theme")
      {:ok, [4]}

  """
  @spec json_strlen(key(), binary()) :: {:ok, integer()} | {:error, binary()}
  def json_strlen(key, path \\ "$") do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.STRLEN", [resolved_key, path], store)
    wrap_result(result)
  end

  @doc """
  Returns the keys of the JSON object at `path` in the document at `key`.

  ## Examples

      iex> FerricStore.json_objkeys("user:42:prefs")
      {:ok, [["theme", "lang"]]}

  """
  @spec json_objkeys(key(), binary()) :: {:ok, list()} | {:error, binary()}
  def json_objkeys(key, path \\ "$") do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.OBJKEYS", [resolved_key, path], store)
    wrap_result(result)
  end

  @doc """
  Returns the number of keys in the JSON object at `path` in the document at `key`.

  ## Examples

      iex> FerricStore.json_objlen("user:42:prefs")
      {:ok, [2]}

  """
  @spec json_objlen(key(), binary()) :: {:ok, integer()} | {:error, binary()}
  def json_objlen(key, path \\ "$") do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.Json.handle("JSON.OBJLEN", [resolved_key, path], store)
    wrap_result(result)
  end

  # ---------------------------------------------------------------------------
  # Native: lock, unlock, extend, ratelimit_add
  # ---------------------------------------------------------------------------

  @doc """
  Acquires a distributed mutex lock on `key` with the given `owner` identity and TTL.

  Only one owner can hold a lock at a time. If the lock is already held by a
  different owner, returns an error. Use `unlock/2` to release and `extend/3`
  to renew the TTL before expiry.

  ## Parameters

    * `key` - the lock key (e.g. `"lock:order:123"`)
    * `owner` - unique owner identifier (e.g. a UUID or node name)
    * `ttl_ms` - lock duration in milliseconds (auto-expires as a safety net)

  ## Returns

    * `:ok` if the lock was acquired.
    * `{:error, reason}` if the lock is held by another owner.

  ## Examples

      iex> FerricStore.lock("lock:order:123", "worker_abc", 30_000)
      :ok

      iex> FerricStore.lock("lock:order:123", "worker_xyz", 30_000)
      {:error, "ERR lock is held by another owner"}

  """
  @spec lock(key(), binary(), pos_integer()) :: :ok | {:error, binary()}
  def lock(key, owner, ttl_ms) do
    resolved_key = sandbox_key(key)
    case Router.lock(resolved_key, owner, ttl_ms) do
      :ok -> :ok
      {:error, _} = err -> err
    end
  end

  @doc """
  Releases the lock on `key`, but only if it is currently held by `owner`.

  This ensures that a lock holder cannot accidentally release someone else's lock
  (e.g. after a timeout and re-acquisition by another process).

  ## Returns

    * `{:ok, 1}` if the lock was released.
    * `{:error, reason}` if the lock is not held by `owner`.

  ## Examples

      iex> FerricStore.unlock("lock:order:123", "worker_abc")
      {:ok, 1}

  """
  @spec unlock(key(), binary()) :: {:ok, 1} | {:error, binary()}
  def unlock(key, owner) do
    resolved_key = sandbox_key(key)
    case Router.unlock(resolved_key, owner) do
      1 -> {:ok, 1}
      {:error, _} = err -> err
    end
  end

  @doc """
  Extends the TTL of a lock on `key`, but only if it is currently held by `owner`.

  Call this periodically to prevent lock expiry while a long-running operation
  is still in progress.

  ## Returns

    * `{:ok, 1}` if the TTL was extended.
    * `{:error, reason}` if the lock is not held by `owner`.

  ## Examples

      iex> FerricStore.extend("lock:order:123", "worker_abc", 30_000)
      {:ok, 1}

  """
  @spec extend(key(), binary(), pos_integer()) :: {:ok, 1} | {:error, binary()}
  def extend(key, owner, ttl_ms) do
    resolved_key = sandbox_key(key)
    case Router.extend(resolved_key, owner, ttl_ms) do
      1 -> {:ok, 1}
      {:error, _} = err -> err
    end
  end

  @doc """
  Records `count` events against the sliding-window rate limiter at `key`.

  Uses a sliding window algorithm to track request counts within a time window.
  Returns the current count and whether the limit has been exceeded. Ideal for
  API rate limiting, abuse prevention, and throttling.

  ## Parameters

    * `key` - the rate limit key (e.g. `"ratelimit:api:user:42"`)
    * `window_ms` - sliding window duration in milliseconds
    * `max` - maximum allowed events within the window
    * `count` - number of events to record (default: 1)

  ## Returns

    * `{:ok, [allowed, current_count]}` where `allowed` is `1` (allowed) or `0`
      (rate limit exceeded), and `current_count` is the total events in the window.

  ## Examples

      iex> FerricStore.ratelimit_add("ratelimit:api:user:42", 60_000, 100)
      {:ok, [1, 1]}

      iex> FerricStore.ratelimit_add("ratelimit:api:user:42", 60_000, 100, 5)
      {:ok, [1, 6]}

  """
  @spec ratelimit_add(key(), pos_integer(), pos_integer(), pos_integer()) :: {:ok, list()}
  def ratelimit_add(key, window_ms, max, count \\ 1) do
    resolved_key = sandbox_key(key)
    result = Router.ratelimit_add(resolved_key, window_ms, max, count)
    {:ok, result}
  end

  # ---------------------------------------------------------------------------
  # HyperLogLog operations
  # ---------------------------------------------------------------------------

  @doc """
  Adds elements to the HyperLogLog at `key` for approximate cardinality counting.

  A HyperLogLog uses ~12KB of memory to estimate the number of unique elements
  in a set with a standard error of 0.81%. Ideal for counting unique visitors,
  distinct IPs, or unique events without storing every value.

  ## Returns

    * `{:ok, true}` if the internal registers were modified (new unique element likely).
    * `{:ok, false}` if the registers were not modified.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.pfadd("visitors:2024-03-28", ["user_1", "user_2", "user_3"])
      {:ok, true}

  """
  @spec pfadd(key(), [binary()]) :: {:ok, boolean()} | {:error, binary()}
  def pfadd(key, elements) when is_list(elements) do
    resolved_key = sandbox_key(key)
    store = build_string_store(resolved_key)
    result = Ferricstore.Commands.HyperLogLog.handle("PFADD", [resolved_key | elements], store)
    case result do
      1 -> {:ok, true}
      0 -> {:ok, false}
      {:error, _} = err -> err
    end
  end

  @doc """
  Returns the approximate number of unique elements across one or more HyperLogLogs.

  When given multiple keys, computes the cardinality of their union without
  modifying the underlying structures.

  ## Returns

    * `{:ok, count}` - estimated unique element count.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.pfcount(["visitors:2024-03-28"])
      {:ok, 3}

      iex> FerricStore.pfcount(["visitors:2024-03-27", "visitors:2024-03-28"])
      {:ok, 5}

  """
  @spec pfcount([key()]) :: {:ok, non_neg_integer()} | {:error, binary()}
  def pfcount(keys) when is_list(keys) do
    resolved_keys = Enum.map(keys, &sandbox_key/1)
    store = build_string_store(hd(resolved_keys))
    result = Ferricstore.Commands.HyperLogLog.handle("PFCOUNT", resolved_keys, store)
    wrap_result(result)
  end

  @doc """
  Merges multiple HyperLogLog keys into `dest_key`.

  The resulting HyperLogLog approximates the cardinality of the union of all
  source sets. Useful for computing weekly/monthly unique counts from daily ones.

  ## Returns

    * `:ok` on success.
    * `{:error, reason}` on failure.

  ## Examples

      iex> FerricStore.pfmerge("visitors:2024-w13", ["visitors:2024-03-25", "visitors:2024-03-26", "visitors:2024-03-27"])
      :ok

  """
  @spec pfmerge(key(), [key()]) :: :ok | {:error, binary()}
  def pfmerge(dest_key, source_keys) when is_list(source_keys) do
    resolved_dest = sandbox_key(dest_key)
    resolved_sources = Enum.map(source_keys, &sandbox_key/1)
    store = build_string_store(resolved_dest)
    result = Ferricstore.Commands.HyperLogLog.handle("PFMERGE", [resolved_dest | resolved_sources], store)
    case result do
      :ok -> :ok
      {:error, _} = err -> err
    end
  end

  # ---------------------------------------------------------------------------
  # Multi/Tx
  # ---------------------------------------------------------------------------

  @doc """
  Executes a sequence of commands atomically as a transaction.

  The provided function receives a `FerricStore.Tx` accumulator and should
  pipe commands into it. All commands execute in order and results are returned.

  ## Examples

      {:ok, [:ok, {:ok, "v1"}]} = FerricStore.multi(fn tx ->
        tx
        |> FerricStore.Tx.set("k1", "v1")
        |> FerricStore.Tx.get("k1")
      end)

  """
  @spec multi((FerricStore.Tx.t() -> FerricStore.Tx.t())) :: {:ok, [term()]} | {:error, binary()}
  def multi(fun) when is_function(fun, 1) do
    tx = fun.(FerricStore.Tx.new())

    case FerricStore.Tx.execute(tx) do
      {:error, _} = err -> err
      results when is_list(results) -> {:ok, results}
    end
  end

  # ---------------------------------------------------------------------------
  # Server: ping, echo, flushall
  # ---------------------------------------------------------------------------

  @doc """
  Health check that returns `{:ok, "PONG"}`.

  ## Examples

      iex> FerricStore.ping()
      {:ok, "PONG"}

  """
  @spec ping() :: {:ok, binary()}
  def ping, do: {:ok, "PONG"}

  @doc """
  Echoes back the given message, useful for connection testing.

  ## Examples

      iex> FerricStore.echo("hello")
      {:ok, "hello"}

  """
  @spec echo(binary()) :: {:ok, binary()}
  def echo(message) when is_binary(message), do: {:ok, message}

  @doc """
  Deletes all keys from the store. In sandbox mode, only sandbox keys are deleted.

  Alias for `flushdb/0`.

  ## Examples

      iex> FerricStore.flushall()
      :ok

  """
  @spec flushall() :: :ok
  def flushall, do: flushdb()

  # ---------------------------------------------------------------------------
  # Pipeline
  # ---------------------------------------------------------------------------

  @doc """
  Batches multiple commands into a single group-commit entry.

  The provided function receives a `FerricStore.Pipe` accumulator and should
  pipe commands into it. All commands are executed atomically on completion.

  ## Examples

      results = FerricStore.pipeline(fn pipe ->
        pipe
        |> FerricStore.Pipe.set("key1", "val1")
        |> FerricStore.Pipe.set("key2", "val2")
        |> FerricStore.Pipe.incr("counter")
      end)

  ## Returns

    * `{:ok, results}` - a list of results for each piped command, in order.

  """
  @spec pipeline((FerricStore.Pipe.t() -> FerricStore.Pipe.t())) :: {:ok, [term()]}
  def pipeline(fun) when is_function(fun, 1) do
    pipe = fun.(FerricStore.Pipe.new())
    results = FerricStore.Pipe.execute(pipe)
    {:ok, results}
  end

  # ---------------------------------------------------------------------------
  # Private — sandbox key prefixing
  # ---------------------------------------------------------------------------

  @sandbox_enabled Application.compile_env(:ferricstore, :sandbox_enabled, false)

  @doc false
  @spec sandbox_key(binary()) :: binary()

  if @sandbox_enabled do
    def sandbox_key(key) do
      case Process.get(:ferricstore_sandbox) do
        nil -> key
        # New struct sandbox: no key prefixing needed -- Router routes to private shards
        %FerricStore.Sandbox{} -> key
        # Old string namespace sandbox: prefix the key
        namespace when is_binary(namespace) -> namespace <> key
      end
    end
  else
    def sandbox_key(key), do: key
  end

  # ---------------------------------------------------------------------------
  # Private — integer parsing for INCR
  # ---------------------------------------------------------------------------

  defp parse_int_value(nil), do: {:ok, 0}

  defp parse_int_value(bin) when is_binary(bin) do
    case Integer.parse(bin) do
      {int_val, ""} -> {:ok, int_val}
      _ -> {:error, "ERR value is not an integer or out of range"}
    end
  end

  # ---------------------------------------------------------------------------
  # Private — result wrapping helper
  # ---------------------------------------------------------------------------

  defp wrap_result({:error, _} = err), do: err
  defp wrap_result(result), do: {:ok, result}

  # ---------------------------------------------------------------------------
  # Private — string store builder for bitmap/json/hyperloglog operations
  # ---------------------------------------------------------------------------

  defp build_string_store(resolved_key) do
    %{
      get: &Router.get/1,
      get_meta: &Router.get_meta/1,
      put: &Router.put/3,
      delete: &Router.delete/1,
      exists?: &Router.exists?/1,
      keys: &Router.keys/0,
      incr: &Router.incr/2,
      incr_float: &Router.incr_float/2,
      append: &Router.append/2,
      getset: &Router.getset/2,
      getdel: &Router.getdel/1,
      getex: &Router.getex/2,
      setrange: &Router.setrange/3,
      compound_get: fn _redis_key, compound_key ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:get, compound_key})
      end,
      compound_put: fn _redis_key, compound_key, value, expire_at_ms ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:put, compound_key, value, expire_at_ms})
      end,
      compound_delete: fn _redis_key, compound_key ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:delete, compound_key})
      end,
      compound_scan: fn _redis_key, prefix ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:scan_prefix, prefix})
      end,
      compound_count: fn _redis_key, prefix ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:count_prefix, prefix})
      end,
      compound_delete_prefix: fn _redis_key, prefix ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:delete_prefix, prefix})
      end
    }
  end

  # ---------------------------------------------------------------------------
  # Private — stream store builder
  # ---------------------------------------------------------------------------

  defp build_stream_store(resolved_key) do
    build_string_store(resolved_key)
  end

  # ---------------------------------------------------------------------------
  # Private — probabilistic structure store builder
  # ---------------------------------------------------------------------------

  defp build_prob_store(resolved_key) do
    # Probabilistic structures (Bloom, Cuckoo, CMS) use per-shard ETS registry
    # tables. The command handlers resolve the shard index from the key via
    # Router.shard_for/1, so we just need the basic store callbacks.
    # Ensure that the per-shard registry ETS tables exist.
    index = Router.shard_for(resolved_key)
    ensure_prob_registry_tables(index)

    %{
      get: &Router.get/1,
      get_meta: &Router.get_meta/1,
      put: &Router.put/3,
      delete: &Router.delete/1,
      exists?: &Router.exists?/1,
      keys: &Router.keys/0
    }
  end

  defp ensure_prob_registry_tables(index) do
    # Ensure Cuckoo registry table exists
    cuckoo_name = :"cuckoo_reg_#{index}"
    case :ets.whereis(cuckoo_name) do
      :undefined ->
        try do
          :ets.new(cuckoo_name, [:set, :public, :named_table, {:read_concurrency, true}])
        rescue
          ArgumentError -> :ok
        end
      _ -> :ok
    end

    # Ensure CMS registry table exists
    cms_name = :"cms_reg_#{index}"
    case :ets.whereis(cms_name) do
      :undefined ->
        try do
          :ets.new(cms_name, [:set, :public, :named_table, {:read_concurrency, true}])
        rescue
          ArgumentError -> :ok
        end
      _ -> :ok
    end

    :ok
  end

  # ---------------------------------------------------------------------------
  # Private — TopK store builder
  # ---------------------------------------------------------------------------

  defp build_topk_store(resolved_key) do
    data_dir = Application.get_env(:ferricstore, :data_dir, "data")
    index = Router.shard_for(resolved_key)
    prob_dir = Path.join([data_dir, "prob", "shard_#{index}"])

    %{
      get: fn key ->
        case Router.get(key) do
          nil -> nil
          bin when is_binary(bin) ->
            try do
              :erlang.binary_to_term(bin)
            rescue
              ArgumentError -> bin
            end
        end
      end,
      put: fn key, val, exp ->
        encoded = if is_tuple(val), do: :erlang.term_to_binary(val), else: val
        Router.put(key, encoded, exp)
      end,
      delete: &Router.delete/1,
      exists?: &Router.exists?/1,
      keys: &Router.keys/0,
      prob_dir: fn -> File.mkdir_p!(prob_dir); prob_dir end
    }
  end

  # ---------------------------------------------------------------------------
  # Private — TDigest store builder
  # ---------------------------------------------------------------------------

  defp build_tdigest_store(_resolved_key) do
    %{
      get: fn key ->
        case Router.get(key) do
          nil -> nil
          bin when is_binary(bin) ->
            try do
              case :erlang.binary_to_term(bin) do
                {:tdigest, _, _} = tuple -> tuple
                _ -> bin
              end
            rescue
              ArgumentError -> bin
            end
        end
      end,
      put: fn key, val, exp ->
        encoded = if is_tuple(val) and tuple_size(val) >= 1 and elem(val, 0) == :tdigest do
          :erlang.term_to_binary(val)
        else
          val
        end
        Router.put(key, encoded, exp)
      end,
      delete: &Router.delete/1,
      exists?: fn key ->
        Router.get(key) != nil
      end,
      keys: &Router.keys/0
    }
  end

  # ---------------------------------------------------------------------------
  # Private — compound key store builder for set/sorted-set operations
  # ---------------------------------------------------------------------------

  # Builds the store map expected by Commands.Set and Commands.SortedSet.
  # The store maps compound key operations to the correct shard GenServer
  # using the Redis key for routing (all sub-keys for one Redis key live
  # on the same shard).
  defp build_compound_store(resolved_key) do
    %{
      get: fn k -> Router.get(k) end,
      get_meta: fn k -> Router.get_meta(k) end,
      put: fn k, v, exp -> Router.put(k, v, exp) end,
      delete: fn k -> Router.delete(k) end,
      exists?: fn k -> Router.exists?(k) end,
      keys: &Router.keys/0,
      compound_get: fn _redis_key, compound_key ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:get, compound_key})
      end,
      compound_put: fn _redis_key, compound_key, value, expire_at_ms ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:put, compound_key, value, expire_at_ms})
      end,
      compound_delete: fn _redis_key, compound_key ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:delete, compound_key})
      end,
      compound_scan: fn _redis_key, prefix ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:scan_prefix, prefix})
      end,
      compound_count: fn _redis_key, prefix ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:count_prefix, prefix})
      end,
      compound_delete_prefix: fn _redis_key, prefix ->
        shard = Router.resolve_shard(Router.shard_for(resolved_key))
        GenServer.call(shard, {:delete_prefix, prefix})
      end
    }
  end

end

defmodule FerricStore.Pipe do
  @moduledoc """
  Pipeline accumulator for batching multiple FerricStore commands.

  Used with `FerricStore.pipeline/1` to batch multiple operations into a single
  Raft entry per shard. Commands are accumulated in reverse order and on execute,
  converted to RESP tuples and dispatched through the Coordinator. Single-shard
  pipelines commit in one Raft round-trip; cross-shard pipelines use the
  anchor-shard mechanism.

  Results are normalized to match the FerricStore public API format (e.g.
  `{:ok, value}` for GET, `:ok` for DEL) rather than raw Dispatcher values.

  ## Usage

      FerricStore.pipeline(fn pipe ->
        pipe
        |> FerricStore.Pipe.set("key1", "val1")
        |> FerricStore.Pipe.set("key2", "val2")
        |> FerricStore.Pipe.incr("counter")
      end)

  """

  @type command ::
          {:set, binary(), binary(), keyword()}
          | {:get, binary()}
          | {:del, binary()}
          | {:incr, binary()}
          | {:incr_by, binary(), integer()}
          | {:hset, binary(), map()}
          | {:hget, binary(), binary()}
          | {:lpush, binary(), [binary()]}
          | {:rpush, binary(), [binary()]}
          | {:sadd, binary(), [binary()]}
          | {:zadd, binary(), [{number(), binary()}]}
          | {:expire, binary(), non_neg_integer()}

  @type t :: %__MODULE__{commands: [command()]}

  defstruct commands: []

  @doc "Creates a new empty pipeline."
  @spec new() :: t()
  def new, do: %__MODULE__{}

  @doc "Adds a SET command to the pipeline."
  @spec set(t(), binary(), binary(), keyword()) :: t()
  def set(%__MODULE__{} = pipe, key, value, opts \\ []) do
    %{pipe | commands: [{:set, key, value, opts} | pipe.commands]}
  end

  @doc "Adds a GET command to the pipeline."
  @spec get(t(), binary()) :: t()
  def get(%__MODULE__{} = pipe, key) do
    %{pipe | commands: [{:get, key} | pipe.commands]}
  end

  @doc "Adds a DEL command to the pipeline."
  @spec del(t(), binary()) :: t()
  def del(%__MODULE__{} = pipe, key) do
    %{pipe | commands: [{:del, key} | pipe.commands]}
  end

  @doc "Adds an INCR command to the pipeline."
  @spec incr(t(), binary()) :: t()
  def incr(%__MODULE__{} = pipe, key) do
    %{pipe | commands: [{:incr, key} | pipe.commands]}
  end

  @doc "Adds an INCRBY command to the pipeline."
  @spec incr_by(t(), binary(), integer()) :: t()
  def incr_by(%__MODULE__{} = pipe, key, amount) do
    %{pipe | commands: [{:incr_by, key, amount} | pipe.commands]}
  end

  @doc "Adds an HSET command to the pipeline."
  @spec hset(t(), binary(), map()) :: t()
  def hset(%__MODULE__{} = pipe, key, fields) do
    %{pipe | commands: [{:hset, key, fields} | pipe.commands]}
  end

  @doc "Adds an HGET command to the pipeline."
  @spec hget(t(), binary(), binary()) :: t()
  def hget(%__MODULE__{} = pipe, key, field) do
    %{pipe | commands: [{:hget, key, field} | pipe.commands]}
  end

  @doc "Adds an LPUSH command to the pipeline."
  @spec lpush(t(), binary(), [binary()]) :: t()
  def lpush(%__MODULE__{} = pipe, key, elements) do
    %{pipe | commands: [{:lpush, key, elements} | pipe.commands]}
  end

  @doc "Adds an RPUSH command to the pipeline."
  @spec rpush(t(), binary(), [binary()]) :: t()
  def rpush(%__MODULE__{} = pipe, key, elements) do
    %{pipe | commands: [{:rpush, key, elements} | pipe.commands]}
  end

  @doc "Adds a SADD command to the pipeline."
  @spec sadd(t(), binary(), [binary()]) :: t()
  def sadd(%__MODULE__{} = pipe, key, members) do
    %{pipe | commands: [{:sadd, key, members} | pipe.commands]}
  end

  @doc "Adds a ZADD command to the pipeline."
  @spec zadd(t(), binary(), [{number(), binary()}]) :: t()
  def zadd(%__MODULE__{} = pipe, key, score_member_pairs) do
    %{pipe | commands: [{:zadd, key, score_member_pairs} | pipe.commands]}
  end

  @doc "Adds an EXPIRE command to the pipeline."
  @spec expire(t(), binary(), non_neg_integer()) :: t()
  def expire(%__MODULE__{} = pipe, key, ttl_ms) do
    %{pipe | commands: [{:expire, key, ttl_ms} | pipe.commands]}
  end

  @doc """
  Executes all accumulated pipeline commands as a single batch Raft entry
  per shard via the Coordinator.

  This is called internally by `FerricStore.pipeline/1`. Commands are converted
  to RESP-style tuples and dispatched through `Ferricstore.Transaction.Coordinator`,
  which groups them by shard and submits each group as a single `{:batch}` or
  `{:tx_execute}` Raft entry. Single-shard pipelines commit in one Raft round-trip;
  cross-shard pipelines use the anchor-shard mechanism.

  Results are returned in the original command order.
  """
  @spec execute(t()) :: [term()]
  def execute(%__MODULE__{commands: []}), do: []

  def execute(%__MODULE__{commands: commands}) do
    ordered = Enum.reverse(commands)

    queue = Enum.map(ordered, &to_resp_command/1)

    # For the old string sandbox, pass the namespace so keys are prefixed.
    # For the new struct sandbox, pass nil -- keys are not prefixed,
    # and shard_for will route to the private sandbox shards.
    sandbox_namespace =
      case Process.get(:ferricstore_sandbox) do
        nil -> nil
        %FerricStore.Sandbox{} -> nil
        ns when is_binary(ns) -> ns
      end

    raw_results = Ferricstore.Transaction.Coordinator.execute(queue, %{}, sandbox_namespace)

    ordered
    |> Enum.zip(raw_results)
    |> Enum.map(fn {cmd, raw} -> normalize_result(cmd, raw) end)
  end

  # The Coordinator returns raw Dispatcher results (RESP-level values).
  # Pipeline callers expect the same format as FerricStore public API calls.
  # This maps Dispatcher results back to the public API format.
  defp normalize_result({:get, _}, {:error, _} = err), do: err
  defp normalize_result({:get, _}, value), do: {:ok, value}

  defp normalize_result({:hget, _, _}, {:error, _} = err), do: err
  defp normalize_result({:hget, _, _}, value), do: {:ok, value}

  defp normalize_result({:del, _}, {:error, _} = err), do: err
  defp normalize_result({:del, _}, _count), do: :ok

  defp normalize_result({:hset, _, _}, {:error, _} = err), do: err
  defp normalize_result({:hset, _, _}, _count), do: :ok

  defp normalize_result({:lpush, _, _}, {:error, _} = err), do: err
  defp normalize_result({:lpush, _, _}, count) when is_integer(count), do: {:ok, count}

  defp normalize_result({:rpush, _, _}, {:error, _} = err), do: err
  defp normalize_result({:rpush, _, _}, count) when is_integer(count), do: {:ok, count}

  defp normalize_result({:sadd, _, _}, {:error, _} = err), do: err
  defp normalize_result({:sadd, _, _}, count) when is_integer(count), do: {:ok, count}

  defp normalize_result({:zadd, _, _}, {:error, _} = err), do: err
  defp normalize_result({:zadd, _, _}, count) when is_integer(count), do: {:ok, count}

  defp normalize_result({:expire, _, _}, {:error, _} = err), do: err
  defp normalize_result({:expire, _, _}, 1), do: {:ok, true}
  defp normalize_result({:expire, _, _}, 0), do: {:ok, false}

  # SET, INCR, INCR_BY already return the correct format from Dispatcher
  defp normalize_result(_, result), do: result

  defp to_resp_command({:set, key, value, opts}) do
    args = [key, value]

    args =
      case Keyword.get(opts, :ttl) do
        nil -> args
        0 -> args
        ms -> args ++ ["PX", Integer.to_string(ms)]
      end

    args =
      case Keyword.get(opts, :ex) do
        nil -> args
        seconds -> args ++ ["EX", Integer.to_string(seconds)]
      end

    args =
      case Keyword.get(opts, :px) do
        nil -> args
        ms -> args ++ ["PX", Integer.to_string(ms)]
      end

    args =
      if Keyword.get(opts, :nx, false), do: args ++ ["NX"], else: args

    args =
      if Keyword.get(opts, :xx, false), do: args ++ ["XX"], else: args

    {"SET", args}
  end

  defp to_resp_command({:get, key}), do: {"GET", [key]}
  defp to_resp_command({:del, key}), do: {"DEL", [key]}
  defp to_resp_command({:incr, key}), do: {"INCR", [key]}
  defp to_resp_command({:incr_by, key, amount}), do: {"INCRBY", [key, Integer.to_string(amount)]}

  defp to_resp_command({:hset, key, fields}) do
    flat = Enum.flat_map(fields, fn {k, v} -> [to_string(k), to_string(v)] end)
    {"HSET", [key | flat]}
  end

  defp to_resp_command({:hget, key, field}), do: {"HGET", [key, field]}
  defp to_resp_command({:lpush, key, elements}), do: {"LPUSH", [key | elements]}
  defp to_resp_command({:rpush, key, elements}), do: {"RPUSH", [key | elements]}
  defp to_resp_command({:sadd, key, members}), do: {"SADD", [key | members]}

  defp to_resp_command({:zadd, key, pairs}) do
    flat = Enum.flat_map(pairs, fn {score, member} ->
      [to_string(score), member]
    end)
    {"ZADD", [key | flat]}
  end

  defp to_resp_command({:expire, key, ttl_ms}) do
    {"PEXPIRE", [key, Integer.to_string(ttl_ms)]}
  end
end

defmodule FerricStore.Tx do
  @moduledoc """
  Transaction accumulator for executing multiple FerricStore commands atomically.

  Used with `FerricStore.multi/1` to batch multiple operations. Commands are
  accumulated in reverse order and executed sequentially when the transaction
  completes.

  ## Usage

      FerricStore.multi(fn tx ->
        tx
        |> FerricStore.Tx.set("key1", "val1")
        |> FerricStore.Tx.get("key1")
      end)

  """

  @type command ::
          {:set, binary(), binary(), keyword()}
          | {:get, binary()}
          | {:del, binary()}
          | {:incr, binary()}
          | {:incr_by, binary(), integer()}
          | {:hset, binary(), map()}
          | {:hget, binary(), binary()}
          | {:lpush, binary(), [binary()]}
          | {:rpush, binary(), [binary()]}
          | {:sadd, binary(), [binary()]}
          | {:zadd, binary(), [{number(), binary()}]}
          | {:expire, binary(), non_neg_integer()}

  @type t :: %__MODULE__{commands: [command()]}

  defstruct commands: []

  @doc "Creates a new empty transaction."
  @spec new() :: t()
  def new, do: %__MODULE__{}

  @doc "Adds a SET command to the transaction."
  @spec set(t(), binary(), binary(), keyword()) :: t()
  def set(%__MODULE__{} = tx, key, value, opts \\ []) do
    %{tx | commands: [{:set, key, value, opts} | tx.commands]}
  end

  @doc "Adds a GET command to the transaction."
  @spec get(t(), binary()) :: t()
  def get(%__MODULE__{} = tx, key) do
    %{tx | commands: [{:get, key} | tx.commands]}
  end

  @doc "Adds a DEL command to the transaction."
  @spec del(t(), binary()) :: t()
  def del(%__MODULE__{} = tx, key) do
    %{tx | commands: [{:del, key} | tx.commands]}
  end

  @doc "Adds an INCR command to the transaction."
  @spec incr(t(), binary()) :: t()
  def incr(%__MODULE__{} = tx, key) do
    %{tx | commands: [{:incr, key} | tx.commands]}
  end

  @doc "Adds an INCRBY command to the transaction."
  @spec incr_by(t(), binary(), integer()) :: t()
  def incr_by(%__MODULE__{} = tx, key, amount) do
    %{tx | commands: [{:incr_by, key, amount} | tx.commands]}
  end

  @doc "Adds an HSET command to the transaction."
  @spec hset(t(), binary(), map()) :: t()
  def hset(%__MODULE__{} = tx, key, fields) do
    %{tx | commands: [{:hset, key, fields} | tx.commands]}
  end

  @doc "Adds an HGET command to the transaction."
  @spec hget(t(), binary(), binary()) :: t()
  def hget(%__MODULE__{} = tx, key, field) do
    %{tx | commands: [{:hget, key, field} | tx.commands]}
  end

  @doc "Adds an LPUSH command to the transaction."
  @spec lpush(t(), binary(), [binary()]) :: t()
  def lpush(%__MODULE__{} = tx, key, elements) do
    %{tx | commands: [{:lpush, key, elements} | tx.commands]}
  end

  @doc "Adds an RPUSH command to the transaction."
  @spec rpush(t(), binary(), [binary()]) :: t()
  def rpush(%__MODULE__{} = tx, key, elements) do
    %{tx | commands: [{:rpush, key, elements} | tx.commands]}
  end

  @doc "Adds a SADD command to the transaction."
  @spec sadd(t(), binary(), [binary()]) :: t()
  def sadd(%__MODULE__{} = tx, key, members) do
    %{tx | commands: [{:sadd, key, members} | tx.commands]}
  end

  @doc "Adds a ZADD command to the transaction."
  @spec zadd(t(), binary(), [{number(), binary()}]) :: t()
  def zadd(%__MODULE__{} = tx, key, score_member_pairs) do
    %{tx | commands: [{:zadd, key, score_member_pairs} | tx.commands]}
  end

  @doc "Adds an EXPIRE command to the transaction."
  @spec expire(t(), binary(), non_neg_integer()) :: t()
  def expire(%__MODULE__{} = tx, key, ttl_ms) do
    %{tx | commands: [{:expire, key, ttl_ms} | tx.commands]}
  end

  @doc """
  Executes all accumulated transaction commands atomically.

  Groups commands by shard. If all target a single shard, dispatches them
  as a batch to the shard GenServer (atomic, no interleaving). If commands
  span multiple shards, returns a CROSSSLOT error.
  """
  @spec execute(t()) :: [term()] | {:error, binary()}
  def execute(%__MODULE__{commands: []}), do: []

  def execute(%__MODULE__{commands: commands}) do
    queue =
      commands
      |> Enum.reverse()
      |> Enum.map(&to_resp_command/1)

    sandbox_namespace =
      case Process.get(:ferricstore_sandbox) do
        nil -> nil
        %FerricStore.Sandbox{} -> nil
        ns when is_binary(ns) -> ns
      end

    Ferricstore.Transaction.Coordinator.execute(queue, %{}, sandbox_namespace)
  end

  defp to_resp_command({:set, key, value, opts}) do
    args = [key, value]

    args =
      case Keyword.get(opts, :ex) do
        nil -> args
        seconds -> args ++ ["EX", Integer.to_string(seconds)]
      end

    args =
      case Keyword.get(opts, :px) do
        nil -> args
        ms -> args ++ ["PX", Integer.to_string(ms)]
      end

    args =
      if Keyword.get(opts, :nx, false), do: args ++ ["NX"], else: args

    args =
      if Keyword.get(opts, :xx, false), do: args ++ ["XX"], else: args

    {"SET", args}
  end

  defp to_resp_command({:get, key}), do: {"GET", [key]}
  defp to_resp_command({:del, key}), do: {"DEL", [key]}
  defp to_resp_command({:incr, key}), do: {"INCR", [key]}
  defp to_resp_command({:incr_by, key, amount}), do: {"INCRBY", [key, Integer.to_string(amount)]}

  defp to_resp_command({:hset, key, fields}) do
    flat = Enum.flat_map(fields, fn {k, v} -> [to_string(k), to_string(v)] end)
    {"HSET", [key | flat]}
  end

  defp to_resp_command({:hget, key, field}), do: {"HGET", [key, field]}
  defp to_resp_command({:lpush, key, elements}), do: {"LPUSH", [key | elements]}
  defp to_resp_command({:rpush, key, elements}), do: {"RPUSH", [key | elements]}
  defp to_resp_command({:sadd, key, members}), do: {"SADD", [key | members]}

  defp to_resp_command({:zadd, key, pairs}) do
    flat = Enum.flat_map(pairs, fn {score, member} ->
      [to_string(score), member]
    end)
    {"ZADD", [key | flat]}
  end

  defp to_resp_command({:expire, key, ttl_ms}) do
    {"PEXPIRE", [key, Integer.to_string(ttl_ms)]}
  end
end