ferricstore

defmodule Ferricstore.Store.Ops do
  @moduledoc """
  Unified interface for store operations.

  Dispatches based on the type of `store` argument:
  - `%FerricStore.Instance{}` struct -> calls Router directly
  - `%LocalTxStore{}` struct -> local ETS for same-shard, Router for remote
  - map (closure-based store) -> calls the closure

  This allows incremental migration from closure maps to instance structs
  without changing command handler logic.
  """

  alias Ferricstore.HLC
  alias Ferricstore.Store.BlobRef
  alias Ferricstore.Store.BlobValue
  alias Ferricstore.Store.ColdRead
  alias Ferricstore.Store.CompoundKey
  alias Ferricstore.Store.Router
  alias Ferricstore.Store.LocalTxStore
  alias Ferricstore.Store.Shard.ETS, as: ShardETS
  alias Ferricstore.Store.Shard.Writes, as: ShardWrites
  alias Ferricstore.Store.Shard.ZSetIndex
  alias Ferricstore.Flow.NativeOrderedIndex, as: NativeFlowIndex

  @typep store :: FerricStore.Instance.t() | LocalTxStore.t() | map()
  @max_int64 9_223_372_036_854_775_807
  @min_int64 -9_223_372_036_854_775_808
  @overflow_error "ERR increment or decrement would overflow"
  @cold_read_timeout_ms 10_000

  defguardp valid_cold_location(file_id, offset, value_size)
            when is_integer(file_id) and file_id >= 0 and is_integer(offset) and offset >= 0 and
                   is_integer(value_size) and value_size >= 0

  defguardp valid_waraft_segment_location(file_id, offset, value_size)
            when is_tuple(file_id) and tuple_size(file_id) == 2 and
                   (elem(file_id, 0) == :waraft_segment or
                      elem(file_id, 0) == :waraft_projection or
                      elem(file_id, 0) == :waraft_apply_projection) and
                   is_integer(elem(file_id, 1)) and elem(file_id, 1) > 0 and
                   is_integer(offset) and offset >= 0 and is_integer(value_size) and
                   value_size >= 0

  # --- Basic key operations ---

  @spec get(store(), binary()) :: binary() | nil
  def get(%FerricStore.Instance{} = ctx, key), do: Router.get(ctx, key)

  def get(%LocalTxStore{} = tx, key) do
    if local?(tx, key) do
      if tx_deleted?(key), do: nil, else: local_read_value(tx, key)
    else
      Router.get(tx.instance_ctx, key)
    end
  end

  def get(store, key) when is_map(store), do: store.get.(key)

  @spec batch_get(store(), [binary()]) :: [binary() | nil]
  def batch_get(%FerricStore.Instance{} = ctx, keys), do: Router.batch_get(ctx, keys)

  def batch_get(%LocalTxStore{} = tx, keys), do: local_batch_get(tx, keys)

  def batch_get(store, keys) when is_map(store) do
    case store do
      %{batch_get: batch_get_fun} when is_function(batch_get_fun, 1) ->
        batch_get_fun.(keys)

      _ ->
        Enum.map(keys, &get(store, &1))
    end
  end

  @spec batch_put(store(), [{binary(), binary()}]) :: :ok | {:error, term()}
  def batch_put(_store, []), do: :ok

  def batch_put(%FerricStore.Instance{} = ctx, kv_pairs),
    do: Router.batch_put(ctx, kv_pairs)

  def batch_put(%LocalTxStore{} = tx, kv_pairs) do
    Enum.reduce_while(kv_pairs, :ok, fn {key, value}, :ok ->
      case put(tx, key, value, 0) do
        :ok -> {:cont, :ok}
        {:error, _} = err -> {:halt, err}
        other -> {:halt, {:error, inspect(other)}}
      end
    end)
  end

  def batch_put(store, kv_pairs) when is_map(store) do
    case store do
      %{batch_put: batch_put_fun} when is_function(batch_put_fun, 1) ->
        batch_put_fun.(kv_pairs)

      _ ->
        Enum.reduce_while(kv_pairs, :ok, fn {key, value}, :ok ->
          case put(store, key, value, 0) do
            :ok -> {:cont, :ok}
            {:error, _} = err -> {:halt, err}
            other -> {:halt, {:error, inspect(other)}}
          end
        end)
    end
  end

  @spec get_meta(store(), binary()) :: {binary(), non_neg_integer()} | nil
  def get_meta(%FerricStore.Instance{} = ctx, key), do: Router.get_meta(ctx, key)

  def get_meta(%LocalTxStore{} = tx, key) do
    if local?(tx, key) do
      if tx_deleted?(key), do: nil, else: local_read_meta(tx, key)
    else
      Router.get_meta(tx.instance_ctx, key)
    end
  end

  def get_meta(store, key) when is_map(store), do: store.get_meta.(key)

  @spec expire_at_ms(store(), binary()) :: non_neg_integer() | nil
  def expire_at_ms(%FerricStore.Instance{} = ctx, key), do: Router.expire_at_ms(ctx, key)

  def expire_at_ms(%LocalTxStore{} = tx, key) do
    if local?(tx, key) do
      case tx_pending_meta(key) do
        {_value, exp} ->
          exp

        nil ->
          case ShardETS.ets_lookup(tx.shard_state, key) do
            {:hit, _value, exp} -> exp
            {:cold, _fid, _off, _vsize, exp} -> exp
            _ -> nil
          end
      end
    else
      Router.expire_at_ms(tx.instance_ctx, key)
    end
  end

  def expire_at_ms(store, key) when is_map(store) do
    case store do
      %{expire_at_ms: expire_at_ms} when is_function(expire_at_ms, 1) ->
        expire_at_ms.(key)

      _ ->
        case get_meta(store, key) do
          nil -> nil
          {_value, expire_at_ms} -> expire_at_ms
        end
    end
  end

  @spec value_size(store(), binary()) :: non_neg_integer() | nil
  def value_size(%FerricStore.Instance{} = ctx, key), do: Router.value_size(ctx, key)

  def value_size(%LocalTxStore{} = tx, key) do
    cond do
      not local?(tx, key) ->
        Router.value_size(tx.instance_ctx, key)

      tx_deleted?(key) ->
        nil

      true ->
        case tx_pending_meta(key) do
          {value, _exp} ->
            stored_value_size(value)

          nil ->
            case ShardETS.ets_lookup(tx.shard_state, key) do
              {:hit, value, _exp} -> stored_value_size(value)
              {:cold, fid, off, vsize, _exp} -> local_cold_value_size(tx, key, fid, off, vsize)
              _ -> nil
            end
        end
    end
  end

  def value_size(store, key) when is_map(store) do
    case store do
      %{value_size: value_size} when is_function(value_size, 1) ->
        value_size.(key)

      _ ->
        case get(store, key) do
          nil -> nil
          value -> stored_value_size(value)
        end
    end
  end

  @spec object_lfu(store(), binary()) :: non_neg_integer() | nil
  def object_lfu(%FerricStore.Instance{} = ctx, key), do: Router.object_lfu(ctx, key)

  def object_lfu(%LocalTxStore{} = tx, key) do
    cond do
      not local?(tx, key) ->
        Router.object_lfu(tx.instance_ctx, key)

      tx_deleted?(key) ->
        nil

      tx_pending_meta(key) != nil ->
        Ferricstore.Store.LFU.initial()

      true ->
        now = HLC.now_ms()

        case :ets.lookup(tx.shard_state.keydir, key) do
          [{^key, value, 0, lfu, _fid, _off, _vsize}] when value != nil ->
            lfu

          [{^key, nil, 0, lfu, fid, off, vsize}] when valid_cold_location(fid, off, vsize) ->
            lfu

          [{^key, value, exp, lfu, _fid, _off, _vsize}] when exp > now and value != nil ->
            lfu

          [{^key, nil, exp, lfu, fid, off, vsize}]
          when exp > now and valid_cold_location(fid, off, vsize) ->
            lfu

          _ ->
            nil
        end
    end
  end

  def object_lfu(store, key) when is_map(store) do
    case store do
      %{object_lfu: object_lfu} when is_function(object_lfu, 1) -> object_lfu.(key)
      _ -> nil
    end
  end

  @spec getrange(store(), binary(), integer(), integer()) :: binary() | nil
  def getrange(%FerricStore.Instance{} = ctx, key, start_idx, end_idx),
    do: Router.getrange(ctx, key, start_idx, end_idx)

  def getrange(%LocalTxStore{} = tx, key, start_idx, end_idx) do
    cond do
      not local?(tx, key) ->
        Router.getrange(tx.instance_ctx, key, start_idx, end_idx)

      tx_deleted?(key) ->
        nil

      true ->
        case tx_pending_meta(key) do
          {value, _exp} ->
            range_from_value(value, start_idx, end_idx)

          nil ->
            case ShardETS.ets_lookup(tx.shard_state, key) do
              {:hit, value, _exp} ->
                range_from_value(value, start_idx, end_idx)

              {:cold, _fid, _off, _vsize, _exp} ->
                Router.getrange(tx.instance_ctx, key, start_idx, end_idx)

              _ ->
                nil
            end
        end
    end
  end

  def getrange(store, key, start_idx, end_idx) when is_map(store) do
    case store do
      %{getrange: getrange_fun} when is_function(getrange_fun, 3) ->
        getrange_fun.(key, start_idx, end_idx)

      _ ->
        case get(store, key) do
          nil -> nil
          value -> range_from_value(value, start_idx, end_idx)
        end
    end
  end

  @spec put(store(), binary(), binary(), non_neg_integer()) :: :ok | {:error, binary()}
  def put(%FerricStore.Instance{} = ctx, key, value, exp), do: Router.put(ctx, key, value, exp)

  def put(%LocalTxStore{} = tx, key, value, exp) do
    if local?(tx, key) do
      ShardETS.ets_insert(tx.shard_state, key, value, exp)
      tx_put_pending(key, value, exp)
      tx_undelete(key)
      send(self(), {:tx_pending_write, key, value, exp})
      :ok
    else
      Router.put(tx.instance_ctx, key, value, exp)
    end
  end

  def put(store, key, value, exp) when is_map(store), do: store.put.(key, value, exp)

  @spec set(store(), binary(), binary(), map()) :: term()
  def set(%FerricStore.Instance{} = ctx, key, value, opts), do: Router.set(ctx, key, value, opts)

  def set(%LocalTxStore{} = tx, key, value, opts) do
    if local?(tx, key) do
      local_set(tx, key, value, opts)
    else
      Router.set(tx.instance_ctx, key, value, opts)
    end
  end

  def set(store, key, value, opts) when is_map(store) do
    case store do
      %{set: set_fun} when is_function(set_fun, 3) ->
        set_fun.(key, value, opts)

      _ ->
        fallback_set(store, key, value, opts)
    end
  end

  @spec delete(store(), binary()) :: :ok
  def delete(%FerricStore.Instance{} = ctx, key), do: Router.delete(ctx, key)

  def delete(%LocalTxStore{} = tx, key) do
    if local?(tx, key) do
      ShardETS.ets_delete_key(tx.shard_state, key)
      tx_drop_pending(key)
      tx_mark_deleted(key)
      send(self(), {:tx_pending_delete, key})
      :ok
    else
      Router.delete(tx.instance_ctx, key)
    end
  end

  def delete(store, key) when is_map(store), do: store.delete.(key)

  @spec exists?(store(), binary()) :: boolean()
  def exists?(%FerricStore.Instance{} = ctx, key), do: Router.exists?(ctx, key)

  def exists?(%LocalTxStore{} = tx, key) do
    if local?(tx, key) do
      local_exists?(tx, key)
    else
      Router.exists?(tx.instance_ctx, key)
    end
  end

  def exists?(store, key) when is_map(store), do: store.exists?.(key)

  @spec keys(store()) :: [binary()]
  def keys(%FerricStore.Instance{} = ctx), do: Router.keys(ctx)
  def keys(%LocalTxStore{} = tx), do: Router.keys(tx.instance_ctx)
  def keys(store) when is_map(store), do: store.keys.()

  @spec dbsize(store()) :: non_neg_integer()
  def dbsize(%FerricStore.Instance{} = ctx), do: Router.dbsize(ctx)
  def dbsize(%LocalTxStore{} = tx), do: Router.dbsize(tx.instance_ctx)
  def dbsize(store) when is_map(store), do: store.dbsize.()

  # --- Numeric operations ---

  @spec incr(store(), binary(), integer()) :: {:ok, integer()} | {:error, binary()}
  def incr(%FerricStore.Instance{} = ctx, key, delta), do: Router.incr(ctx, key, delta)

  def incr(%LocalTxStore{} = tx, key, delta) do
    if local?(tx, key) do
      {current, expire_at_ms} = local_read_meta_for_rmw(tx, key)

      case current do
        nil ->
          case checked_integer_add(0, delta) do
            {:ok, new_val} ->
              ShardETS.ets_insert(tx.shard_state, key, new_val, 0)
              tx_put_pending(key, new_val, 0)
              send(self(), {:tx_pending_write, key, new_val, 0})
              {:ok, new_val}

            :overflow ->
              {:error, @overflow_error}
          end

        value ->
          case ShardETS.coerce_integer(value) do
            {:ok, int_val} ->
              case checked_integer_add(int_val, delta) do
                {:ok, new_val} ->
                  ShardETS.ets_insert(tx.shard_state, key, new_val, expire_at_ms)
                  tx_put_pending(key, new_val, expire_at_ms)
                  send(self(), {:tx_pending_write, key, new_val, expire_at_ms})
                  {:ok, new_val}

                :overflow ->
                  {:error, @overflow_error}
              end

            :error ->
              {:error, "ERR value is not an integer or out of range"}
          end
      end
    else
      Router.incr(tx.instance_ctx, key, delta)
    end
  end

  def incr(store, key, delta) when is_map(store), do: store.incr.(key, delta)

  defp checked_integer_add(value, delta) do
    result = value + delta

    if result > @max_int64 or result < @min_int64 do
      :overflow
    else
      {:ok, result}
    end
  end

  @spec incr_float(store(), binary(), float()) :: {:ok, binary()} | {:error, binary()}
  def incr_float(%FerricStore.Instance{} = ctx, key, delta),
    do: Router.incr_float(ctx, key, delta)

  def incr_float(%LocalTxStore{} = tx, key, delta) do
    if local?(tx, key) do
      {current, expire_at_ms} = local_read_meta_for_rmw(tx, key)

      case current do
        nil ->
          new_val = delta * 1.0
          ShardETS.ets_insert(tx.shard_state, key, new_val, 0)
          tx_put_pending(key, new_val, 0)
          send(self(), {:tx_pending_write, key, new_val, 0})
          {:ok, new_val}

        value ->
          case ShardETS.coerce_float(value) do
            {:ok, float_val} ->
              new_val = float_val + delta
              ShardETS.ets_insert(tx.shard_state, key, new_val, expire_at_ms)
              tx_put_pending(key, new_val, expire_at_ms)
              send(self(), {:tx_pending_write, key, new_val, expire_at_ms})
              {:ok, new_val}

            :error ->
              {:error, "ERR value is not a valid float"}
          end
      end
    else
      Router.incr_float(tx.instance_ctx, key, delta)
    end
  end

  def incr_float(store, key, delta) when is_map(store), do: store.incr_float.(key, delta)

  # --- String mutation operations ---

  @spec append(store(), binary(), binary()) :: {:ok, non_neg_integer()}
  def append(%FerricStore.Instance{} = ctx, key, suffix), do: Router.append(ctx, key, suffix)

  def append(%LocalTxStore{} = tx, key, suffix) do
    if local?(tx, key) do
      {current, expire_at_ms} =
        case local_read_meta_for_rmw(tx, key) do
          {nil, _exp} -> {"", 0}
          {value, exp} -> {ShardETS.to_disk_binary(value), exp}
        end

      new_val = current <> suffix
      ShardETS.ets_insert(tx.shard_state, key, new_val, expire_at_ms)
      tx_put_pending(key, new_val, expire_at_ms)
      send(self(), {:tx_pending_write, key, new_val, expire_at_ms})
      {:ok, byte_size(new_val)}
    else
      Router.append(tx.instance_ctx, key, suffix)
    end
  end

  def append(store, key, suffix) when is_map(store), do: store.append.(key, suffix)

  @spec getset(store(), binary(), binary()) :: binary() | nil
  def getset(%FerricStore.Instance{} = ctx, key, value), do: Router.getset(ctx, key, value)

  def getset(%LocalTxStore{} = tx, key, new_value) do
    if local?(tx, key) do
      old = local_read_value_for_rmw(tx, key)
      ShardETS.ets_insert(tx.shard_state, key, new_value, 0)
      tx_put_pending(key, new_value, 0)
      send(self(), {:tx_pending_write, key, new_value, 0})
      old
    else
      Router.getset(tx.instance_ctx, key, new_value)
    end
  end

  def getset(store, key, value) when is_map(store), do: store.getset.(key, value)

  @spec getdel(store(), binary()) :: binary() | nil
  def getdel(%FerricStore.Instance{} = ctx, key), do: Router.getdel(ctx, key)

  def getdel(%LocalTxStore{} = tx, key) do
    if local?(tx, key) do
      old = local_read_value_for_rmw(tx, key)

      if old do
        ShardETS.ets_delete_key(tx.shard_state, key)
        tx_drop_pending(key)
        tx_mark_deleted(key)
        send(self(), {:tx_pending_delete, key})
      end

      old
    else
      Router.getdel(tx.instance_ctx, key)
    end
  end

  def getdel(store, key) when is_map(store), do: store.getdel.(key)

  @spec getex(store(), binary(), non_neg_integer()) :: binary() | nil
  def getex(%FerricStore.Instance{} = ctx, key, exp), do: Router.getex(ctx, key, exp)

  def getex(%LocalTxStore{} = tx, key, expire_at_ms) do
    if local?(tx, key) do
      value = local_read_value_for_rmw(tx, key)

      if value do
        ShardETS.ets_insert(tx.shard_state, key, value, expire_at_ms)
        tx_put_pending(key, value, expire_at_ms)
        send(self(), {:tx_pending_write, key, value, expire_at_ms})
      end

      value
    else
      Router.getex(tx.instance_ctx, key, expire_at_ms)
    end
  end

  def getex(store, key, exp) when is_map(store), do: store.getex.(key, exp)

  @spec setrange(store(), binary(), non_neg_integer(), binary()) :: {:ok, non_neg_integer()}
  def setrange(%FerricStore.Instance{} = ctx, key, offset, value),
    do: Router.setrange(ctx, key, offset, value)

  def setrange(%LocalTxStore{} = tx, key, offset, value) do
    if local?(tx, key) do
      {old, expire_at_ms} =
        case local_read_meta_for_rmw(tx, key) do
          {nil, _exp} -> {"", 0}
          {v, exp} -> {ShardETS.to_disk_binary(v), exp}
        end

      new_val = ShardWrites.apply_setrange(old, offset, value)
      ShardETS.ets_insert(tx.shard_state, key, new_val, expire_at_ms)
      tx_put_pending(key, new_val, expire_at_ms)
      send(self(), {:tx_pending_write, key, new_val, expire_at_ms})
      {:ok, byte_size(new_val)}
    else
      Router.setrange(tx.instance_ctx, key, offset, value)
    end
  end

  def setrange(store, key, offset, value) when is_map(store),
    do: store.setrange.(key, offset, value)

  # --- Native operations ---

  @spec cas(store(), binary(), binary(), binary(), non_neg_integer() | nil) :: 1 | 0 | nil
  def cas(%FerricStore.Instance{} = ctx, key, expected, new_val, ttl),
    do: Router.cas(ctx, key, expected, new_val, ttl)

  def cas(%LocalTxStore{} = tx, key, expected, new_val, ttl),
    do: Router.cas(tx.instance_ctx, key, expected, new_val, ttl)

  def cas(store, key, expected, new_val, ttl) when is_map(store),
    do: store.cas.(key, expected, new_val, ttl)

  @spec lock(store(), binary(), binary(), pos_integer()) :: :ok | {:error, binary()}
  def lock(%FerricStore.Instance{} = ctx, key, owner, ttl), do: Router.lock(ctx, key, owner, ttl)

  def lock(%LocalTxStore{} = tx, key, owner, ttl),
    do: Router.lock(tx.instance_ctx, key, owner, ttl)

  def lock(store, key, owner, ttl) when is_map(store), do: store.lock.(key, owner, ttl)

  @spec unlock(store(), binary(), binary()) :: 1 | {:error, binary()}
  def unlock(%FerricStore.Instance{} = ctx, key, owner), do: Router.unlock(ctx, key, owner)
  def unlock(%LocalTxStore{} = tx, key, owner), do: Router.unlock(tx.instance_ctx, key, owner)
  def unlock(store, key, owner) when is_map(store), do: store.unlock.(key, owner)

  @spec extend(store(), binary(), binary(), pos_integer()) :: 1 | {:error, binary()}
  def extend(%FerricStore.Instance{} = ctx, key, owner, ttl),
    do: Router.extend(ctx, key, owner, ttl)

  def extend(%LocalTxStore{} = tx, key, owner, ttl),
    do: Router.extend(tx.instance_ctx, key, owner, ttl)

  def extend(store, key, owner, ttl) when is_map(store), do: store.extend.(key, owner, ttl)

  @spec ratelimit_add(store(), binary(), pos_integer(), pos_integer(), pos_integer()) :: [term()]
  def ratelimit_add(%FerricStore.Instance{} = ctx, key, window, max, count),
    do: Router.ratelimit_add(ctx, key, window, max, count)

  def ratelimit_add(%LocalTxStore{} = tx, key, window, max, count),
    do: Router.ratelimit_add(tx.instance_ctx, key, window, max, count)

  def ratelimit_add(store, key, window, max, count) when is_map(store),
    do: store.ratelimit_add.(key, window, max, count)

  # --- List operations ---

  @spec list_op(store(), binary(), term()) :: term()
  def list_op(%FerricStore.Instance{} = ctx, key, op), do: Router.list_op(ctx, key, op)
  def list_op(%LocalTxStore{} = tx, key, op), do: Router.list_op(tx.instance_ctx, key, op)
  def list_op(store, key, op) when is_map(store), do: store.list_op.(key, op)

  # --- Compound key capability check ---

  @spec has_compound?(store()) :: boolean()
  def has_compound?(%FerricStore.Instance{}), do: true
  def has_compound?(%LocalTxStore{}), do: true
  def has_compound?(store) when is_map(store), do: is_map_key(store, :compound_get)

  # --- Compound key operations ---

  @spec compound_get(store(), binary(), binary()) :: binary() | nil
  def compound_get(%FerricStore.Instance{} = ctx, redis_key, compound_key),
    do: Router.compound_get(ctx, redis_key, compound_key)

  def compound_get(%LocalTxStore{} = tx, redis_key, compound_key) do
    if local?(tx, redis_key) do
      case promoted_path(tx, redis_key) do
        nil ->
          local_read_value(tx, compound_key)

        dedicated_path ->
          if shared_log_compound_key?(compound_key) do
            local_read_value(tx, compound_key)
          else
            local_promoted_read_value(tx, compound_key, dedicated_path)
          end
      end
    else
      idx = Router.shard_for(tx.instance_ctx, redis_key)

      case Router.safe_read_call(tx.instance_ctx, idx, {:compound_get, redis_key, compound_key}) do
        {:ok, value} -> value
        :unavailable -> nil
      end
    end
  end

  def compound_get(store, redis_key, compound_key) when is_map(store),
    do: store.compound_get.(redis_key, compound_key)

  @spec compound_batch_get(store(), binary(), [binary()]) :: [binary() | nil]
  def compound_batch_get(%FerricStore.Instance{} = ctx, redis_key, compound_keys),
    do: Router.compound_batch_get(ctx, redis_key, compound_keys)

  def compound_batch_get(%LocalTxStore{} = tx, redis_key, compound_keys) do
    if local?(tx, redis_key) do
      case promoted_path(tx, redis_key) do
        nil ->
          local_batch_read_values(tx, compound_keys, tx.shard_state.shard_data_path)

        dedicated_path ->
          if Enum.any?(compound_keys, &shared_log_compound_key?/1) do
            local_promoted_batch_read_values(tx, compound_keys, dedicated_path)
          else
            local_batch_read_values(tx, compound_keys, dedicated_path)
          end
      end
    else
      idx = Router.shard_for(tx.instance_ctx, redis_key)

      case Router.safe_read_call(
             tx.instance_ctx,
             idx,
             {:compound_batch_get, redis_key, compound_keys}
           ) do
        {:ok, values} -> values
        :unavailable -> List.duplicate(nil, length(compound_keys))
      end
    end
  end

  def compound_batch_get(store, redis_key, compound_keys) when is_map(store) do
    case store do
      %{compound_batch_get: compound_batch_get_fun} when is_function(compound_batch_get_fun, 2) ->
        compound_batch_get_fun.(redis_key, compound_keys)

      _ ->
        Enum.map(compound_keys, &compound_get(store, redis_key, &1))
    end
  end

  @spec compound_get_meta(store(), binary(), binary()) :: {binary(), non_neg_integer()} | nil
  def compound_get_meta(%FerricStore.Instance{} = ctx, redis_key, compound_key),
    do: Router.compound_get_meta(ctx, redis_key, compound_key)

  def compound_get_meta(%LocalTxStore{} = tx, redis_key, compound_key) do
    if local?(tx, redis_key) do
      case promoted_path(tx, redis_key) do
        nil ->
          local_read_meta(tx, compound_key)

        dedicated_path ->
          if shared_log_compound_key?(compound_key) do
            local_read_meta(tx, compound_key)
          else
            local_promoted_read_meta(tx, compound_key, dedicated_path)
          end
      end
    else
      idx = Router.shard_for(tx.instance_ctx, redis_key)

      case Router.safe_read_call(
             tx.instance_ctx,
             idx,
             {:compound_get_meta, redis_key, compound_key}
           ) do
        {:ok, meta} -> meta
        :unavailable -> nil
      end
    end
  end

  def compound_get_meta(store, redis_key, compound_key) when is_map(store),
    do: store.compound_get_meta.(redis_key, compound_key)

  @spec compound_batch_get_meta(store(), binary(), [binary()]) ::
          [{binary(), non_neg_integer()} | nil]
  def compound_batch_get_meta(%FerricStore.Instance{} = ctx, redis_key, compound_keys),
    do: Router.compound_batch_get_meta(ctx, redis_key, compound_keys)

  def compound_batch_get_meta(%LocalTxStore{} = tx, redis_key, compound_keys) do
    if local?(tx, redis_key) do
      case promoted_path(tx, redis_key) do
        nil ->
          local_batch_read_meta(tx, compound_keys, tx.shard_state.shard_data_path)

        dedicated_path ->
          if Enum.any?(compound_keys, &shared_log_compound_key?/1) do
            local_promoted_batch_read_meta(tx, compound_keys, dedicated_path)
          else
            local_batch_read_meta(tx, compound_keys, dedicated_path)
          end
      end
    else
      idx = Router.shard_for(tx.instance_ctx, redis_key)

      case Router.safe_read_call(
             tx.instance_ctx,
             idx,
             {:compound_batch_get_meta, redis_key, compound_keys}
           ) do
        {:ok, metas} -> metas
        :unavailable -> List.duplicate(nil, length(compound_keys))
      end
    end
  end

  def compound_batch_get_meta(store, redis_key, compound_keys) when is_map(store) do
    case store do
      %{compound_batch_get_meta: compound_batch_get_meta_fun}
      when is_function(compound_batch_get_meta_fun, 2) ->
        compound_batch_get_meta_fun.(redis_key, compound_keys)

      _ ->
        Enum.map(compound_keys, &compound_get_meta(store, redis_key, &1))
    end
  end

  @spec compound_put(store(), binary(), binary(), binary(), non_neg_integer()) :: :ok
  def compound_put(%FerricStore.Instance{} = ctx, redis_key, compound_key, value, exp),
    do: Router.compound_put(ctx, redis_key, compound_key, value, exp)

  def compound_put(%LocalTxStore{} = tx, redis_key, compound_key, value, expire_at_ms) do
    if local?(tx, redis_key) do
      ShardETS.ets_insert(tx.shard_state, compound_key, value, expire_at_ms)
      tx_put_pending(compound_key, value, expire_at_ms)
      tx_undelete(compound_key)
      send(self(), tx_compound_write_message(tx, redis_key, compound_key, value, expire_at_ms))
      :ok
    else
      Router.compound_put(tx.instance_ctx, redis_key, compound_key, value, expire_at_ms)
    end
  end

  def compound_put(store, redis_key, compound_key, value, exp) when is_map(store),
    do: store.compound_put.(redis_key, compound_key, value, exp)

  @spec compound_batch_put(store(), binary(), [{binary(), binary(), non_neg_integer()}]) :: :ok
  def compound_batch_put(_store, _redis_key, []), do: :ok

  def compound_batch_put(%FerricStore.Instance{} = ctx, redis_key, entries),
    do: Router.compound_batch_put(ctx, redis_key, entries)

  def compound_batch_put(%LocalTxStore{} = tx, redis_key, entries) do
    if local?(tx, redis_key) do
      Enum.each(entries, fn {compound_key, value, expire_at_ms} ->
        ShardETS.ets_insert(tx.shard_state, compound_key, value, expire_at_ms)
        tx_put_pending(compound_key, value, expire_at_ms)
        tx_undelete(compound_key)
        send(self(), tx_compound_write_message(tx, redis_key, compound_key, value, expire_at_ms))
      end)

      :ok
    else
      Router.compound_batch_put(tx.instance_ctx, redis_key, entries)
    end
  end

  def compound_batch_put(store, redis_key, entries) when is_map(store) do
    case store do
      %{compound_batch_put: compound_batch_put_fun} when is_function(compound_batch_put_fun, 2) ->
        compound_batch_put_fun.(redis_key, entries)

      _ ->
        Enum.reduce_while(entries, :ok, fn {compound_key, value, expire_at_ms}, :ok ->
          case compound_put(store, redis_key, compound_key, value, expire_at_ms) do
            :ok -> {:cont, :ok}
            {:error, _} = err -> {:halt, err}
            other -> {:halt, {:error, inspect(other)}}
          end
        end)
    end
  end

  @spec compound_delete(store(), binary(), binary()) :: :ok
  def compound_delete(%FerricStore.Instance{} = ctx, redis_key, compound_key),
    do: Router.compound_delete(ctx, redis_key, compound_key)

  def compound_delete(%LocalTxStore{} = tx, redis_key, compound_key) do
    if local?(tx, redis_key) do
      ShardETS.ets_delete_key(tx.shard_state, compound_key)
      tx_drop_pending(compound_key)
      tx_mark_deleted(compound_key)
      send(self(), tx_compound_delete_message(tx, redis_key, compound_key))
      :ok
    else
      Router.compound_delete(tx.instance_ctx, redis_key, compound_key)
    end
  end

  def compound_delete(store, redis_key, compound_key) when is_map(store),
    do: store.compound_delete.(redis_key, compound_key)

  @spec compound_batch_delete(store(), binary(), [binary()]) :: :ok | {:error, term()}
  def compound_batch_delete(_store, _redis_key, []), do: :ok

  def compound_batch_delete(%FerricStore.Instance{} = ctx, redis_key, compound_keys),
    do: Router.compound_batch_delete(ctx, redis_key, compound_keys)

  def compound_batch_delete(%LocalTxStore{} = tx, redis_key, compound_keys) do
    if local?(tx, redis_key) do
      Enum.each(compound_keys, fn compound_key ->
        ShardETS.ets_delete_key(tx.shard_state, compound_key)
        tx_drop_pending(compound_key)
        tx_mark_deleted(compound_key)
        send(self(), tx_compound_delete_message(tx, redis_key, compound_key))
      end)

      :ok
    else
      Router.compound_batch_delete(tx.instance_ctx, redis_key, compound_keys)
    end
  end

  def compound_batch_delete(store, redis_key, compound_keys) when is_map(store) do
    case store do
      %{compound_batch_delete: compound_batch_delete_fun}
      when is_function(compound_batch_delete_fun, 2) ->
        compound_batch_delete_fun.(redis_key, compound_keys)

      _ ->
        Enum.reduce_while(compound_keys, :ok, fn compound_key, :ok ->
          case compound_delete(store, redis_key, compound_key) do
            :ok -> {:cont, :ok}
            {:error, _} = err -> {:halt, err}
            other -> {:halt, {:error, inspect(other)}}
          end
        end)
    end
  end

  @spec compound_scan(store(), binary(), binary()) :: [{binary(), binary()}]
  def compound_scan(%FerricStore.Instance{} = ctx, redis_key, prefix),
    do: Router.compound_scan(ctx, redis_key, prefix)

  def compound_scan(%LocalTxStore{} = tx, redis_key, prefix) do
    if local?(tx, redis_key) do
      shard_data_path = promoted_path(tx, redis_key) || tx.shard_state.shard_data_path
      results = ShardETS.prefix_scan_entries(tx.shard_state, prefix, shard_data_path)

      results
      |> merge_tx_pending_prefix(prefix)
      |> Enum.sort_by(fn {field, _} -> field end)
    else
      idx = Router.shard_for(tx.instance_ctx, redis_key)

      case Router.safe_read_call(tx.instance_ctx, idx, {:compound_scan, redis_key, prefix}) do
        {:ok, results} -> results
        :unavailable -> []
      end
    end
  end

  def compound_scan(store, redis_key, prefix) when is_map(store),
    do: store.compound_scan.(redis_key, prefix)

  @spec compound_fields(store(), binary(), binary()) :: [binary()]
  def compound_fields(%FerricStore.Instance{} = ctx, redis_key, prefix),
    do: Router.compound_fields(ctx, redis_key, prefix)

  def compound_fields(store, redis_key, prefix) do
    store
    |> compound_scan(redis_key, prefix)
    |> Enum.map(fn {field, _value} -> field end)
  end

  @spec compound_count(store(), binary(), binary()) :: non_neg_integer()
  def compound_count(%FerricStore.Instance{} = ctx, redis_key, prefix),
    do: Router.compound_count(ctx, redis_key, prefix)

  def compound_count(%LocalTxStore{} = tx, redis_key, prefix) do
    if local?(tx, redis_key) do
      ShardETS.prefix_count_entries(tx.shard_state, prefix)
    else
      idx = Router.shard_for(tx.instance_ctx, redis_key)

      case Router.safe_read_call(tx.instance_ctx, idx, {:compound_count, redis_key, prefix}) do
        {:ok, count} -> count
        :unavailable -> 0
      end
    end
  end

  def compound_count(store, redis_key, prefix) when is_map(store),
    do: store.compound_count.(redis_key, prefix)

  @spec zset_score_range(store(), binary(), term(), term(), boolean()) ::
          {:ok, [{binary(), float()}]} | :unavailable
  def zset_score_range(%FerricStore.Instance{} = ctx, redis_key, min_bound, max_bound, reverse?),
    do: Router.zset_score_range(ctx, redis_key, min_bound, max_bound, reverse?)

  def zset_score_range(%LocalTxStore{} = tx, redis_key, min_bound, max_bound, reverse?) do
    local_zset_index_read(tx, redis_key, fn state ->
      {:ok, ZSetIndex.range(state.zset_score_index, redis_key, min_bound, max_bound, reverse?)}
    end)
  end

  def zset_score_range(store, redis_key, min_bound, max_bound, reverse?) when is_map(store) do
    case store do
      %{zset_score_range: fun} when is_function(fun, 4) ->
        fun.(redis_key, min_bound, max_bound, reverse?)

      _ ->
        :unavailable
    end
  end

  @spec zset_score_range_slice(
          store(),
          binary(),
          term(),
          term(),
          boolean(),
          non_neg_integer(),
          non_neg_integer() | :all
        ) ::
          {:ok, [{binary(), float()}]} | :unavailable
  def zset_score_range_slice(
        %FerricStore.Instance{} = ctx,
        redis_key,
        min_bound,
        max_bound,
        reverse?,
        offset,
        count
      ),
      do:
        Router.zset_score_range_slice(
          ctx,
          redis_key,
          min_bound,
          max_bound,
          reverse?,
          offset,
          count
        )

  def zset_score_range_slice(
        %LocalTxStore{} = tx,
        redis_key,
        min_bound,
        max_bound,
        reverse?,
        offset,
        count
      ) do
    local_zset_index_read(tx, redis_key, fn state ->
      {:ok,
       ZSetIndex.range_slice(
         state.zset_score_index,
         redis_key,
         min_bound,
         max_bound,
         reverse?,
         offset,
         count
       )}
    end)
  end

  def zset_score_range_slice(store, redis_key, min_bound, max_bound, reverse?, offset, count)
      when is_map(store) do
    case store do
      %{zset_score_range_slice: fun} when is_function(fun, 6) ->
        fun.(redis_key, min_bound, max_bound, reverse?, offset, count)

      _ ->
        :unavailable
    end
  end

  @spec zset_score_count(store(), binary(), term(), term()) ::
          {:ok, non_neg_integer()} | :unavailable
  def zset_score_count(%FerricStore.Instance{} = ctx, redis_key, min_bound, max_bound),
    do: Router.zset_score_count(ctx, redis_key, min_bound, max_bound)

  def zset_score_count(%LocalTxStore{} = tx, redis_key, min_bound, max_bound) do
    local_zset_index_read(tx, redis_key, fn state ->
      {:ok,
       ZSetIndex.count(
         state.zset_score_index,
         state.zset_score_lookup,
         redis_key,
         min_bound,
         max_bound
       )}
    end)
  end

  def zset_score_count(store, redis_key, min_bound, max_bound) when is_map(store) do
    case store do
      %{zset_score_count: fun} when is_function(fun, 3) ->
        fun.(redis_key, min_bound, max_bound)

      _ ->
        :unavailable
    end
  end

  @spec zset_rank_range(store(), binary(), non_neg_integer(), non_neg_integer(), boolean()) ::
          {:ok, [{binary(), float()}]} | :unavailable
  def zset_rank_range(%FerricStore.Instance{} = ctx, redis_key, start_idx, stop_idx, reverse?),
    do: Router.zset_rank_range(ctx, redis_key, start_idx, stop_idx, reverse?)

  def zset_rank_range(%LocalTxStore{} = tx, redis_key, start_idx, stop_idx, reverse?) do
    local_zset_index_read(tx, redis_key, fn state ->
      {:ok,
       ZSetIndex.rank_range(state.zset_score_index, redis_key, start_idx, stop_idx, reverse?)}
    end)
  end

  def zset_rank_range(store, redis_key, start_idx, stop_idx, reverse?) when is_map(store) do
    case store do
      %{zset_rank_range: fun} when is_function(fun, 4) ->
        fun.(redis_key, start_idx, stop_idx, reverse?)

      _ ->
        :unavailable
    end
  end

  @spec zset_member_rank(store(), binary(), binary(), boolean()) ::
          {:ok, non_neg_integer() | nil} | :unavailable
  def zset_member_rank(%FerricStore.Instance{} = ctx, redis_key, member, reverse?),
    do: Router.zset_member_rank(ctx, redis_key, member, reverse?)

  def zset_member_rank(%LocalTxStore{} = tx, redis_key, member, reverse?) do
    local_zset_index_read(tx, redis_key, fn state ->
      {:ok,
       ZSetIndex.member_rank(
         state.zset_score_index,
         state.zset_score_lookup,
         redis_key,
         member,
         reverse?
       )}
    end)
  end

  def zset_member_rank(store, redis_key, member, reverse?) when is_map(store) do
    case store do
      %{zset_member_rank: fun} when is_function(fun, 3) ->
        fun.(redis_key, member, reverse?)

      _ ->
        :unavailable
    end
  end

  @spec compound_delete_prefix(store(), binary(), binary()) :: :ok
  def compound_delete_prefix(%FerricStore.Instance{} = ctx, redis_key, prefix),
    do: Router.compound_delete_prefix(ctx, redis_key, prefix)

  def compound_delete_prefix(%LocalTxStore{} = tx, redis_key, prefix) do
    if local?(tx, redis_key) do
      keys_to_delete = ShardETS.prefix_collect_keys(tx.shard_state.keydir, prefix)

      Enum.each(keys_to_delete, fn key ->
        ShardETS.ets_delete_key(tx.shard_state, key)
        tx_drop_pending(key)
        tx_mark_deleted(key)
        send(self(), tx_compound_delete_message(tx, redis_key, key))
      end)

      :ok
    else
      Router.compound_delete_prefix(tx.instance_ctx, redis_key, prefix)
    end
  end

  def compound_delete_prefix(store, redis_key, prefix) when is_map(store),
    do: store.compound_delete_prefix.(redis_key, prefix)

  # --- Prob operations ---

  @spec prob_write(store(), tuple()) :: term()
  def prob_write(%FerricStore.Instance{} = ctx, command), do: Router.prob_write(ctx, command)
  def prob_write(%LocalTxStore{} = tx, command), do: Router.prob_write(tx.instance_ctx, command)
  def prob_write(store, command) when is_map(store), do: store.prob_write.(command)

  @spec prob_dir(store(), binary()) :: binary()
  def prob_dir(%FerricStore.Instance{} = ctx, key) do
    idx = Router.shard_for(ctx, key)
    shard_path = Ferricstore.DataDir.shard_data_path(ctx.data_dir, idx)
    Path.join(shard_path, "prob")
  end

  def prob_dir(%LocalTxStore{} = tx, _key) do
    Path.join(tx.shard_state.shard_data_path, "prob")
  end

  def prob_dir(store, key) when is_map(store) and is_map_key(store, :prob_dir_for_key),
    do: store.prob_dir_for_key.(key)

  def prob_dir(store, _key) when is_map(store) and is_map_key(store, :prob_dir),
    do: store.prob_dir.()

  # --- Flush ---

  @spec flush(store()) :: :ok | {:error, term()}
  def flush(%FerricStore.Instance{} = ctx) do
    with :ok <- flush_keys(ctx, Router.keys(ctx)),
         :ok <- clear_flow_projection_storage(ctx) do
      clear_stream_tables()
      NativeFlowIndex.reset_all(ctx.name, ctx.shard_count)
      :ok
    end
  end

  def flush(%LocalTxStore{} = tx) do
    with :ok <- flush_keys(tx.instance_ctx, Router.keys(tx.instance_ctx)),
         :ok <- clear_flow_projection_storage(tx.instance_ctx) do
      clear_stream_tables()
      NativeFlowIndex.reset_all(tx.instance_ctx.name, tx.instance_ctx.shard_count)
      :ok
    end
  end

  def flush(store) when is_map(store), do: store.flush.()

  defp clear_flow_projection_storage(ctx) do
    Ferricstore.Flow.LMDBWriter.suspend_all(ctx.name, ctx.shard_count, flush: false)

    try do
      with :ok <- discard_flow_history_projectors(ctx),
           :ok <- discard_flow_lmdb_writers(ctx),
           :ok <- Ferricstore.Flow.LMDB.clear_all(ctx.data_dir, ctx.shard_count),
           :ok <- clear_flow_history_dirs(ctx) do
        :ok
      end
    after
      Ferricstore.Flow.LMDBWriter.resume_all(ctx.name, ctx.shard_count)
    end
  end

  defp discard_flow_lmdb_writers(%{name: name, shard_count: shard_count})
       when is_atom(name) and is_integer(shard_count) and shard_count >= 0 do
    Ferricstore.Flow.LMDBWriter.discard_all(name, shard_count)
  end

  defp discard_flow_lmdb_writers(_ctx), do: :ok

  defp discard_flow_history_projectors(ctx) do
    0..max(ctx.shard_count - 1, -1)//1
    |> Enum.reduce_while(:ok, fn shard_index, :ok ->
      case Ferricstore.Flow.HistoryProjector.discard(ctx, shard_index, 5_000) do
        :ok -> {:cont, :ok}
        {:error, _reason} = error -> {:halt, error}
      end
    end)
  end

  defp clear_flow_history_dirs(ctx) do
    0..max(ctx.shard_count - 1, -1)//1
    |> Enum.reduce_while(:ok, fn shard_index, :ok ->
      history_dir =
        ctx.data_dir
        |> Ferricstore.DataDir.shard_data_path(shard_index)
        |> Ferricstore.Flow.HistoryProjector.history_dir()

      case Ferricstore.FS.rm_rf(history_dir) do
        :ok -> {:cont, :ok}
        {:error, _reason} = error -> {:halt, error}
      end
    end)
  end

  defp flush_keys(store, keys) do
    keys
    |> CompoundKey.user_visible_keys()
    |> Enum.reduce_while(:ok, fn key, :ok ->
      case flush_key(store, key) do
        :ok -> {:cont, :ok}
        {:error, _reason} = error -> {:halt, error}
        other -> {:halt, {:error, {:flush_key_failed, key, other}}}
      end
    end)
  end

  defp flush_key(store, key) do
    type_key = CompoundKey.type_key(key)

    case compound_get(store, key, type_key) do
      "hash" ->
        run_flush_steps(key, [
          fn -> compound_delete_prefix(store, key, CompoundKey.hash_prefix(key)) end,
          fn -> compound_delete(store, key, type_key) end
        ])

      "list" ->
        run_flush_steps(key, [
          fn -> compound_delete_prefix(store, key, CompoundKey.list_prefix(key)) end,
          fn -> compound_delete(store, key, CompoundKey.list_meta_key(key)) end,
          fn -> compound_delete(store, key, type_key) end,
          fn -> delete(store, key) end
        ])

      "set" ->
        run_flush_steps(key, [
          fn -> compound_delete_prefix(store, key, CompoundKey.set_prefix(key)) end,
          fn -> compound_delete(store, key, type_key) end
        ])

      "zset" ->
        run_flush_steps(key, [
          fn -> compound_delete_prefix(store, key, CompoundKey.zset_prefix(key)) end,
          fn -> compound_delete(store, key, type_key) end
        ])

      "stream" ->
        run_flush_steps(key, [
          fn -> compound_delete_prefix(store, key, "X:" <> key <> <<0>>) end,
          fn -> compound_delete_prefix(store, key, CompoundKey.stream_group_prefix(key)) end,
          fn -> compound_delete(store, key, CompoundKey.stream_meta_key(key)) end,
          fn -> compound_delete(store, key, type_key) end,
          fn -> delete(store, key) end
        ])

      nil ->
        delete(store, key)

      _unknown ->
        run_flush_steps(key, [
          fn -> compound_delete(store, key, type_key) end,
          fn -> delete(store, key) end
        ])
    end
  end

  defp run_flush_steps(key, steps) do
    Enum.reduce_while(steps, :ok, fn step, :ok ->
      case step.() do
        :ok -> {:cont, :ok}
        {:error, _reason} = error -> {:halt, error}
        other -> {:halt, {:error, {:flush_key_failed, key, other}}}
      end
    end)
  end

  defp clear_stream_tables do
    Ferricstore.Stream.LocalState.clear()
  end

  # --- On push callback (for Waiters notification) ---

  @spec on_push(store(), binary(), non_neg_integer()) :: [pid()] | pid() | nil
  def on_push(store, key, count \\ 1) do
    case store do
      %FerricStore.Instance{} -> Ferricstore.Waiters.notify_push(key, count)
      %LocalTxStore{} -> Ferricstore.Waiters.notify_push(key, count)
      store when is_map(store) ->
        case store[:on_push] do
          fun when is_function(fun, 2) -> fun.(key, count)
          fun when is_function(fun, 1) -> fun.(key)
          _ -> nil
        end
    end
  end

  # ===================================================================
  # Private helpers for LocalTxStore
  # ===================================================================

  defp local?(tx, key), do: Router.shard_for(tx.instance_ctx, key) == tx.shard_index

  defp local_zset_index_read(%LocalTxStore{} = tx, redis_key, fun) do
    cond do
      not local?(tx, redis_key) ->
        :unavailable

      not tx_zset_index_clean?() ->
        :unavailable

      not local_zset_tables?(tx.shard_state) ->
        :unavailable

      true ->
        redis_key
        |> local_zset_index_state(tx)
        |> fun.()
    end
  end

  defp local_zset_index_state(redis_key, tx) do
    data_path = promoted_path(tx, redis_key) || tx.shard_state.shard_data_path
    prefix = CompoundKey.zset_prefix(redis_key)

    ZSetIndex.ensure(tx.shard_state, redis_key, prefix, data_path)
  end

  defp local_zset_tables?(%{zset_score_index: index, zset_score_lookup: lookup})
       when is_atom(index) and is_atom(lookup) do
    :ets.info(index) != :undefined and :ets.info(lookup) != :undefined
  end

  defp local_zset_tables?(_state), do: false

  defp tx_zset_index_clean? do
    map_size(Process.get(:tx_pending_values, %{})) == 0 and
      MapSet.size(Process.get(:tx_deleted_keys, MapSet.new())) == 0
  end

  defp stored_value_size(value) when is_binary(value), do: byte_size(value)
  defp stored_value_size(value) when is_integer(value), do: byte_size(Integer.to_string(value))
  defp stored_value_size(value) when is_float(value), do: byte_size(Float.to_string(value))
  defp stored_value_size(value), do: value |> to_string() |> byte_size()

  defp range_from_value(value, start_idx, end_idx) when is_binary(value),
    do: slice_binary_range(value, start_idx, end_idx)

  defp range_from_value(value, start_idx, end_idx) when is_integer(value),
    do: value |> Integer.to_string() |> slice_binary_range(start_idx, end_idx)

  defp range_from_value(value, start_idx, end_idx) when is_float(value),
    do: value |> Float.to_string() |> slice_binary_range(start_idx, end_idx)

  defp range_from_value(value, start_idx, end_idx),
    do: value |> to_string() |> slice_binary_range(start_idx, end_idx)

  defp slice_binary_range(value, start_idx, end_idx) do
    size = byte_size(value)
    start_norm = if start_idx < 0, do: max(size + start_idx, 0), else: start_idx
    end_norm = if end_idx < 0, do: size + end_idx, else: end_idx

    start_clamped = min(start_norm, size)
    end_clamped = min(end_norm, size - 1)

    if start_clamped > end_clamped do
      ""
    else
      binary_part(value, start_clamped, end_clamped - start_clamped + 1)
    end
  end

  defp tx_deleted?(key) do
    deleted = Process.get(:tx_deleted_keys, MapSet.new())
    MapSet.member?(deleted, key)
  end

  defp tx_pending_meta(key) do
    pending = Process.get(:tx_pending_values, %{})

    case Map.get(pending, key) do
      {value, 0} ->
        {value, 0}

      {value, exp} ->
        if exp > HLC.now_ms() do
          {value, exp}
        else
          tx_drop_pending(key)
          nil
        end

      nil ->
        nil
    end
  end

  defp tx_put_pending(key, value, expire_at_ms) do
    pending = Process.get(:tx_pending_values, %{})
    Process.put(:tx_pending_values, Map.put(pending, key, {value, expire_at_ms}))
    tx_undelete(key)
  end

  defp tx_drop_pending(key) do
    pending = Process.get(:tx_pending_values, %{})
    Process.put(:tx_pending_values, Map.delete(pending, key))
  end

  defp tx_mark_deleted(key) do
    deleted = Process.get(:tx_deleted_keys, MapSet.new())
    Process.put(:tx_deleted_keys, MapSet.put(deleted, key))
  end

  defp tx_undelete(key) do
    deleted = Process.get(:tx_deleted_keys, MapSet.new())

    if MapSet.member?(deleted, key) do
      Process.put(:tx_deleted_keys, MapSet.delete(deleted, key))
    end
  end

  # Read value from local ETS, cold-read fallback. Returns value or nil.
  defp local_read_value(tx, key) do
    case tx_pending_meta(key) do
      {value, _exp} -> normalize_get_value(value)
      nil -> tx |> local_read_value_from_ets(key) |> normalize_get_value()
    end
  end

  defp local_read_value_from_ets(tx, key) do
    case ShardETS.ets_lookup_warm(tx.shard_state, key) do
      {:hit, value, _exp} ->
        value

      :expired ->
        nil

      :miss ->
        nil
    end
  end

  defp local_exists?(tx, key) do
    cond do
      tx_deleted?(key) ->
        false

      tx_pending_meta(key) != nil ->
        true

      true ->
        case ShardETS.ets_lookup(tx.shard_state, key) do
          {:hit, _value, _exp} -> true
          {:cold, _fid, _off, _vsize, _exp} -> true
          :expired -> false
          :miss -> false
        end
    end
  end

  # Read {value, expire_at_ms} from local ETS, cold-read fallback. Returns {value, exp} or nil.
  defp local_read_meta(tx, key) do
    case tx_pending_meta(key) do
      {value, exp} -> {value, exp}
      nil -> local_read_meta_from_ets(tx, key)
    end
  end

  defp local_read_meta_from_ets(tx, key) do
    case ShardETS.ets_lookup_warm(tx.shard_state, key) do
      {:hit, value, exp} ->
        {value, exp}

      :expired ->
        nil

      :miss ->
        nil
    end
  end

  defp local_batch_get(tx, keys) do
    {local_entries, remote_entries} =
      keys
      |> Enum.with_index()
      |> Enum.reduce({[], []}, fn {key, index}, {local_acc, remote_acc} ->
        if local?(tx, key) do
          if tx_deleted?(key),
            do: {local_acc, remote_acc},
            else: {[{index, key} | local_acc], remote_acc}
        else
          {local_acc, [{index, key} | remote_acc]}
        end
      end)

    results =
      local_entries
      |> Enum.reverse()
      |> local_batch_results(%{}, fn entries ->
        entries
        |> Enum.map(fn {_index, key} -> key end)
        |> then(&local_batch_read_values(tx, &1, tx.shard_state.shard_data_path))
      end)

    results =
      remote_entries
      |> Enum.reverse()
      |> local_batch_results(results, fn entries ->
        entries
        |> Enum.map(fn {_index, key} -> key end)
        |> then(&Router.batch_get(tx.instance_ctx, &1))
      end)

    keys
    |> Enum.with_index()
    |> Enum.map(fn {_key, index} -> Map.get(results, index) end)
  end

  defp local_batch_results([], results, _read_fun), do: results

  defp local_batch_results(entries, results, read_fun) do
    entries
    |> Enum.zip(read_fun.(entries))
    |> Enum.reduce(results, fn {{index, _key}, value}, acc ->
      Map.put(acc, index, value)
    end)
  end

  defp local_batch_read_values(tx, keys, data_path) do
    tx
    |> local_batch_read_meta(keys, data_path)
    |> Enum.map(fn
      {value, _exp} -> normalize_get_value(value)
      nil -> nil
    end)
  end

  defp normalize_get_value(nil), do: nil
  defp normalize_get_value(value) when is_integer(value), do: Integer.to_string(value)

  defp normalize_get_value(value) when is_float(value),
    do: Ferricstore.Store.ValueCodec.format_float(value)

  defp normalize_get_value(value), do: value

  defp local_promoted_batch_read_values(tx, keys, dedicated_path) do
    local_promoted_batch_read(tx, keys, dedicated_path, &local_batch_read_values/3)
  end

  defp local_promoted_batch_read_meta(tx, keys, dedicated_path) do
    local_promoted_batch_read(tx, keys, dedicated_path, &local_batch_read_meta/3)
  end

  defp local_promoted_batch_read(tx, keys, dedicated_path, read_fun) do
    {shared_entries, dedicated_entries} =
      keys
      |> Enum.with_index()
      |> Enum.split_with(fn {key, _index} -> shared_log_compound_key?(key) end)

    %{}
    |> local_promoted_batch_partition(
      tx,
      shared_entries,
      tx.shard_state.shard_data_path,
      read_fun
    )
    |> local_promoted_batch_partition(tx, dedicated_entries, dedicated_path, read_fun)
    |> then(fn results ->
      keys
      |> Enum.with_index()
      |> Enum.map(fn {_key, index} -> Map.get(results, index) end)
    end)
  end

  defp local_promoted_batch_partition(results, _tx, [], _data_path, _read_fun), do: results

  defp local_promoted_batch_partition(results, tx, entries, data_path, read_fun) do
    partition_keys = Enum.map(entries, fn {key, _index} -> key end)

    entries
    |> Enum.zip(read_fun.(tx, partition_keys, data_path))
    |> Enum.reduce(results, fn {{_key, index}, value}, acc -> Map.put(acc, index, value) end)
  end

  defp local_batch_read_meta(tx, keys, data_path) do
    now = HLC.now_ms()

    {warm_results, cold_reads} =
      keys
      |> Enum.with_index()
      |> Enum.reduce({%{}, []}, fn {key, index}, {results, cold} ->
        case tx_pending_meta(key) do
          {value, exp} ->
            {Map.put(results, index, {value, exp}), cold}

          nil ->
            local_batch_collect_ets(tx, key, index, data_path, now, results, cold)
        end
      end)

    results = local_batch_read_cold(tx, warm_results, Enum.reverse(cold_reads))

    keys
    |> Enum.with_index()
    |> Enum.map(fn {_key, index} -> Map.get(results, index) end)
  end

  defp local_batch_collect_ets(tx, key, index, data_path, now, results, cold) do
    case :ets.lookup(tx.shard_state.keydir, key) do
      [{^key, value, 0, _lfu, _fid, _off, _vsize}] when value != nil ->
        {Map.put(results, index, {value, 0}), cold}

      [{^key, value, exp, _lfu, _fid, _off, _vsize}] when exp > now and value != nil ->
        {Map.put(results, index, {value, exp}), cold}

      [{^key, nil, 0, _lfu, fid, off, vsize}]
      when valid_cold_location(fid, off, vsize) ->
        path = ShardETS.file_path(data_path, fid)
        {results, [{index, key, path, fid, off, vsize, 0} | cold]}

      [{^key, nil, exp, _lfu, fid, off, vsize}]
      when exp > now and valid_cold_location(fid, off, vsize) ->
        path = ShardETS.file_path(data_path, fid)
        {results, [{index, key, path, fid, off, vsize, exp} | cold]}

      [{^key, _value, _exp, _lfu, _fid, _off, _vsize}] ->
        ShardETS.ets_delete_key(tx.shard_state, key)
        {results, cold}

      _ ->
        {results, cold}
    end
  rescue
    ArgumentError ->
      {results, cold}
  end

  defp local_batch_read_cold(_tx, results, []), do: results

  defp local_batch_read_cold(tx, results, cold_reads) do
    {unique_reads, fanout_indexes} = dedupe_local_batch_cold_reads(cold_reads)
    unique_results = read_unique_local_batch_cold(tx, unique_reads)

    Enum.reduce(fanout_indexes, results, fn {original_index, unique_index}, acc ->
      case Map.fetch(unique_results, unique_index) do
        {:ok, value} -> Map.put(acc, original_index, value)
        :error -> acc
      end
    end)
  end

  defp dedupe_local_batch_cold_reads(cold_reads) do
    {unique_reads, _index_by_location, fanout_indexes} =
      Enum.reduce(cold_reads, {[], %{}, []}, fn read, {unique_acc, index_acc, fanout_acc} ->
        original_index = elem(read, 0)
        location = local_batch_cold_read_location(read)

        case Map.fetch(index_acc, location) do
          {:ok, unique_index} ->
            {unique_acc, index_acc, [{original_index, unique_index} | fanout_acc]}

          :error ->
            unique_index = map_size(index_acc)
            unique_read = put_elem(read, 0, unique_index)

            {[unique_read | unique_acc], Map.put(index_acc, location, unique_index),
             [{original_index, unique_index} | fanout_acc]}
        end
      end)

    {Enum.reverse(unique_reads), Enum.reverse(fanout_indexes)}
  end

  defp local_batch_cold_read_location({_index, key, path, _fid, off, _vsize, _exp}) do
    {path, off, key}
  end

  defp read_unique_local_batch_cold(tx, cold_reads) do
    locations =
      Enum.map(cold_reads, fn {_index, key, path, _fid, off, _vsize, _exp} -> {path, off, key} end)

    case ColdRead.pread_batch_keyed(locations, @cold_read_timeout_ms) do
      {:ok, values} when is_list(values) and length(values) == length(cold_reads) ->
        cold_reads
        |> Enum.zip(local_materialize_blob_values(tx, values))
        |> Enum.reduce(%{}, fn
          {{index, key, _path, fid, off, vsize, exp}, {:ok, materialized}}, acc ->
            ShardETS.cold_read_warm_ets(
              tx.shard_state,
              key,
              materialized,
              exp,
              fid,
              off,
              vsize
            )

            Map.put(acc, index, {materialized, exp})

          {{_index, _key, path, _fid, _off, _vsize, _exp}, {:error, reason}}, acc ->
            ColdRead.emit_pread_error(path, reason)
            acc

          {_read, _missing_or_error}, acc ->
            acc
        end)

      {:ok, _bad_values} ->
        emit_local_batch_cold_errors(cold_reads, :batch_result_length_mismatch)
        %{}

      {:error, reason} ->
        emit_local_batch_cold_errors(cold_reads, reason)
        %{}
    end
  end

  defp local_materialize_blob_values(tx, values) do
    {binary_values, indexed_results} =
      values
      |> Enum.with_index()
      |> Enum.reduce({[], %{}}, fn
        {value, index}, {binary_values, indexed_results} when is_binary(value) ->
          {[{index, value} | binary_values], indexed_results}

        {{:error, reason}, index}, {binary_values, indexed_results} ->
          {binary_values, Map.put(indexed_results, index, {:error, reason})}

        {_unexpected, index}, {binary_values, indexed_results} ->
          {binary_values, Map.put(indexed_results, index, :skip)}
      end)

    indexed_results =
      if binary_values == [] do
        indexed_results
      else
        ordered_values = Enum.reverse(binary_values)

        ordered_values
        |> Enum.map(fn {_index, value} -> value end)
        |> then(fn values ->
          BlobValue.maybe_materialize_many(
            tx.shard_state.data_dir,
            tx.shard_index,
            BlobValue.threshold(tx.instance_ctx),
            values
          )
        end)
        |> then(fn materialized -> Enum.zip(ordered_values, materialized) end)
        |> Enum.reduce(indexed_results, fn {{index, _value}, result}, acc ->
          Map.put(acc, index, result)
        end)
      end

    values
    |> Enum.with_index()
    |> Enum.map(fn {_value, index} -> Map.fetch!(indexed_results, index) end)
  end

  defp local_materialize_blob_value(tx, value) do
    BlobValue.maybe_materialize(
      tx.shard_state.data_dir,
      tx.shard_index,
      BlobValue.threshold(tx.instance_ctx),
      value
    )
  end

  defp local_cold_value_size(tx, key, fid, off, vsize) do
    cond do
      valid_waraft_segment_location(fid, off, vsize) ->
        Router.value_size(tx.instance_ctx, key)

      BlobValue.threshold(tx.instance_ctx) > 0 and BlobRef.encoded_size?(vsize) ->
        path = ShardETS.file_path(tx.shard_state.shard_data_path, fid)

        case ColdRead.pread_keyed(path, off, key, @cold_read_timeout_ms) do
          {:ok, value} ->
            case BlobRef.decode(value) do
              {:ok, %BlobRef{size: logical_size}} -> logical_size
              :error -> vsize
            end

          {:error, reason} ->
            ColdRead.emit_pread_error(path, reason)
            vsize
        end

      true ->
        vsize
    end
  end

  defp emit_local_batch_cold_errors(cold_reads, reason) do
    cold_reads
    |> Enum.reduce(%{}, fn {_index, _key, path, _fid, _off, _vsize, _exp}, acc ->
      Map.update(acc, path, 1, &(&1 + 1))
    end)
    |> Enum.each(fn {path, count} -> ColdRead.emit_pread_error(path, reason, count) end)
  end

  defp local_set(tx, key, value, opts) do
    get? = Map.get(opts, :get, false)
    current = local_set_current_meta(tx, key, get?)

    {old_value, effective_expire} =
      case current do
        nil ->
          {nil, opts.expire_at_ms}

        {old_val, old_exp} ->
          {old_val, if(opts.keepttl, do: old_exp, else: opts.expire_at_ms)}
      end

    skip? =
      cond do
        opts.nx and current != nil -> true
        opts.xx and current == nil -> true
        true -> false
      end

    if skip? do
      if get?, do: old_value, else: nil
    else
      ShardETS.ets_insert(tx.shard_state, key, value, effective_expire)
      tx_put_pending(key, value, effective_expire)
      tx_undelete(key)
      send(self(), {:tx_pending_write, key, value, effective_expire})
      if get?, do: old_value, else: :ok
    end
  end

  defp fallback_set(store, key, value, opts) do
    get? = Map.get(opts, :get, false)
    current = fallback_set_current_meta(store, key, get?, opts.keepttl)

    {old_value, effective_expire} =
      case current do
        nil ->
          {nil, opts.expire_at_ms}

        {old_val, old_exp} ->
          {old_val, if(opts.keepttl, do: old_exp, else: opts.expire_at_ms)}
      end

    skip? =
      cond do
        opts.nx and exists?(store, key) -> true
        opts.xx and not exists?(store, key) -> true
        true -> false
      end

    if skip? do
      if get?, do: old_value, else: nil
    else
      put(store, key, value, effective_expire)
      if get?, do: old_value, else: :ok
    end
  end

  defp local_set_current_meta(tx, key, true), do: local_read_meta(tx, key)

  defp local_set_current_meta(tx, key, false) do
    if tx_deleted?(key) do
      nil
    else
      case tx_pending_meta(key) do
        {_value, _exp} = pending -> pending_expire_meta(pending)
        nil -> ets_expire_meta(tx, key)
      end
    end
  end

  defp pending_expire_meta({_value, exp}), do: {nil, exp}

  defp ets_expire_meta(tx, key) do
    case ShardETS.ets_lookup(tx.shard_state, key) do
      {:hit, _value, exp} -> {nil, exp}
      {:cold, _fid, _off, _vsize, exp} -> {nil, exp}
      _ -> nil
    end
  end

  defp fallback_set_current_meta(store, key, true, _keepttl), do: get_meta(store, key)

  defp fallback_set_current_meta(store, key, false, true) do
    case expire_at_ms(store, key) do
      nil -> nil
      exp -> {nil, exp}
    end
  end

  defp fallback_set_current_meta(_store, _key, false, false), do: nil

  # Read {value, expire_at_ms} for local transaction read-modify-write ops.
  # This preserves Redis' TTL semantics for INCR/APPEND/SETRANGE while keeping
  # the same one-lookup hot path as the old value-only helper.
  defp local_read_meta_for_rmw(tx, key) do
    case tx_pending_meta(key) do
      {value, exp} -> {value, exp}
      nil -> local_read_meta_for_rmw_from_ets(tx, key)
    end
  end

  defp local_read_meta_for_rmw_from_ets(tx, key) do
    case ShardETS.ets_lookup_warm(tx.shard_state, key) do
      {:hit, value, exp} ->
        {value, exp}

      :expired ->
        {nil, 0}

      :miss ->
        {nil, 0}
    end
  end

  # Read value for read-modify-write ops that intentionally discard TTL
  # (GETSET) or delete/change it explicitly (GETDEL/GETEX).
  defp local_read_value_for_rmw(tx, key) do
    case tx_pending_meta(key) do
      {value, _exp} -> value
      nil -> local_read_value_for_rmw_from_ets(tx, key)
    end
  end

  defp local_read_value_for_rmw_from_ets(tx, key) do
    case ShardETS.ets_lookup_warm(tx.shard_state, key) do
      {:hit, value, _exp} ->
        value

      :expired ->
        nil

      :miss ->
        nil
    end
  end

  defp promoted_path(%LocalTxStore{} = tx, redis_key) do
    case tx.shard_state.promoted_instances do
      %{^redis_key => %{path: path}} -> path
      _ -> nil
    end
  end

  defp shared_log_compound_key?(<<"T:", _rest::binary>>), do: true
  defp shared_log_compound_key?(<<"PM:", _rest::binary>>), do: true
  defp shared_log_compound_key?(_key), do: false

  defp tx_compound_write_message(tx, redis_key, compound_key, value, expire_at_ms) do
    case promoted_path(tx, redis_key) do
      nil ->
        {:tx_pending_write, compound_key, value, expire_at_ms}

      _dedicated_path ->
        if shared_log_compound_key?(compound_key) do
          {:tx_pending_write, compound_key, value, expire_at_ms}
        else
          {:tx_pending_compound_write, redis_key, compound_key, value, expire_at_ms}
        end
    end
  end

  defp tx_compound_delete_message(tx, redis_key, compound_key) do
    case promoted_path(tx, redis_key) do
      nil ->
        {:tx_pending_delete, compound_key}

      _dedicated_path ->
        if shared_log_compound_key?(compound_key) do
          {:tx_pending_delete, compound_key}
        else
          {:tx_pending_compound_delete, redis_key, compound_key}
        end
    end
  end

  defp local_promoted_read_value(tx, compound_key, dedicated_path) do
    case tx_pending_meta(compound_key) ||
           local_promoted_read_meta(tx, compound_key, dedicated_path) do
      {value, _exp} -> value
      nil -> nil
    end
  end

  defp local_promoted_read_meta(tx, compound_key, dedicated_path) do
    case tx_pending_meta(compound_key) do
      nil -> local_promoted_read_meta_from_ets(tx, compound_key, dedicated_path)
      meta -> meta
    end
  end

  defp local_promoted_read_meta_from_ets(tx, compound_key, dedicated_path) do
    now = HLC.now_ms()
    keydir = tx.shard_state.keydir

    case :ets.lookup(keydir, compound_key) do
      [{^compound_key, value, 0, _lfu, _fid, _off, _vsize}] when value != nil ->
        {value, 0}

      [{^compound_key, value, exp, _lfu, _fid, _off, _vsize}] when exp > now and value != nil ->
        {value, exp}

      [{^compound_key, nil, 0, _lfu, fid, off, vsize}]
      when valid_cold_location(fid, off, vsize) ->
        read_promoted_cold_value(tx, compound_key, dedicated_path, fid, off, vsize, 0)

      [{^compound_key, nil, exp, _lfu, fid, off, vsize}]
      when exp > now and valid_cold_location(fid, off, vsize) ->
        read_promoted_cold_value(tx, compound_key, dedicated_path, fid, off, vsize, exp)

      [{^compound_key, _value, _exp, _lfu, _fid, _off, _vsize}] ->
        ShardETS.ets_delete_key(tx.shard_state, compound_key)
        nil

      _ ->
        nil
    end
  end

  defp merge_tx_pending_prefix(results, prefix) do
    deleted = Process.get(:tx_deleted_keys, MapSet.new())
    prefix_len = byte_size(prefix)
    now_ms = HLC.now_ms()

    base =
      results
      |> Enum.reject(fn {field, _value} -> MapSet.member?(deleted, prefix <> field) end)
      |> Map.new()

    Process.get(:tx_pending_values, %{})
    |> Enum.reduce(base, fn
      {key, {value, exp}}, acc when is_binary(key) and byte_size(key) >= prefix_len ->
        if String.starts_with?(key, prefix) and not MapSet.member?(deleted, key) and
             (exp == 0 or exp > now_ms) do
          field =
            case :binary.split(key, <<0>>) do
              [_pre, sub] -> sub
              _ -> key
            end

          Map.put(acc, field, value)
        else
          acc
        end

      _other, acc ->
        acc
    end)
    |> Map.to_list()
  end

  defp read_promoted_cold_value(tx, compound_key, dedicated_path, fid, off, vsize, exp) do
    path = ShardETS.file_path(dedicated_path, fid)

    case read_cold_async(path, off, compound_key) do
      {:ok, value} ->
        case local_materialize_blob_value(tx, value) do
          {:ok, materialized} ->
            ShardETS.cold_read_warm_ets(
              tx.shard_state,
              compound_key,
              materialized,
              exp,
              fid,
              off,
              vsize
            )

            {materialized, exp}

          {:error, reason} ->
            ColdRead.emit_pread_error(path, reason)
            nil
        end

      {:error, reason} ->
        ColdRead.emit_pread_error(path, reason)
        nil

      _ ->
        nil
    end
  end

  defp read_cold_async(path, offset, key) do
    Ferricstore.Store.ColdRead.pread_keyed(path, offset, key, @cold_read_timeout_ms)
  end
end