acts 0.25.0

use super::map_db_err;
use crate::store::{
    DbCollection, DbCollectionIden, Expr, ExprOp, Filter, FilterExpr, KvStore, OrderBy, PageData,
    Query, ScanOperation, ScanOptions, Sort, StoreBatchOp, query::FilterType,
};
use crate::utils::consts::{KEY_SEP, KEY_SEP_SUCC};
use crate::{ActError, Result};
use parking_lot::Mutex;
use serde::{Serialize, de::DeserializeOwned};
use serde_json::Value as JsonValue;
use std::sync::LazyLock;
use std::{
    cmp::Ordering,
    collections::{BTreeMap, HashMap, HashSet},
    fmt::Debug,
    marker::PhantomData,
    sync::Arc,
};
use tokio::sync::{Mutex as AsyncMutex, OwnedMutexGuard};

pub struct KvCollection<T> {
    prefix: String,
    kv: Arc<dyn KvStore>,
    _t: PhantomData<T>,
}

/// Process-wide registry of the per-document mutation locks.
///
/// A document's data row and its index rows stay consistent only if no other
/// mutation reads the stored row in between: `update`/`delete` read the
/// document to compute which index rows to drop, then commit the data row and
/// the new index rows as one [`KvStore::batch`]. Two mutations of the same id
/// that both read the same old document each compute their drops from that
/// version, so whichever batch lands second leaves index rows of a value the
/// data row no longer holds (a query returns a phantom id) or drops index rows
/// of the value it does hold (a query misses it).
///
/// [`lock_docs`] serializes every read-modify-write of a document from its
/// read until its batch has been applied. The guarantee ends at the process
/// boundary — the store's contract is one writer per database — and both edges
/// of it are worth stating, because neither is enforced anywhere:
///
/// - The registry is a process-wide `static`, so every engine and every store
///   handle in this process share one table. Keys are full document keys, so
///   two collections with the same prefix and id (a second engine, a second
///   handle over the same keys) serialize on them even when the databases
///   behind them are unrelated. That costs contention and never correctness —
///   a shared table can wrongly exclude, not wrongly admit — and it is what
///   makes two engines over one database in a single process safe without any
///   further coordination.
/// - There is no cross-process mutex. Two processes writing one database (a
///   multi-instance deployment over a remote postgres/redis/nats backend, or a
///   second server on the same sqlite/sled file) each hold a private registry,
///   so the read in `update_ops`/`delete_ops` and the batch computed from it
///   can interleave and both inconsistent outcomes above are reachable. Such a
///   deployment needs coordination the engine does not provide: one writer per
///   database (the supported deployment), a backend conditional write (a
///   compare-and-swap on the row), or a backend lock held across the read —
///   [`KvStore::batch`] makes one write atomic, not a read against another
///   process's write.
static DOC_LOCKS: LazyLock<DocLockRegistry> = LazyLock::new(DocLockRegistry::default);

#[derive(Default)]
struct DocLockRegistry {
    /// Keyed by the full document key. Two collections over the same key (or
    /// two stores with equal keys) share one lock — conservative where the
    /// stores are unrelated, and required when two handles address the same
    /// database. An entry is dropped once the last holder released it, so a
    /// long-running engine does not accumulate one lock per document id it has
    /// ever touched: the map holds only the documents being mutated right now.
    entries: Mutex<HashMap<String, Arc<AsyncMutex<()>>>>,
}

/// The document locks held by one read-modify-write, released on drop.
///
/// The locks MUST be held until the batch computed from the locked documents
/// has been applied.
pub(crate) struct DocLocks {
    locks: Vec<DocLock>,
}

struct DocLock {
    registry: &'static DocLockRegistry,
    key: String,
    /// `None` only inside `Drop`, which drops the reference before deciding
    /// whether the registry entry is unreferenced.
    entry: Option<Arc<AsyncMutex<()>>>,
    guard: Option<OwnedMutexGuard<()>>,
}

/// Lock every document in `keys` (full document keys) for a read-modify-write.
///
/// The keys are acquired in ascending order and deduplicated, so two mutations
/// touching the same documents in a different discovery order cannot deadlock
/// against each other.
pub(crate) async fn lock_docs(keys: impl IntoIterator<Item = String>) -> DocLocks {
    let mut keys: Vec<String> = keys.into_iter().collect();
    keys.sort_unstable();
    keys.dedup();

    let registry: &'static DocLockRegistry = &DOC_LOCKS;
    let mut locks = Vec::with_capacity(keys.len());
    for key in keys {
        locks.push(DocLock::acquire(registry, key).await);
    }
    DocLocks { locks }
}

impl DocLocks {
    /// Lock the documents a mutation extends to while it already holds the
    /// lock of the document that owns them.
    ///
    /// Only a model row and its own trigger rows use this: the model is locked
    /// first, the trigger rows it declares (and the stale ones it drops)
    /// after, and no code path takes a trigger lock before a model lock, so
    /// this nesting cannot form a wait cycle. Independent documents must go to
    /// one [`lock_docs`] call instead.
    pub(crate) async fn lock_more(&mut self, keys: impl IntoIterator<Item = String>) {
        self.locks.extend(lock_docs(keys).await.locks);
    }
}

impl DocLock {
    async fn acquire(registry: &'static DocLockRegistry, key: String) -> Self {
        let entry = {
            let mut entries = registry.entries.lock();
            entries.entry(key.clone()).or_default().clone()
        };
        let guard = entry.clone().lock_owned().await;
        Self {
            registry,
            key,
            entry: Some(entry),
            guard: Some(guard),
        }
    }
}

impl Drop for DocLock {
    fn drop(&mut self) {
        // Release the mutex before the registry entry can go away: a mutation
        // looking the entry up between these steps must still observe it
        // locked.
        drop(self.guard.take());
        drop(self.entry.take());
        let mut entries = self.registry.entries.lock();
        // The registry's own reference is the last one exactly when no task
        // holds or awaits this lock any more; a waiter that cloned the entry
        // before this point keeps it alive and drops it after its own release.
        if entries
            .get(&self.key)
            .is_some_and(|e| Arc::strong_count(e) == 1)
        {
            entries.remove(&self.key);
        }
    }
}

impl<T> KvCollection<T> {
    pub fn new(prefix: &str, kv: Arc<dyn KvStore>) -> Self {
        Self {
            prefix: prefix.to_string(),
            kv,
            _t: PhantomData,
        }
    }

    pub(crate) fn data_key(&self, id: &str) -> String {
        format!("{}{}id{}{}", self.prefix, KEY_SEP, KEY_SEP, id)
    }

    fn data_prefix(&self) -> String {
        format!("{}{}id{}", self.prefix, KEY_SEP, KEY_SEP)
    }

    fn index_keys(&self, json: &JsonValue, id: &str) -> Vec<String>
    where
        T: DbCollectionIden,
    {
        let fields = T::indexed_fields();
        if fields.is_empty() {
            return Vec::new();
        }
        let mut keys = Vec::with_capacity(fields.len());
        for field in fields {
            if let Some(val) = json.get(field) {
                let val_str = json_value_to_key_str(val);
                keys.push(format!(
                    "{}{}{}{}{}{}{}",
                    self.prefix, KEY_SEP, field, KEY_SEP, val_str, KEY_SEP, id
                ));
            }
        }
        keys
    }

    async fn read_json(&self, id: &str) -> Result<Option<JsonValue>> {
        let key = self.data_key(id);
        self.kv
            .one(&key)
            .await?
            .map(|data| serde_json::from_slice(&data).map_err(map_db_err))
            .transpose()
    }

    /// Read documents in the caller-supplied order with one backend batch
    /// operation where one is available.
    async fn read_json_many(&self, ids: &[String]) -> Result<Vec<JsonValue>> {
        if ids.is_empty() {
            return Ok(Vec::new());
        }

        let keys: Vec<String> = ids.iter().map(|id| self.data_key(id)).collect();
        let values = self.kv.many(&keys).await?;
        let mut docs = Vec::with_capacity(values.len());
        for data in values.into_iter().flatten() {
            docs.push(serde_json::from_slice(&data).map_err(map_db_err)?);
        }
        Ok(docs)
    }

    /// The ids matching `filter` — every id of the collection when `None` —
    /// as a set.
    ///
    /// `order_by` only selects the scan direction of the index path (see
    /// `expr_ids`); the resulting set is the same either way.
    async fn filter_id_set(
        &self,
        filter: Option<&Filter>,
        order_by: &[OrderBy],
    ) -> Result<HashSet<String>>
    where
        T: DbCollectionIden,
    {
        match filter {
            Some(filter) => self.filter_ids(filter, T::indexed_fields(), order_by).await,
            None => {
                // No filter — scan all data entries to collect all IDs
                let scan_key = self.data_prefix();
                let options = ScanOptions::new(ScanOperation::Eq, scan_key.clone(), false);
                let entries = self.kv.scan_prefix(&scan_key, options).await?;
                Ok(entries
                    .iter()
                    .filter_map(|(key, _)| key.strip_prefix(&scan_key).map(str::to_string))
                    .collect())
            }
        }
    }

    /// The mutations of an insert: the data row and the index rows of `data`,
    /// with no cleanup of a stored version. Only valid when the document is
    /// known to be absent — a row already stored under the same id would keep
    /// its index rows, so `create`/`update` (which drop them) are the writes
    /// for any id that may exist.
    ///
    /// Lets a caller fold several document writes — e.g. a model and its
    /// trigger rows on `deploy` — into one atomic [`KvStore::batch`]. The
    /// caller MUST hold the lock of every document it writes (see
    /// [`lock_docs`]) until that batch is applied, so the absence it relies on
    /// cannot change underneath it.
    pub(crate) fn create_ops(&self, data: &T) -> Result<Vec<StoreBatchOp>>
    where
        T: DbCollectionIden + Serialize,
    {
        let json = serde_json::to_value(data).map_err(map_db_err)?;
        self.create_ops_json(&json)
    }

    fn create_ops_json(&self, json: &JsonValue) -> Result<Vec<StoreBatchOp>>
    where
        T: DbCollectionIden,
    {
        let id = extract_id(json)?;
        let bytes = serde_json::to_vec(json).map_err(map_db_err)?;

        let mut ops = Vec::with_capacity(1 + T::indexed_fields().len());
        ops.push(StoreBatchOp::Put {
            key: self.data_key(&id),
            value: bytes,
        });
        for idx_key in self.index_keys(json, &id) {
            ops.push(StoreBatchOp::Put {
                key: idx_key,
                value: vec![],
            });
        }
        Ok(ops)
    }

    /// The mutations [`DbCollection::update`] would apply: drop the old index
    /// keys the new document no longer carries (keys re-created with the same
    /// value are left alone — a delete+put of one key is a no-op), then write
    /// the data row and the new index rows.
    ///
    /// The caller MUST hold the lock of every document it writes (see
    /// [`lock_docs`]) until that batch is applied: the dropped keys are the
    /// ones the stored document holds at this moment.
    pub(crate) async fn update_ops(&self, data: &T) -> Result<Vec<StoreBatchOp>>
    where
        T: DbCollectionIden + Serialize,
    {
        let new_json = serde_json::to_value(data).map_err(map_db_err)?;
        self.update_ops_json(&new_json).await
    }

    async fn update_ops_json(&self, new_json: &JsonValue) -> Result<Vec<StoreBatchOp>>
    where
        T: DbCollectionIden,
    {
        let id = extract_id(new_json)?;
        let new_bytes = serde_json::to_vec(new_json).map_err(map_db_err)?;
        let new_index = self.index_keys(new_json, &id);
        let mut ops = Vec::with_capacity(new_index.len() + 1);
        if let Some(old_json) = self.read_json(&id).await? {
            let new_keys: HashSet<&str> = new_index.iter().map(String::as_str).collect();
            for idx_key in self.index_keys(&old_json, &id) {
                if !new_keys.contains(idx_key.as_str()) {
                    ops.push(StoreBatchOp::Delete { key: idx_key });
                }
            }
        }
        ops.push(StoreBatchOp::Put {
            key: self.data_key(&id),
            value: new_bytes,
        });
        for idx_key in new_index {
            ops.push(StoreBatchOp::Put {
                key: idx_key,
                value: vec![],
            });
        }
        Ok(ops)
    }

    /// The mutations [`DbCollection::delete`] would apply: every index row of
    /// the current document, then the data row itself.
    ///
    /// The caller MUST hold the lock of the document (see [`lock_docs`]) until
    /// that batch is applied: the index rows are the ones the stored document
    /// holds at this moment.
    pub(crate) async fn delete_ops(&self, id: &str) -> Result<Vec<StoreBatchOp>>
    where
        T: DbCollectionIden,
    {
        let mut ops = Vec::new();
        if let Some(old_json) = self.read_json(id).await? {
            for idx_key in self.index_keys(&old_json, id) {
                ops.push(StoreBatchOp::Delete { key: idx_key });
            }
        }
        ops.push(StoreBatchOp::Delete {
            key: self.data_key(id),
        });
        Ok(ops)
    }

    /// Compute the set of IDs matching a single expression.
    async fn expr_ids(
        &self,
        expr: &Expr,
        indexed: &[&str],
        order_by: &[OrderBy],
    ) -> Result<HashSet<String>> {
        Self::validate_expr(expr)?;
        // Match is substring matching (contains), which an index prefix scan
        // cannot serve, and range/inequality scans are exact only for the
        // fixed-width, order-preserving numeric encoding (see
        // `is_index_exact`); everything else falls through to the
        // non-indexed path.
        if indexed.contains(&expr.key.as_str()) && Self::is_index_exact(expr) {
            // Determine scan direction from order_by for this expression's field
            let is_rev = order_by
                .iter()
                .find(|ob| ob.field == expr.key)
                .map(|ob| ob.order == Sort::Desc)
                .unwrap_or(false);

            // field_prefix bounds the scan to this field: {prefix}-{field}-
            let field_prefix = format!("{}{}{}{}", self.prefix, KEY_SEP, expr.key, KEY_SEP);

            // All keys of one value `v` are `..-{v}-{id}`: with `-` excluded
            // from the value encoding that group is contiguous and sorted by
            // value. `lower(e)` is the first possible key of group `e`;
            // `after(e)` is an exclusive upper bound that covers the whole
            // group and nothing above it (`KEY_SEP_SUCC` is one byte above
            // `KEY_SEP`, below every character that can follow a value
            // segment).
            let lower = |e: &str| format!("{}{}", field_prefix, e);
            let after = |e: &str| format!("{}{}{}", field_prefix, e, KEY_SEP_SUCC);

            // `eq_prefix` is the exact value-key prefix when the scan can only
            // return keys of that value (Eq); otherwise ids are recovered by
            // cutting the trailing `-{id}` segment.
            let (scan_op, scan_key, eq_prefix) = match expr.op {
                ExprOp::EQ => {
                    let v = json_value_to_key_str(&expr.value);
                    let vk = format!("{}{}{}", field_prefix, v, KEY_SEP);
                    (ScanOperation::Eq, vk.clone(), Some(vk))
                }
                ExprOp::NE => {
                    let v = json_value_to_key_str(&expr.value);
                    (
                        ScanOperation::Ne,
                        format!("{}{}{}", field_prefix, v, KEY_SEP),
                        None,
                    )
                }
                ExprOp::GT => (
                    ScanOperation::Range {
                        lower: Some(after(&json_value_to_key_str(&expr.value))),
                        upper: None,
                    },
                    field_prefix.clone(),
                    None,
                ),
                ExprOp::GE => (
                    ScanOperation::Range {
                        lower: Some(lower(&json_value_to_key_str(&expr.value))),
                        upper: None,
                    },
                    field_prefix.clone(),
                    None,
                ),
                ExprOp::LT => (
                    ScanOperation::Range {
                        lower: None,
                        upper: Some(lower(&json_value_to_key_str(&expr.value))),
                    },
                    field_prefix.clone(),
                    None,
                ),
                ExprOp::LE => (
                    ScanOperation::Range {
                        lower: None,
                        upper: Some(after(&json_value_to_key_str(&expr.value))),
                    },
                    field_prefix.clone(),
                    None,
                ),
                ExprOp::Between => {
                    let empty = vec![];
                    let arr = expr.value.as_array().unwrap_or(&empty);
                    if arr.is_empty() || arr.len() < 2 {
                        return Err(ActError::Store(
                            "Between operator requires an array of two values".to_string(),
                        ));
                    }
                    let from = json_value_to_key_str(&arr[0]);
                    let to = json_value_to_key_str(&arr[1]);
                    (
                        ScanOperation::Range {
                            lower: Some(lower(&from)),
                            upper: Some(after(&to)),
                        },
                        field_prefix.clone(),
                        None,
                    )
                }
                ExprOp::In => {
                    let empty = vec![];
                    let arr = expr.value.as_array().unwrap_or(&empty);
                    if arr.is_empty() {
                        return Err(ActError::Store(
                            "In operator requires a non-empty array".to_string(),
                        ));
                    }
                    let values: Vec<String> = arr
                        .iter()
                        .map(|val| {
                            let v_str = json_value_to_key_str(val);
                            format!("{}{}{}", field_prefix, v_str, KEY_SEP)
                        })
                        .collect();
                    (ScanOperation::In { values }, field_prefix.clone(), None)
                }
                ExprOp::Match => unreachable!("Match is excluded by is_index_exact"),
            };

            let options = ScanOptions::new(scan_op, field_prefix.clone(), is_rev);
            let entries = self.kv.scan_prefix(&scan_key, options).await?;

            let ids: HashSet<String> = match eq_prefix {
                // Eq: every returned key starts with the value-key prefix
                Some(vk) => entries
                    .iter()
                    .filter_map(|(key, _)| key.strip_prefix(&vk).map(str::to_string))
                    .collect(),
                // Other ops (In/Ne/range/Between): skip the field prefix and
                // the value segment. The value can never contain `KEY_SEP`
                // (it is escaped away by the key encoding), while the id MAY
                // contain it (e.g. `p-acked`) — so the FIRST separator after
                // the field prefix is the value/id boundary; the last one
                // would sit inside the id and truncate it.
                None => entries
                    .iter()
                    .filter_map(|(key, _)| {
                        let rest = key.strip_prefix(&field_prefix)?;
                        let sep_pos = rest.find(KEY_SEP)?;
                        Some(rest[sep_pos + KEY_SEP.len()..].to_string())
                    })
                    .collect(),
            };
            Ok(ids)
        } else {
            // Fallback: scan all data entries and filter in-memory
            let scan_key = self.data_prefix();
            let options = ScanOptions::new(ScanOperation::Eq, scan_key.clone(), false);
            let entries = self.kv.scan_prefix(&scan_key, options).await?;
            let ids: HashSet<String> = entries
                .iter()
                .filter_map(|(key, bytes)| {
                    let id = key.strip_prefix(&scan_key)?;
                    let v: JsonValue = serde_json::from_slice(bytes).ok()?;
                    if let Some(field_val) = v.get(&expr.key)
                        && expr.op(field_val, &expr.value)
                    {
                        return Some(id.to_string());
                    }
                    None
                })
                .collect();
            Ok(ids)
        }
    }

    /// Whether an expression over an indexed field can be answered exactly by
    /// index-key scans.
    ///
    /// Eq/Ne/In only rely on prefix matching over the injective value
    /// encoding, so any JSON value type is safe. Range and inequality scans
    /// compare encoded value segments lexicographically, which is exact only
    /// for the fixed-width numeric encoding: non-negative i64/u64 padded to
    /// 20 digits. Strings (escaped characters sort outside alphanumerics) and
    /// negative integers (zero-padding reverses their order) must not use the
    /// index path, so they fall back to the full data scan.
    fn is_index_exact(expr: &Expr) -> bool {
        fn orderable(v: &JsonValue) -> bool {
            v.as_i64().is_some_and(|i| i >= 0) || v.as_u64().is_some()
        }
        match &expr.op {
            ExprOp::EQ | ExprOp::NE | ExprOp::In => true,
            ExprOp::GT | ExprOp::GE | ExprOp::LT | ExprOp::LE => orderable(&expr.value),
            ExprOp::Between => match expr.value.as_array() {
                Some(arr) if arr.len() == 2 => orderable(&arr[0]) && orderable(&arr[1]),
                _ => false,
            },
            ExprOp::Match => false,
        }
    }

    /// Walk a single FilterExpr node and return matching IDs.
    async fn filter_expr_ids(
        &self,
        filter_expr: &FilterExpr,
        indexed: &[&str],
        order_by: &[OrderBy],
    ) -> Result<HashSet<String>> {
        match filter_expr {
            FilterExpr::Expr(expr) => self.expr_ids(expr, indexed, order_by).await,
            // boxed: `filter_ids` recurses back here (mutual async recursion)
            FilterExpr::Filter(filter) => {
                Box::pin(self.filter_ids(filter, indexed, order_by)).await
            }
        }
    }

    /// Order matching IDs by an indexed field without reading document bodies.
    ///
    /// Index entries are stored as `{field}-{value}-{id}`. The intended order
    /// is encoded value order and, within one value, id order -- the same
    /// stable tie-break used by [`cmp_order_docs`]. That order is rebuilt here
    /// rather than taken from the scan: [`KvStore::scan_prefix`] promises
    /// nothing about entry order (Redis `SCAN` returns keys in arbitrary
    /// order), so entries arrive neither value-sorted nor grouped. Sorting the
    /// encoded value strings restores query order only because
    /// `ordered_index_fields` lists just the fields whose values all use the
    /// fixed-width, order-preserving integer encoding.
    ///
    /// A descending query reverses the value groups while keeping IDs
    /// ascending inside each group. Rows with a missing/null value sort first
    /// ascending and last descending.
    async fn ordered_index_ids(
        &self,
        ids: &HashSet<String>,
        field: &str,
        desc: bool,
    ) -> Result<Vec<String>> {
        let field_prefix = format!("{}{}{}{}", self.prefix, KEY_SEP, field, KEY_SEP);
        let options = ScanOptions::new(ScanOperation::Eq, field_prefix.clone(), false);
        let entries = self.kv.scan_prefix(&field_prefix, options).await?;

        // Group by encoded value: ordering the map restores value order, and
        // the per-group id sort restores the tie-break.
        let mut groups: BTreeMap<String, Vec<String>> = BTreeMap::new();
        let mut present = HashSet::new();
        for (key, _) in entries {
            let Some(rest) = key.strip_prefix(&field_prefix) else {
                continue;
            };
            // A value never contains `KEY_SEP` (the key encoding escapes it
            // away) while an id may, so the first separator ends the value.
            let Some(sep_pos) = rest.find(KEY_SEP) else {
                continue;
            };
            let value = &rest[..sep_pos];
            let id = &rest[sep_pos + KEY_SEP.len()..];
            // `cmp_order_docs` treats JSON null exactly like a missing value.
            if value == "null" || !ids.contains(id) {
                continue;
            }
            if !present.insert(id.to_string()) {
                continue;
            }
            groups
                .entry(value.to_string())
                .or_default()
                .push(id.to_string());
        }

        // Documents whose order value is null or missing carry no group.
        let mut missing: Vec<String> = ids.difference(&present).cloned().collect();
        missing.sort();

        let mut ordered_ids = Vec::with_capacity(ids.len());
        if desc {
            for group in groups.values_mut().rev() {
                group.sort();
                ordered_ids.append(group);
            }
            ordered_ids.extend(missing);
        } else {
            ordered_ids.append(&mut missing);
            for group in groups.values_mut() {
                group.sort();
                ordered_ids.append(group);
            }
        }
        Ok(ordered_ids)
    }

    /// Walk the filter tree and combine ID sets using AND/OR.
    async fn filter_ids(
        &self,
        filter: &Filter,
        indexed: &[&str],
        order_by: &[OrderBy],
    ) -> Result<HashSet<String>> {
        // Validate before short-circuiting so filter errors cannot depend on
        // the data or on branch order.
        Self::validate_filter(filter)?;

        if filter.r#type == FilterType::And {
            return self.and_filter_ids(filter, indexed, order_by).await;
        }

        // OR is order-insensitive semantically (it only affects scheduling of
        // work), so retain caller order there.
        let mut result: Option<HashSet<String>> = None;
        for cond in &filter.exprs {
            // boxed: `filter_expr_ids` recurses back here (mutual async recursion)
            let ids = Box::pin(self.filter_expr_ids(cond, indexed, order_by)).await?;
            result = Some(match result {
                None => ids,
                Some(existing) => existing.union(&ids).cloned().collect(),
            });
        }
        Ok(result.unwrap_or_default())
    }

    /// Evaluate AND branches by exact candidate cardinality and stop as soon
    /// as any branch is empty.
    ///
    /// The scan that produces a branch also gives its selectivity: indexed
    /// entries map one-to-one to candidate IDs, so the resulting set length is
    /// the cheap cardinality estimate. Intersecting smallest first minimizes
    /// the size of every later intermediate set.
    async fn and_filter_ids(
        &self,
        filter: &Filter,
        indexed: &[&str],
        order_by: &[OrderBy],
    ) -> Result<HashSet<String>> {
        let mut branches: Vec<HashSet<String>> = Vec::with_capacity(filter.exprs.len());
        for cond in &filter.exprs {
            // boxed: `filter_expr_ids` recurses back here (mutual async recursion)
            let ids = Box::pin(self.filter_expr_ids(cond, indexed, order_by)).await?;
            if ids.is_empty() {
                return Ok(ids);
            }
            branches.push(ids);
        }

        branches.sort_by_key(HashSet::len);
        if branches.is_empty() {
            return Ok(HashSet::new());
        }
        let mut result = branches.remove(0);
        for ids in branches {
            result.retain(|id| ids.contains(id));
            if result.is_empty() {
                break;
            }
        }
        Ok(result)
    }

    fn validate_filter(filter: &Filter) -> Result<()> {
        for cond in &filter.exprs {
            match cond {
                FilterExpr::Expr(expr) => Self::validate_expr(expr)?,
                FilterExpr::Filter(filter) => Self::validate_filter(filter)?,
            }
        }
        Ok(())
    }

    /// Validate array-shaped operators independently of the selected scan
    /// path, so short-circuiting cannot hide an invalid later branch.
    fn validate_expr(expr: &Expr) -> Result<()> {
        match &expr.op {
            ExprOp::Between => {
                let arr = expr.value.as_array().map(Vec::as_slice).unwrap_or(&[]);
                if arr.len() < 2 {
                    return Err(ActError::Store(
                        "Between operator requires an array of two values".to_string(),
                    ));
                }
            }
            ExprOp::In if !expr.value.as_array().is_some_and(|a| !a.is_empty()) => {
                return Err(ActError::Store(
                    "In operator requires a non-empty array".to_string(),
                ));
            }
            _ => {}
        }
        Ok(())
    }

    /// Rebuild every index entry of this collection from the stored data
    /// documents.
    ///
    /// Needed after a key-encoding change (e.g. `KEY_SEP` escaping rules):
    /// index keys written by older code encode values differently, so field
    /// scans silently miss or cross-match entries until the index region is
    /// recreated from the authoritative `{prefix}-id-` data region.
    pub async fn rebuild_index(&self) -> Result<usize>
    where
        T: DbCollectionIden,
    {
        // Drop the whole per-field index region, then recreate from data.
        for field in T::indexed_fields() {
            let field_prefix = format!("{}{}{}{}", self.prefix, KEY_SEP, field, KEY_SEP);
            let options = ScanOptions::new(ScanOperation::Eq, field_prefix.clone(), false);
            let stale = self.kv.scan_prefix(&field_prefix, options).await?;
            for (key, _) in stale {
                self.kv.delete(&key).await?;
            }
        }
        let data_prefix = format!("{}{}id{}", self.prefix, KEY_SEP, KEY_SEP);
        let options = ScanOptions::new(ScanOperation::Eq, data_prefix.clone(), false);
        let docs = self.kv.scan_prefix(&data_prefix, options).await?;
        for (_, bytes) in &docs {
            let json: JsonValue = serde_json::from_slice(bytes).map_err(map_db_err)?;
            let id = extract_id(&json)?;
            // Re-read the document under its mutation lock: the index region
            // was just wiped, so the keys must come from a version no
            // concurrent update can replace between this read and the puts.
            // A document deleted meanwhile is skipped instead of being
            // re-indexed.
            let _lock = lock_docs([self.data_key(&id)]).await;
            let Some(json) = self.read_json(&id).await? else {
                continue;
            };
            for idx_key in self.index_keys(&json, &id) {
                self.kv.put(&idx_key, vec![]).await?;
            }
        }
        Ok(docs.len())
    }
}

/// Convert a JSON value to a string suitable for use as an index-key segment.
///
/// Integers (i64, u64) are zero-padded to 20 digits so that lexicographic
/// ordering matches numeric ordering (otherwise "10" < "5").
/// Compare two JsonValues for ordering.
///
/// Numbers use the exact numeric order shared with `order_by`; everything
/// else is compared as a string.
fn cmp_json_val(a: &JsonValue, b: &JsonValue) -> Ordering {
    if let (JsonValue::Number(na), JsonValue::Number(nb)) = (a, b) {
        cmp_order_numbers(na, nb)
    } else {
        a.to_string().cmp(&b.to_string())
    }
}

/// Compare two documents by the query's `order_by` keys, in listed priority.
///
/// A field that is missing or JSON `null` counts as "no value": under `Asc`
/// it sorts before every value, so under `Desc` it lands last. Numbers
/// compare numerically (exact when both sides are integers); every other
/// pair compares by canonical JSON text, which keeps the order total and
/// deterministic. Rows that compare equal on every key keep their input
/// order (stable sort), which is id-ascending by construction in `query`.
fn cmp_order_docs(a: &JsonValue, b: &JsonValue, order_by: &[OrderBy]) -> Ordering {
    let mut ret = Ordering::Equal;
    for ob in order_by {
        let av = a.get(&ob.field).filter(|v| !v.is_null());
        let bv = b.get(&ob.field).filter(|v| !v.is_null());
        let mut cmp = match (av, bv) {
            (Some(av), Some(bv)) => cmp_order_values(av, bv),
            (Some(_), None) => Ordering::Greater,
            (None, Some(_)) => Ordering::Less,
            (None, None) => Ordering::Equal,
        };
        if ob.order == Sort::Desc {
            cmp = cmp.reverse();
        }
        ret = ret.then(cmp);
    }
    ret
}

/// Total order over two present JSON values for `order_by` comparison.
fn cmp_order_values(a: &JsonValue, b: &JsonValue) -> Ordering {
    match (a, b) {
        (JsonValue::Number(na), JsonValue::Number(nb)) => cmp_order_numbers(na, nb),
        _ => a.to_string().cmp(&b.to_string()),
    }
}

/// Exact numeric order for JSON numbers.
fn cmp_order_numbers(a: &serde_json::Number, b: &serde_json::Number) -> Ordering {
    if let (Some(x), Some(y)) = (a.as_i64(), b.as_i64()) {
        return x.cmp(&y);
    }
    if let (Some(x), Some(y)) = (a.as_u64(), b.as_u64()) {
        return x.cmp(&y);
    }
    // Mixed signedness has no shared integer view: the sign decides first,
    // then the magnitudes compare exactly as u64.
    if let (Some(x), Some(y)) = (a.as_i64(), b.as_u64()) {
        return if x < 0 {
            Ordering::Less
        } else {
            (x as u64).cmp(&y)
        };
    }
    if let (Some(x), Some(y)) = (a.as_u64(), b.as_i64()) {
        return if y < 0 {
            Ordering::Greater
        } else {
            x.cmp(&(y as u64))
        };
    }
    // At least one side is a float.
    let fa = a.as_f64().unwrap_or_default();
    let fb = b.as_f64().unwrap_or_default();
    fa.partial_cmp(&fb).unwrap_or(Ordering::Equal)
}

/// Characters in `[a-zA-Z0-9]` pass through unchanged. Every other character
/// — including `%`, `_`, `\`, `|`, `=`, `.` and `-` — is encoded as `=XX`
/// (2-digit uppercase hex for code points 0–255, 6-digit for code points
/// above 255). The `=` escape-prefix is itself valid in NATS KV keys (the
/// strictest backend), and the encoding is applied identically during key
/// creation and query scan-key construction so that lookups always match.
///
/// `-` MUST stay encoded: it is `KEY_SEP`, the delimiter between the field,
/// value and id segments of an index key. Keeping it out of the value charset
/// guarantees every value group is a contiguous, monotonically ordered key
/// range that closed range bounds can address exactly (see `KEY_SEP_SUCC`).
fn encode_key_str(s: &str) -> String {
    let mut result = String::with_capacity(s.len());
    for c in s.chars() {
        match c {
            'a'..='z' | 'A'..='Z' | '0'..='9' => result.push(c),
            other => {
                result.push('=');
                let code = other as u32;
                if code <= 0xFF {
                    result.push_str(&format!("{:02X}", code));
                } else {
                    result.push_str(&format!("{:06X}", code));
                }
            }
        }
    }
    result
}

fn json_value_to_key_str(v: &JsonValue) -> String {
    match v {
        JsonValue::String(s) => encode_key_str(s),
        JsonValue::Number(n) => {
            if let Some(i) = n.as_i64() {
                format!("{:020}", i)
            } else if let Some(u) = n.as_u64() {
                format!("{:020}", u)
            } else {
                n.to_string()
            }
        }
        other => other.to_string(),
    }
}

#[async_trait::async_trait]
impl<T> DbCollection for KvCollection<T>
where
    T: DbCollectionIden + Serialize + DeserializeOwned + Send + Sync + Clone + Debug + 'static,
{
    type Item = T;

    async fn exists(&self, id: &str) -> crate::Result<bool> {
        let key = self.data_key(id);
        self.kv.one(&key).await.map(|v| v.is_some())
    }

    async fn find(&self, id: &str) -> crate::Result<Self::Item> {
        self.find_opt(id)
            .await?
            .ok_or_else(|| ActError::Store(format!("cannot find {} by '{}'", self.prefix, id)))
    }

    async fn find_opt(&self, id: &str) -> crate::Result<Option<Self::Item>> {
        let key = self.data_key(id);
        let Some(data) = self.kv.one(&key).await? else {
            return Ok(None);
        };
        let json: JsonValue = serde_json::from_slice(&data).map_err(map_db_err)?;
        Ok(Some(T::upcast(json)?))
    }

    async fn query(&self, q: &Query) -> crate::Result<PageData<Self::Item>> {
        // Step 0: Validate query parameters
        if q.limit == 0 {
            return Err(ActError::Store(
                "query limit must be greater than 0".to_string(),
            ));
        }
        let indexed = T::indexed_fields();

        // Step 1 & 2: Compute matching ID set from filter and combine with AND/OR
        let id_set = self
            .filter_id_set(q.filter.as_ref(), q.get_order_by())
            .await?;

        let count = id_set.len();

        // Step 3: Paginate. Sorting happens BEFORE pagination when `order_by`
        // is set: every page must be the global top-N slice, not a re-sorted
        // batch of an arbitrary page. Without `order_by` only the page ids
        // are read from the store.
        let order_by = q.get_order_by();
        let index_ordered = match order_by.as_slice() {
            [only]
                if indexed.contains(&only.field.as_str())
                    && T::ordered_index_fields().contains(&only.field.as_str()) =>
            {
                Some(only.order == Sort::Desc)
            }
            _ => None,
        };

        let rows: Vec<T> = if let Some(desc) = index_ordered {
            // The order index itself supplies the global order, so only the
            // page's document bodies need to be materialized. `ids` remains
            // the authoritative count source (including rows missing/null in
            // the order field, which `ordered_index_ids` places correctly).
            let ordered_ids = self
                .ordered_index_ids(&id_set, &order_by[0].field, desc)
                .await?;
            let page_ids = ordered_ids
                .into_iter()
                .skip(q.offset)
                .take(q.limit)
                .collect::<Vec<_>>();
            self.read_json_many(&page_ids)
                .await?
                .into_iter()
                .map(|row| T::upcast(row))
                .collect::<Result<Vec<T>>>()?
        } else if order_by.is_empty() {
            // IDs are ascending: this is both the implicit page order and the
            // stable tie-break that makes offsets deterministic.
            let mut ids: Vec<String> = id_set.into_iter().collect();
            ids.sort();
            let page_ids = ids
                .into_iter()
                .skip(q.offset)
                .take(q.limit)
                .collect::<Vec<_>>();
            self.read_json_many(&page_ids)
                .await?
                .into_iter()
                .map(|row| T::upcast(row))
                .collect::<Result<Vec<T>>>()?
        } else {
            let mut ids: Vec<String> = id_set.into_iter().collect();
            ids.sort();
            let mut docs = self.read_json_many(&ids).await?;
            docs.sort_by(|a, b| cmp_order_docs(a, b, order_by));
            docs.into_iter()
                .skip(q.offset)
                .take(q.limit)
                .map(|row| T::upcast(row))
                .collect::<Result<Vec<T>>>()?
        };

        let page_count = count.div_ceil(q.limit);
        let page_num = q.offset.checked_div(q.limit).map_or(1, |n| n + 1);

        Ok(PageData {
            count,
            page_size: q.limit,
            page_num,
            page_count,
            rows,
        })
    }

    async fn create(&self, data: &Self::Item) -> crate::Result<bool> {
        // `create` and `update` are the same write. Both may land on an id
        // that already holds a document — the engine's find-then-create
        // upserts (`publish`, `upsert_proc`, the task/vars rows) and two
        // racing callers of either — and the invariant they maintain is one:
        // after the write, the index rows of this id are exactly the keys
        // derived from the stored document. A create that merely overwrote the
        // data row would leave the previous document's index rows behind,
        // answering queries for a value the row no longer holds.
        self.write_document(data).await
    }

    async fn update(&self, data: &Self::Item) -> crate::Result<bool> {
        self.write_document(data).await
    }

    async fn delete(&self, id: &str) -> crate::Result<bool> {
        // Index rows and the data row are removed as one atomic batch: the
        // index rows are the ones the stored document holds, read under the
        // document's lock (`lock_docs`).
        let _lock = lock_docs([self.data_key(id)]).await;
        let ops = self.delete_ops(id).await?;
        self.kv.batch(&ops).await?;
        Ok(true)
    }

    async fn query_all(&self, q: &Query) -> crate::Result<Vec<Self::Item>> {
        let ids = self.matching_ids(q.filter.as_ref()).await?;
        // `q.limit` sizes one document read (`mget`) batch here, never the
        // result: this is the exhaustive read.
        let mut docs = Vec::with_capacity(ids.len());
        for page in ids.chunks(q.limit.max(1)) {
            docs.extend(self.read_json_many(page).await?);
        }
        if !q.order_by.is_empty() {
            docs.sort_by(|a, b| cmp_order_docs(a, b, &q.order_by));
        }
        docs.into_iter().map(T::upcast).collect()
    }

    async fn matching_ids(&self, filter: Option<&Filter>) -> crate::Result<Vec<String>> {
        let mut ids: Vec<String> = self.filter_id_set(filter, &[]).await?.into_iter().collect();
        ids.sort();
        Ok(ids)
    }

    async fn find_matching(
        &self,
        q: &Query,
        pred: &(dyn for<'a> Fn(&'a Self::Item) -> bool + Sync),
    ) -> crate::Result<Option<Self::Item>> {
        let ids = self.matching_ids(q.filter.as_ref()).await?;
        for page in ids.chunks(q.limit.max(1)) {
            for json in self.read_json_many(page).await? {
                let row = T::upcast(json)?;
                if pred(&row) {
                    return Ok(Some(row));
                }
            }
        }
        Ok(None)
    }
}

impl<T> KvCollection<T>
where
    T: DbCollectionIden + Serialize,
{
    /// Write `data` as the document of its id: the stale index rows of the
    /// stored version, the data row and the new index rows commit as one
    /// atomic batch, computed while holding the document's lock, so no
    /// concurrent mutation of the same id can interleave its own index
    /// computation with this one.
    async fn write_document(&self, data: &T) -> crate::Result<bool> {
        let json = serde_json::to_value(data).map_err(map_db_err)?;
        let _lock = lock_docs([self.data_key(&extract_id(&json)?)]).await;
        let ops = self.update_ops_json(&json).await?;
        self.kv.batch(&ops).await?;
        Ok(true)
    }
}

fn extract_id(json: &JsonValue) -> crate::Result<String> {
    json.get("id")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string())
        .ok_or_else(|| ActError::Store("missing id field".to_string()))
}

impl Expr {
    pub fn op(&self, l: &serde_json::Value, r: &serde_json::Value) -> bool {
        match &self.op {
            ExprOp::EQ => l == r,
            ExprOp::NE => l != r,
            ExprOp::LT => {
                if let (serde_json::Value::Number(v1), serde_json::Value::Number(v2)) = (l, r) {
                    return cmp_order_numbers(v1, v2) == Ordering::Less;
                }
                false
            }
            ExprOp::LE => {
                if let (serde_json::Value::Number(v1), serde_json::Value::Number(v2)) = (l, r) {
                    return cmp_order_numbers(v1, v2) != Ordering::Greater;
                }
                false
            }
            ExprOp::GT => {
                if let (serde_json::Value::Number(v1), serde_json::Value::Number(v2)) = (l, r) {
                    return cmp_order_numbers(v1, v2) == Ordering::Greater;
                }
                false
            }
            ExprOp::GE => {
                if let (serde_json::Value::Number(v1), serde_json::Value::Number(v2)) = (l, r) {
                    return cmp_order_numbers(v1, v2) != Ordering::Less;
                }
                false
            }
            ExprOp::Match => {
                // Extract raw strings for comparison (not JSON-encoded to_string,
                // which would escape \ as \\ and cause false negatives)
                let l_str: String = match l {
                    JsonValue::String(v) => v.clone(),
                    other => other.to_string(),
                };
                let r_str: String = match r {
                    JsonValue::String(v) => v.clone(),
                    other => other.to_string(),
                };
                l_str.contains(&r_str)
            }
            ExprOp::Between => {
                let arr = match r.as_array() {
                    Some(a) if a.len() >= 2 => a,
                    _ => return false,
                };
                cmp_json_val(l, &arr[0]) != Ordering::Less
                    && cmp_json_val(l, &arr[1]) != Ordering::Greater
            }
            ExprOp::In => {
                let arr = match r.as_array() {
                    Some(a) => a,
                    None => return false,
                };
                arr.iter().any(|v| cmp_json_val(l, v) == Ordering::Equal)
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{JsonValue, encode_key_str, json_value_to_key_str};
    use crate::store::Expr;
    use serde_json::json;

    #[test]
    fn encode_key_str_passthrough() {
        // Characters in the safe set pass through unchanged
        assert_eq!(encode_key_str("hello"), "hello");
        assert_eq!(encode_key_str("abcABC123"), "abcABC123");
        // `-` is KEY_SEP and must not appear inside a value segment
        assert_eq!(encode_key_str("hello-world"), "hello=2Dworld");
        assert_eq!(encode_key_str("with_underscore"), "with=5Funderscore");
        assert_eq!(encode_key_str(""), "");
    }

    #[test]
    fn encode_key_str_hyphen_is_escaped() {
        // '-' must never collide with the KEY_SEP delimiter
        assert_eq!(encode_key_str("a-b-c"), "a=2Db=2Dc");
        assert_eq!(encode_key_str("-"), "=2D");
        assert_eq!(encode_key_str("my-workflow-v2"), "my=2Dworkflow=2Dv2");
    }

    #[test]
    fn encode_key_str_percent() {
        assert_eq!(encode_key_str("50%off"), "50=25off");
    }

    #[test]
    fn encode_key_str_pipe() {
        assert_eq!(encode_key_str("a|b|c"), "a=7Cb=7Cc");
    }

    #[test]
    fn encode_key_str_backslash() {
        assert_eq!(encode_key_str(r"a\b"), "a=5Cb");
    }

    #[test]
    fn encode_key_str_equals() {
        // = itself is encoded so it never appears un-escaped in the output
        assert_eq!(encode_key_str("a=b"), "a=3Db");
    }

    #[test]
    fn encode_key_str_dot() {
        assert_eq!(encode_key_str("file.txt"), "file=2Etxt");
    }

    #[test]
    fn encode_key_str_mixed_special() {
        assert_eq!(encode_key_str("a%b|c\\d"), "a=25b=7Cc=5Cd");
    }

    #[test]
    fn encode_key_str_emoji() {
        // Non-BMP character encoded with 6-digit hex
        let s = encode_key_str("hi😀");
        assert!(s.starts_with("hi="));
        assert!(s.len() > 4);
    }

    #[test]
    fn json_value_to_key_str_string() {
        assert_eq!(json_value_to_key_str(&json!("hello")), "hello");
    }

    #[test]
    fn json_value_to_key_str_string_with_special() {
        // Special characters are encoded via encode_key_str
        assert_eq!(json_value_to_key_str(&json!("a%b")), "a=25b");
    }

    #[test]
    fn json_value_to_key_str_i64_zero_pads() {
        assert_eq!(json_value_to_key_str(&json!(5)), "00000000000000000005");
        assert_eq!(json_value_to_key_str(&json!(10)), "00000000000000000010");
        assert_eq!(json_value_to_key_str(&json!(100)), "00000000000000000100");
    }

    #[test]
    fn json_value_to_key_str_i64_negative() {
        assert_eq!(json_value_to_key_str(&json!(-5)), "-0000000000000000005");
    }

    #[test]
    fn json_value_to_key_str_u64_zero_pads() {
        let big: u64 = u64::MAX;
        assert_eq!(json_value_to_key_str(&json!(big)), "18446744073709551615");
    }

    #[test]
    fn json_value_to_key_str_lexicographic_order() {
        // Verify that zero-padded integers sort correctly lexicographically:
        // after zero-padding, "000000000...5" < "000000000...10"
        let key1 = json_value_to_key_str(&json!(1));
        let key2 = json_value_to_key_str(&json!(2));
        let key5 = json_value_to_key_str(&json!(5));
        let key10 = json_value_to_key_str(&json!(10));
        let key100 = json_value_to_key_str(&json!(100));

        let mut sorted = vec![&key10, &key100, &key1, &key5, &key2];
        sorted.sort();
        assert_eq!(sorted, vec![&key1, &key2, &key5, &key10, &key100]);
    }

    #[test]
    fn json_value_to_key_str_float_no_padding() {
        // Floats are not padded — they can't be ordered lexicographically anyway
        let v = json!(2.71);
        let s = json_value_to_key_str(&v);
        assert!(s.contains("2.71"));
    }

    #[test]
    fn json_value_to_key_str_bool() {
        assert_eq!(json_value_to_key_str(&json!(true)), "true");
        assert_eq!(json_value_to_key_str(&json!(false)), "false");
    }

    #[test]
    fn json_value_to_key_str_null() {
        assert_eq!(json_value_to_key_str(&json!(null)), "null");
    }

    // ========== Expr::op() Between / In / cmp_json_val tests ==========

    #[test]
    fn store_expr_op_between_numbers_inside() {
        let expr = Expr::between("field", 10, 20);
        assert!(expr.op(&json!(10), &json!([10, 20])));
        assert!(expr.op(&json!(15), &json!([10, 20])));
        assert!(expr.op(&json!(20), &json!([10, 20])));
    }

    #[test]
    fn store_expr_op_between_numbers_outside() {
        let expr = Expr::between("field", 10, 20);
        assert!(!expr.op(&json!(9), &json!([10, 20])));
        assert!(!expr.op(&json!(21), &json!([10, 20])));
        assert!(!expr.op(&json!(100), &json!([10, 20])));
    }

    #[test]
    fn store_expr_op_between_strings() {
        let expr = Expr::between("field", "b", "d");
        assert!(!expr.op(&json!("a"), &json!(["b", "d"])));
        assert!(expr.op(&json!("b"), &json!(["b", "d"])));
        assert!(expr.op(&json!("c"), &json!(["b", "d"])));
        assert!(expr.op(&json!("d"), &json!(["b", "d"])));
        assert!(!expr.op(&json!("e"), &json!(["b", "d"])));
    }

    #[test]
    fn store_expr_op_between_invalid_array() {
        let expr = Expr::between("field", 1, 9);
        // Not an array — returns false
        assert!(!expr.op(&json!(5), &json!("not_array")));
        // Array with single element — returns false
        assert!(!expr.op(&json!(5), &json!([1])));
        // Empty array — returns false
        assert!(!expr.op(&json!(5), &json!([])));
    }

    #[test]
    fn store_expr_op_between_float() {
        let expr = Expr::between("field", 1.5, 3.5);
        assert!(!expr.op(&json!(1.0), &json!([1.5, 3.5])));
        assert!(expr.op(&json!(1.5), &json!([1.5, 3.5])));
        assert!(expr.op(&json!(2.0), &json!([1.5, 3.5])));
        assert!(expr.op(&json!(3.5), &json!([1.5, 3.5])));
        assert!(!expr.op(&json!(4.0), &json!([1.5, 3.5])));
    }

    #[test]
    fn store_expr_op_in_numbers() {
        let expr = Expr::r#in("field", vec![1, 3, 5]);
        assert!(expr.op(&json!(1), &json!([1, 3, 5])));
        assert!(expr.op(&json!(3), &json!([1, 3, 5])));
        assert!(expr.op(&json!(5), &json!([1, 3, 5])));
        assert!(!expr.op(&json!(0), &json!([1, 3, 5])));
        assert!(!expr.op(&json!(2), &json!([1, 3, 5])));
        assert!(!expr.op(&json!(6), &json!([1, 3, 5])));
    }

    #[test]
    fn store_expr_op_in_strings() {
        let expr = Expr::r#in("field", vec!["running", "completed"]);
        assert!(expr.op(&json!("running"), &json!(["running", "completed"])));
        assert!(expr.op(&json!("completed"), &json!(["running", "completed"])));
        assert!(!expr.op(&json!("pending"), &json!(["running", "completed"])));
        assert!(!expr.op(&json!("none"), &json!(["running", "completed"])));
    }

    #[test]
    fn store_expr_op_in_invalid() {
        let expr = Expr::r#in("field", vec![1, 2]);
        // Not an array — returns false
        assert!(!expr.op(&json!(1), &json!("not_array")));
        // Null — returns false
        assert!(!expr.op(&json!(1), &json!(null)));
    }

    #[test]
    fn store_expr_op_in_empty() {
        let expr = Expr::r#in("field", Vec::<i32>::new());
        // No values to match, always false
        assert!(!expr.op(&json!(1), &json!([])));
        assert!(!expr.op(&json!("a"), &json!([])));
    }

    // ========== cmp_json_val comparison tests ==========

    #[test]
    fn store_cmp_json_val_numbers() {
        use super::cmp_json_val;
        use std::cmp::Ordering;
        assert_eq!(cmp_json_val(&json!(10), &json!(5)), Ordering::Greater);
        assert_eq!(cmp_json_val(&json!(5), &json!(10)), Ordering::Less);
        assert_eq!(cmp_json_val(&json!(5), &json!(5)), Ordering::Equal);
    }

    #[test]
    fn store_cmp_json_val_strings() {
        use super::cmp_json_val;
        use std::cmp::Ordering;
        assert_eq!(cmp_json_val(&json!("abc"), &json!("abc")), Ordering::Equal);
        assert_eq!(cmp_json_val(&json!("abc"), &json!("def")), Ordering::Less);
        assert_eq!(
            cmp_json_val(&json!("def"), &json!("abc")),
            Ordering::Greater
        );
    }

    #[test]
    fn store_cmp_json_val_mixed_types() {
        use super::cmp_json_val;
        use std::cmp::Ordering;
        // number vs string — compared via to_string()
        // json!(10).to_string() = "10", json!("5").to_string() = "\"5\""
        // "10" > "\"5\"" because '1' (49) > '"' (34)
        let result = cmp_json_val(&json!(10), &json!("5"));
        assert_eq!(result, Ordering::Greater); // "10" > "\"5\"" lexicographically
    }

    #[test]
    fn store_cmp_json_val_floats() {
        use super::cmp_json_val;
        use std::cmp::Ordering;
        assert_eq!(cmp_json_val(&json!(1.5), &json!(1.5)), Ordering::Equal);
        assert_eq!(cmp_json_val(&json!(1.5), &json!(2.0)), Ordering::Less);
        assert_eq!(cmp_json_val(&json!(3.0), &json!(2.5)), Ordering::Greater);
    }

    #[test]
    fn store_cmp_json_val_exact_integers_beyond_f64_precision() {
        use super::cmp_json_val;
        use std::cmp::Ordering;

        assert_eq!(
            cmp_json_val(
                &json!(9_007_199_254_740_992_u64),
                &json!(9_007_199_254_740_993_u64)
            ),
            Ordering::Less
        );
        assert_eq!(
            cmp_json_val(&json!(i64::MAX), &json!(u64::MAX)),
            Ordering::Less
        );
    }

    #[test]
    fn store_expr_op_between_and_in_exact_large_integers() {
        let between = Expr::between(
            "field",
            9_007_199_254_740_992_u64,
            9_007_199_254_740_992_u64,
        );
        assert!(!between.op(
            &json!(9_007_199_254_740_993_u64),
            &json!([9_007_199_254_740_992_u64, 9_007_199_254_740_992_u64])
        ));

        let r#in = Expr::r#in("field", vec![u64::MAX - 1]);
        assert!(r#in.op(&json!(u64::MAX - 1), &json!([u64::MAX - 1])));
        assert!(!r#in.op(&json!(u64::MAX), &json!([u64::MAX - 1])));
    }

    // ========== Expr::op() NE / LT / LE / GT / GE / Match tests ==========

    #[test]
    fn store_expr_op_ne_numbers() {
        let expr = Expr::ne("field", 10);
        assert!(!expr.op(&json!(10), &json!(10)));
        assert!(expr.op(&json!(5), &json!(10)));
        assert!(expr.op(&json!(20), &json!(10)));
    }

    #[test]
    fn store_expr_op_ne_strings() {
        let expr = Expr::ne("field", "hello");
        assert!(!expr.op(&json!("hello"), &json!("hello")));
        assert!(expr.op(&json!("world"), &json!("hello")));
        assert!(expr.op(&json!(""), &json!("hello")));
    }

    #[test]
    fn store_expr_op_ne_mixed_types() {
        let expr = Expr::ne("field", 10);
        // NE returns true when types differ (l != r)
        assert!(expr.op(&json!("10"), &json!(10)));
    }

    #[test]
    fn store_expr_op_lt_numbers() {
        let expr = Expr::lt("field", 10);
        assert!(expr.op(&json!(5), &json!(10)));
        assert!(!expr.op(&json!(10), &json!(10)));
        assert!(!expr.op(&json!(15), &json!(10)));
    }

    #[test]
    fn store_expr_op_lt_non_number_returns_false() {
        let expr = Expr::lt("field", 10);
        // LT only works for numbers; non-numbers always return false
        assert!(!expr.op(&json!("5"), &json!(10)));
        assert!(!expr.op(&json!(null), &json!(10)));
    }

    #[test]
    fn store_expr_op_le_numbers() {
        let expr = Expr::le("field", 10);
        assert!(expr.op(&json!(5), &json!(10)));
        assert!(expr.op(&json!(10), &json!(10)));
        assert!(!expr.op(&json!(15), &json!(10)));
    }

    #[test]
    fn store_expr_op_le_non_number_returns_false() {
        let expr = Expr::le("field", 10);
        assert!(!expr.op(&json!("5"), &json!(10)));
    }

    #[test]
    fn store_expr_op_range_numbers_exact_mixed_numeric_types() {
        assert!(Expr::lt("field", 3.5).op(&json!(3), &json!(3.5)));
        assert!(!Expr::ge("field", 3.5).op(&json!(3), &json!(3.5)));

        assert!(Expr::lt("field", u64::MAX).op(&json!(5), &json!(u64::MAX)));
        assert!(!Expr::gt("field", u64::MAX).op(&json!(5), &json!(u64::MAX)));

        assert!(Expr::le("field", u64::MAX).op(&json!(i64::MAX), &json!(u64::MAX)));
        assert!(!Expr::ge("field", u64::MAX).op(&json!(i64::MAX), &json!(u64::MAX)));

        assert!(Expr::gt("field", -1).op(&json!(u64::MAX), &json!(-1)));
        assert!(!Expr::lt("field", -1).op(&json!(u64::MAX), &json!(-1)));
    }

    #[test]
    fn store_expr_op_gt_numbers() {
        let expr = Expr::gt("field", 10);
        assert!(!expr.op(&json!(5), &json!(10)));
        assert!(!expr.op(&json!(10), &json!(10)));
        assert!(expr.op(&json!(15), &json!(10)));
    }

    #[test]
    fn store_expr_op_gt_non_number_returns_false() {
        let expr = Expr::gt("field", 10);
        assert!(!expr.op(&json!("15"), &json!(10)));
    }

    #[test]
    fn store_expr_op_ge_numbers() {
        let expr = Expr::ge("field", 10);
        assert!(!expr.op(&json!(5), &json!(10)));
        assert!(expr.op(&json!(10), &json!(10)));
        assert!(expr.op(&json!(15), &json!(10)));
    }

    #[test]
    fn store_expr_op_ge_non_number_returns_false() {
        let expr = Expr::ge("field", 10);
        assert!(!expr.op(&json!("15"), &json!(10)));
    }

    #[test]
    fn store_expr_op_match_contains() {
        let expr = Expr::matches("field", "ello");
        assert!(expr.op(&json!("hello"), &json!("ello")));
        assert!(!expr.op(&json!("hello"), &json!("xyz")));
    }

    // ========== index-key range semantics regression tests ==========

    use super::KvCollection;
    use crate::store::{DbCollection, Filter, KvStore, Query, Sort};
    use serde::{Deserialize, Serialize};
    use std::sync::Arc;

    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct Doc {
        id: String,
        state: String,
        timestamp: i64,
    }

    impl crate::store::DbCollectionIden for Doc {
        fn iden() -> crate::store::StoreIden {
            crate::store::StoreIden::Ops
        }
        fn indexed_fields() -> &'static [&'static str] {
            &["state", "timestamp"]
        }
    }

    fn ids(page: &crate::store::PageData<Doc>) -> Vec<String> {
        page.rows.iter().map(|d| d.id.clone()).collect()
    }

    async fn query(col: &KvCollection<Doc>, filter: Filter) -> crate::store::PageData<Doc> {
        col.query(&Query::new().filter(filter)).await.unwrap()
    }

    #[tokio::test]
    async fn index_range_closed_boundaries_exact() {
        let kv: Arc<crate::store::MemoryStore> = Arc::new(crate::store::MemoryStore::new());
        let col = KvCollection::new("docs", kv.clone());
        for ts in [100i64, 200, 300] {
            col.create(&Doc {
                id: format!("d{ts}"),
                state: "idle".to_string(),
                timestamp: ts,
            })
            .await
            .unwrap();
        }
        // Inclusive Between keeps both exact boundaries (was: value == to dropped)
        let page = query(
            &col,
            Filter::and().expr(Expr::between("timestamp", 100, 200)),
        )
        .await;
        assert_eq!(ids(&page), vec!["d100", "d200"]);
        // Degenerate inclusive range returns the exact single value
        let page = query(
            &col,
            Filter::and().expr(Expr::between("timestamp", 100, 100)),
        )
        .await;
        assert_eq!(ids(&page), vec!["d100"]);
        // Single-sided comparisons are exact at the equality boundary
        let page = query(&col, Filter::and().expr(Expr::gt("timestamp", 100))).await;
        assert_eq!(ids(&page), vec!["d200", "d300"]);
        let page = query(&col, Filter::and().expr(Expr::ge("timestamp", 200))).await;
        assert_eq!(ids(&page), vec!["d200", "d300"]);
        let page = query(&col, Filter::and().expr(Expr::lt("timestamp", 200))).await;
        assert_eq!(ids(&page), vec!["d100"]);
        let page = query(&col, Filter::and().expr(Expr::le("timestamp", 200))).await;
        assert_eq!(ids(&page), vec!["d100", "d200"]);
        // Rows are sorted by id, so ids() must be sorted before comparing
    }

    #[tokio::test]
    async fn index_gate_falls_back_for_negative_bounds() {
        // Negative bounds are not indexable (zero-padding reverses their
        // order), so range/inequality scans must fall back to the full-data
        // scan. Stored negative timestamps make any index-path mistake visible.
        let kv: Arc<crate::store::MemoryStore> = Arc::new(crate::store::MemoryStore::new());
        let col = KvCollection::new("docs", kv.clone());
        for (i, ts) in [-200i64, -100, 100, 200].into_iter().enumerate() {
            col.create(&Doc {
                id: format!("d{i}"),
                state: "idle".to_string(),
                timestamp: ts,
            })
            .await
            .unwrap();
        }
        fn ts(page: &crate::store::PageData<Doc>) -> Vec<i64> {
            page.rows.iter().map(|d| d.timestamp).collect()
        }
        let page = query(
            &col,
            Filter::and().expr(Expr::between("timestamp", -150, 150)),
        )
        .await;
        assert_eq!(ts(&page), vec![-100, 100]);
        let page = query(&col, Filter::and().expr(Expr::ge("timestamp", -1))).await;
        assert_eq!(ts(&page), vec![100, 200]);
        let page = query(
            &col,
            Filter::and().expr(Expr::between("timestamp", -250, -50)),
        )
        .await;
        assert_eq!(ts(&page), vec![-200, -100]);
    }

    #[tokio::test]
    async fn index_eq_isolates_hyphenated_values() {
        let kv: Arc<crate::store::MemoryStore> = Arc::new(crate::store::MemoryStore::new());
        let col = KvCollection::new("docs", kv.clone());
        for (id, state) in [("a", "w9"), ("b", "w9-foo"), ("c", "other")] {
            col.create(&Doc {
                id: id.to_string(),
                state: state.to_string(),
                timestamp: 0,
            })
            .await
            .unwrap();
        }
        // '-' is escaped in the value segment: Eq on "w9" must not reach "w9-foo"
        let page = query(&col, Filter::and().expr(Expr::eq("state", "w9"))).await;
        assert_eq!(ids(&page), vec!["a"]);
        let page = query(&col, Filter::and().expr(Expr::eq("state", "w9-foo"))).await;
        assert_eq!(ids(&page), vec!["b"]);
        let page = query(&col, Filter::and().expr(Expr::ne("state", "w9"))).await;
        assert_eq!(ids(&page), vec!["b", "c"]);
    }

    #[tokio::test]
    async fn rebuild_index_repairs_stale_or_legacy_keys() {
        let kv: Arc<crate::store::MemoryStore> = Arc::new(crate::store::MemoryStore::new());
        let col = KvCollection::new("docs", kv.clone());
        for (id, state) in [("a", "w9"), ("b", "w9-foo")] {
            col.create(&Doc {
                id: id.to_string(),
                state: state.to_string(),
                timestamp: 0,
            })
            .await
            .unwrap();
        }
        // Simulate a pre-fix persisted index: value "-" not escaped, id suffixed
        let legacy = format!("docs-state-{}-{}", "w9-foo", "b");
        kv.put(&legacy, vec![]).await.unwrap();
        // Legacy key is a prefix-extension of the "w9" value group -> pollutes
        // Eq with a phantom id ("foo-b"), inflating count while the row is lost
        let page = query(&col, Filter::and().expr(Expr::eq("state", "w9"))).await;
        assert_eq!(page.count, 2, "legacy phantom id inflates count");
        assert_eq!(ids(&page), vec!["a"], "phantom row cannot be fetched");
        // Drop the fresh index key of doc b, then rebuild restores exactness
        let fresh = format!("docs-state-{}-{}", encode_key_str("w9-foo"), "b");
        kv.delete(&fresh).await.unwrap();
        let page = query(&col, Filter::and().expr(Expr::eq("state", "w9-foo"))).await;
        assert_eq!(page.count, 0, "fresh key deleted, doc b unreachable");
        assert!(col.rebuild_index().await.unwrap() >= 2);
        assert!(
            kv.one(&legacy).await.unwrap().is_none(),
            "legacy key removed"
        );
        assert!(
            kv.one(&fresh).await.unwrap().is_some(),
            "fresh key restored"
        );
        let page = query(&col, Filter::and().expr(Expr::eq("state", "w9"))).await;
        assert_eq!(ids(&page), vec!["a"], "no pollution after rebuild");
        let page = query(&col, Filter::and().expr(Expr::eq("state", "w9-foo"))).await;
        assert_eq!(ids(&page), vec!["b"]);
    }

    // ========== order_by semantics tests ==========

    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct SortDoc {
        id: String,
        // Optional keys exercise missing/null sort keys while staying
        // upcastable: `None` is serialized away, i.e. the key is absent.
        #[serde(skip_serializing_if = "Option::is_none")]
        group: Option<i64>,
        #[serde(skip_serializing_if = "Option::is_none")]
        ord: Option<i64>,
    }

    impl crate::store::DbCollectionIden for SortDoc {
        fn iden() -> crate::store::StoreIden {
            crate::store::StoreIden::Ops
        }
        fn indexed_fields() -> &'static [&'static str] {
            &[]
        }
    }

    fn sort_col() -> (Arc<crate::store::MemoryStore>, KvCollection<SortDoc>) {
        let kv: Arc<crate::store::MemoryStore> = Arc::new(crate::store::MemoryStore::new());
        let col = KvCollection::new("sortdocs", kv.clone());
        (kv, col)
    }

    fn mk_doc(id: &str, group: Option<i64>, ord: Option<i64>) -> SortDoc {
        SortDoc {
            id: id.to_string(),
            group,
            ord,
        }
    }

    async fn sort_query_ids(col: &KvCollection<SortDoc>, q: &Query) -> Vec<String> {
        col.query(q)
            .await
            .unwrap()
            .rows
            .iter()
            .map(|d| d.id.clone())
            .collect()
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct OrderedDoc {
        id: String,
        state: String,
        ord: JsonValue,
    }

    impl crate::store::DbCollectionIden for OrderedDoc {
        fn iden() -> crate::store::StoreIden {
            crate::store::StoreIden::Ops
        }
        fn indexed_fields() -> &'static [&'static str] {
            &["state", "ord"]
        }
        fn ordered_index_fields() -> &'static [&'static str] {
            &["ord"]
        }
    }

    #[tokio::test]
    async fn indexed_order_pages_without_inverting_ties() {
        let kv: Arc<crate::store::MemoryStore> = Arc::new(crate::store::MemoryStore::new());
        let col = KvCollection::<OrderedDoc>::new("ordered", kv.clone());
        for (id, ord) in [
            ("a", json!(2)),
            ("b", json!(2)),
            ("c", json!(1)),
            ("d", json!(1)),
            ("e", json!(null)),
            ("f", json!(5)),
        ] {
            col.create(&OrderedDoc {
                id: id.to_string(),
                state: "idle".to_string(),
                ord,
            })
            .await
            .unwrap();
        }

        let asc = col
            .query(&Query::new().order("ord", Sort::Asc).limit(100))
            .await
            .unwrap();
        assert_eq!(
            asc.rows.iter().map(|d| d.id.clone()).collect::<Vec<_>>(),
            vec!["e", "c", "d", "a", "b", "f"]
        );
        // Desc reverses value groups, but equal values retain ascending IDs.
        let desc_page = col
            .query(&Query::new().order("ord", Sort::Desc).limit(3).offset(2))
            .await
            .unwrap();
        assert_eq!(
            desc_page
                .rows
                .iter()
                .map(|d| d.id.clone())
                .collect::<Vec<_>>(),
            vec!["b", "c", "d"]
        );
        assert_eq!(desc_page.count, 6);
    }

    /// Kv wrapper whose `scan_prefix` returns entries in reverse key order,
    /// standing in for a backend that makes no ordering promise (Redis
    /// `SCAN`). The ordered-index pagination must not depend on it.
    struct UnorderedScanKv {
        inner: MemoryStore,
    }

    #[async_trait::async_trait]
    impl KvStore for UnorderedScanKv {
        async fn one(&self, key: &str) -> crate::Result<Option<Vec<u8>>> {
            self.inner.one(key).await
        }

        async fn put(&self, key: &str, value: Vec<u8>) -> crate::Result<()> {
            self.inner.put(key, value).await
        }

        async fn delete(&self, key: &str) -> crate::Result<()> {
            self.inner.delete(key).await
        }

        async fn scan_prefix(
            &self,
            key: &str,
            options: ScanOptions,
        ) -> crate::Result<Vec<(String, Vec<u8>)>> {
            let mut entries = self.inner.scan_prefix(key, options).await?;
            entries.reverse();
            Ok(entries)
        }
    }

    #[tokio::test]
    async fn indexed_order_does_not_trust_scan_order() {
        let kv: Arc<dyn KvStore> = Arc::new(UnorderedScanKv {
            inner: MemoryStore::new(),
        });
        let col = KvCollection::<OrderedDoc>::new("unordered", kv);
        for (id, ord) in [
            ("a", json!(2)),
            ("b", json!(2)),
            ("c", json!(1)),
            ("d", json!(1)),
            ("e", json!(null)),
            ("f", json!(5)),
        ] {
            col.create(&OrderedDoc {
                id: id.to_string(),
                state: "idle".to_string(),
                ord,
            })
            .await
            .unwrap();
        }

        let asc = col
            .query(&Query::new().order("ord", Sort::Asc).limit(100))
            .await
            .unwrap();
        assert_eq!(
            asc.rows.iter().map(|d| d.id.clone()).collect::<Vec<_>>(),
            vec!["e", "c", "d", "a", "b", "f"]
        );

        let desc = col
            .query(&Query::new().order("ord", Sort::Desc).limit(100))
            .await
            .unwrap();
        assert_eq!(
            desc.rows.iter().map(|d| d.id.clone()).collect::<Vec<_>>(),
            vec!["f", "a", "b", "c", "d", "e"]
        );
    }

    #[tokio::test]
    async fn order_by_sorts_before_pagination() {
        let (_, col) = sort_col();
        for (id, group, ord) in [
            ("a", Some(1), Some(3)),
            ("b", Some(1), Some(2)),
            ("c", Some(2), Some(1)),
            ("d", Some(1), Some(1)),
            ("e", Some(2), Some(5)),
            ("f", None, None),       // no value on every key: first under Asc
            ("g", Some(2), Some(5)), // ties with "e" -> id-ascending break
        ] {
            col.create(&mk_doc(id, group, ord)).await.unwrap();
        }
        // group asc, ord desc: f | a b d | e g c
        let order = Query::new()
            .order("group", Sort::Asc)
            .order("ord", Sort::Desc);
        let full = sort_query_ids(&col, &order.clone().limit(100)).await;
        assert_eq!(full, vec!["f", "a", "b", "d", "e", "g", "c"]);

        // Each page must be the corresponding global slice, not a re-sorted
        // arbitrary batch: concatenated pages equal the full sorted order.
        let page1 = sort_query_ids(&col, &order.clone().limit(2).offset(0)).await;
        let page2 = sort_query_ids(&col, &order.clone().limit(2).offset(2)).await;
        let page3 = sort_query_ids(&col, &order.clone().limit(2).offset(4)).await;
        let page4 = sort_query_ids(&col, &order.clone().limit(2).offset(6)).await;
        assert_eq!(page1, vec!["f", "a"]);
        assert_eq!(page2, vec!["b", "d"]);
        assert_eq!(page3, vec!["e", "g"]);
        assert_eq!(page4, vec!["c"]);
        let page = col.query(&order.clone().limit(2).offset(4)).await.unwrap();
        assert_eq!((page.count, page.page_num, page.page_count), (7, 3, 4));
    }

    #[tokio::test]
    async fn order_by_numeric_not_lexicographic() {
        let (_, col) = sort_col();
        for (id, ord) in [("ten", Some(10)), ("nine", Some(9)), ("one", Some(1))] {
            col.create(&mk_doc(id, Some(1), ord)).await.unwrap();
        }
        // "10" < "9" lexicographically; numeric order must give 1, 9, 10.
        let q = Query::new().order("ord", Sort::Asc);
        assert_eq!(sort_query_ids(&col, &q).await, vec!["one", "nine", "ten"]);
        let q = Query::new().order("ord", Sort::Desc);
        assert_eq!(sort_query_ids(&col, &q).await, vec!["ten", "nine", "one"]);
    }

    #[tokio::test]
    async fn order_by_no_value_first_asc_last_desc() {
        let (kv, col) = sort_col();
        for (id, ord) in [
            ("low", Some(1)),
            ("nil", None), // serialized as null
            ("high", Some(5)),
            ("mid", Some(3)),
        ] {
            col.create(&mk_doc(id, None, ord)).await.unwrap();
        }
        // Raw doc whose `ord` key is missing entirely must sort like null.
        let raw = serde_json::json!({"id": "absent", "group": null});
        kv.put(&col.data_key("absent"), serde_json::to_vec(&raw).unwrap())
            .await
            .unwrap();
        let q = Query::new().order("ord", Sort::Asc);
        assert_eq!(
            sort_query_ids(&col, &q).await,
            vec!["absent", "nil", "low", "mid", "high"]
        );
        let q = Query::new().order("ord", Sort::Desc);
        // no-value rows land last; among them the id-ascending tie-break wins
        assert_eq!(
            sort_query_ids(&col, &q).await,
            vec!["high", "mid", "low", "absent", "nil"]
        );
    }

    // ========== atomic batch write tests ==========

    use crate::store::{MemoryStore, ScanOptions, StoreBatchOp};
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// Kv wrapper that counts how the collection writes: a document write
    /// (data row + index rows) must go through exactly one `batch` call and
    /// never through raw `put`/`delete`, or a mid-write failure could tear
    /// the document from its indexes.
    #[derive(Default)]
    struct CountingKv {
        inner: MemoryStore,
        batches: AtomicUsize,
        mgets: AtomicUsize,
        puts: AtomicUsize,
        deletes: AtomicUsize,
        scans: AtomicUsize,
    }

    #[async_trait::async_trait]
    impl KvStore for CountingKv {
        async fn one(&self, key: &str) -> crate::Result<Option<Vec<u8>>> {
            self.inner.one(key).await
        }

        async fn put(&self, key: &str, value: Vec<u8>) -> crate::Result<()> {
            self.puts.fetch_add(1, Ordering::SeqCst);
            self.inner.put(key, value).await
        }

        async fn delete(&self, key: &str) -> crate::Result<()> {
            self.deletes.fetch_add(1, Ordering::SeqCst);
            self.inner.delete(key).await
        }

        async fn batch(&self, ops: &[StoreBatchOp]) -> crate::Result<()> {
            self.batches.fetch_add(1, Ordering::SeqCst);
            self.inner.batch(ops).await
        }

        async fn many(&self, keys: &[String]) -> crate::Result<Vec<Option<Vec<u8>>>> {
            self.mgets.fetch_add(1, Ordering::SeqCst);
            self.inner.many(keys).await
        }

        async fn scan_prefix(
            &self,
            key: &str,
            options: ScanOptions,
        ) -> crate::Result<Vec<(String, Vec<u8>)>> {
            self.scans.fetch_add(1, Ordering::SeqCst);
            self.inner.scan_prefix(key, options).await
        }
    }

    fn counting_col() -> (Arc<CountingKv>, KvCollection<Doc>) {
        let kv = Arc::new(CountingKv::default());
        let col = KvCollection::new("docs", kv.clone());
        (kv, col)
    }

    fn doc(id: &str, state: &str, timestamp: i64) -> Doc {
        Doc {
            id: id.to_string(),
            state: state.to_string(),
            timestamp,
        }
    }

    #[tokio::test]
    async fn create_commits_data_and_indexes_in_one_batch() {
        let (kv, col) = counting_col();
        col.create(&doc("d1", "idle", 5)).await.unwrap();

        assert_eq!(
            kv.batches.load(Ordering::SeqCst),
            1,
            "create must be a single atomic batch"
        );
        assert_eq!(
            (
                kv.puts.load(Ordering::SeqCst),
                kv.deletes.load(Ordering::SeqCst)
            ),
            (0, 0),
            "create must not fall back to raw per-key writes"
        );
        // the data row exists and the index answers queries
        assert_eq!(col.find("d1").await.unwrap().timestamp, 5);
        let page = query(&col, Filter::and().expr(Expr::eq("state", "idle"))).await;
        assert_eq!(ids(&page), vec!["d1"]);
    }

    #[tokio::test]
    async fn update_commits_stale_index_drop_and_rewrite_in_one_batch() {
        let (kv, col) = counting_col();
        col.create(&doc("d1", "idle", 5)).await.unwrap();
        kv.batches.store(0, Ordering::SeqCst);

        col.update(&doc("d1", "running", 9)).await.unwrap();
        assert_eq!(
            kv.batches.load(Ordering::SeqCst),
            1,
            "update must be a single atomic batch"
        );
        assert_eq!(
            (
                kv.puts.load(Ordering::SeqCst),
                kv.deletes.load(Ordering::SeqCst)
            ),
            (0, 0),
            "update must not fall back to raw per-key writes"
        );
        // the state index moved atomically: running sees the doc, idle does not
        assert_eq!(
            ids(&query(&col, Filter::and().expr(Expr::eq("state", "running"))).await),
            vec!["d1"]
        );
        assert_eq!(
            query(&col, Filter::and().expr(Expr::eq("state", "idle")))
                .await
                .count,
            0
        );
        assert_eq!(col.find("d1").await.unwrap().timestamp, 9);
    }

    /// `query_all` is the exhaustive read: `query` bounds its rows by the page
    /// size while `count` still reports the whole match set, so any recovery
    /// or cleanup path that reads rows must go through `query_all`.
    #[tokio::test]
    async fn query_all_reads_past_the_page_limit_in_order() {
        let (_kv, col) = counting_col();
        for (id, ts) in [("d2", 2i64), ("d1", 1), ("d3", 3)] {
            col.create(&doc(id, "idle", ts)).await.unwrap();
        }

        let page = col
            .query(&Query::new().limit(1).order("timestamp", Sort::Asc))
            .await
            .unwrap();
        assert_eq!(
            (page.count, ids(&page).len()),
            (3, 1),
            "one page, three matches"
        );

        let all = col
            .query_all(&Query::new().limit(1).order("timestamp", Sort::Asc))
            .await
            .unwrap();
        assert_eq!(
            all.iter().map(|d| d.id.as_str()).collect::<Vec<_>>(),
            vec!["d1", "d2", "d3"],
            "query_all must return every match in order_by order"
        );

        let all = col.query_all(&Query::new().limit(2)).await.unwrap();
        assert_eq!(
            all.iter().map(|d| d.id.as_str()).collect::<Vec<_>>(),
            vec!["d1", "d2", "d3"],
            "without order_by, an exhaustive read is id-ascending"
        );
    }

    /// `delete_all` must remove every matching row together with its index
    /// rows — a limit-capped cleanup leaves orphan rows that keep answering
    /// queries for documents that are gone.
    #[tokio::test]
    async fn delete_all_removes_every_match_and_its_index_rows() {
        let (_kv, col) = counting_col();
        let col = Arc::new(col);
        for (id, state) in [("d1", "idle"), ("d2", "gone"), ("d3", "gone")] {
            col.create(&doc(id, state, 1)).await.unwrap();
        }

        col.delete_all(Some(&Filter::and().expr(Expr::eq("state", "gone"))))
            .await
            .unwrap();

        assert_eq!(
            ids_of(&col, "state", json!("gone")).await,
            Vec::<String>::new()
        );
        assert_eq!(ids_of(&col, "state", json!("idle")).await, vec!["d1"]);
        assert_eq!(ids_of(&col, "timestamp", json!(1)).await, vec!["d1"]);
        assert!(col.find("d2").await.is_err());
        assert!(col.find("d3").await.is_err());
    }

    #[tokio::test]
    async fn delete_removes_data_and_indexes_in_one_batch() {
        let (kv, col) = counting_col();
        col.create(&doc("d1", "idle", 5)).await.unwrap();
        kv.batches.store(0, Ordering::SeqCst);

        col.delete("d1").await.unwrap();
        assert_eq!(
            kv.batches.load(Ordering::SeqCst),
            1,
            "delete must be a single atomic batch"
        );
        assert_eq!(
            (
                kv.puts.load(Ordering::SeqCst),
                kv.deletes.load(Ordering::SeqCst)
            ),
            (0, 0),
            "delete must not fall back to raw per-key writes"
        );
        assert!(col.find("d1").await.is_err(), "data row must be gone");
        for filter in ["state", "timestamp"] {
            let page = query(&col, Filter::and().expr(Expr::eq(filter, "idle"))).await;
            assert_eq!(page.count, 0, "no index row may survive the delete");
        }
    }

    #[tokio::test]
    async fn page_reads_use_batched_kv_reads() {
        let (kv, col) = counting_col();
        for i in 0..5 {
            col.create(&doc(&format!("d{i}"), "idle", i)).await.unwrap();
        }
        kv.mgets.store(0, Ordering::SeqCst);

        let page = col.query(&Query::new().limit(2).offset(1)).await.unwrap();
        assert_eq!(ids(&page), vec!["d1", "d2"]);
        assert_eq!(
            kv.mgets.load(Ordering::SeqCst),
            1,
            "implicit-id pagination must use one logical mget"
        );
    }

    #[tokio::test]
    async fn and_filter_stops_on_empty_branch() {
        let (kv, col) = counting_col();
        col.create(&doc("d1", "idle", 5)).await.unwrap();
        kv.scans.store(0, Ordering::SeqCst);

        let page = query(
            &col,
            Filter::and()
                .expr(Expr::eq("state", "idle"))
                .expr(Expr::eq("state", "missing"))
                .expr(Expr::eq("timestamp", 5)),
        )
        .await;
        assert_eq!(page.count, 0);
        assert_eq!(
            kv.scans.load(Ordering::SeqCst),
            2,
            "an empty AND branch must prevent evaluation of later branches"
        );
    }

    #[tokio::test]
    async fn empty_and_filter_is_empty_result() {
        let (_kv, col) = counting_col();
        col.create(&doc("d1", "idle", 5)).await.unwrap();

        let page = col
            .query(&Query::new().filter(Filter::and()))
            .await
            .unwrap();
        assert_eq!(page.count, 0);
        assert!(page.rows.is_empty());
    }

    // ========== concurrent mutation of one document ==========

    /// KV store that can be switched, from the test thread, to block inside the
    /// two store calls a collection mutation makes around its index
    /// computation: the document read (`get`) and the data+index write
    /// (`batch`). `entered` counts every caller that reached either gate, so a
    /// test can prove that a second mutation did NOT reach the store.
    #[derive(Default)]
    struct GatedKv {
        inner: crate::store::MemoryStore,
        gate: std::sync::atomic::AtomicBool,
        entered: std::sync::atomic::AtomicUsize,
    }

    impl GatedKv {
        fn ordered(&self) -> bool {
            self.gate.load(std::sync::atomic::Ordering::SeqCst)
        }

        fn arm(&self) {
            self.gate.store(true, std::sync::atomic::Ordering::SeqCst);
        }

        fn disarm(&self) {
            self.gate.store(false, std::sync::atomic::Ordering::SeqCst);
        }

        fn entered(&self) -> usize {
            self.entered.load(std::sync::atomic::Ordering::SeqCst)
        }

        async fn wait_entered(&self, entered: usize) {
            while self.entered() < entered {
                tokio::task::yield_now().await;
            }
        }

        async fn park(&self) {
            self.entered
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            while self.ordered() {
                tokio::time::sleep(std::time::Duration::from_millis(1)).await;
            }
        }
    }

    #[async_trait::async_trait]
    impl KvStore for GatedKv {
        async fn one(&self, key: &str) -> crate::Result<Option<Vec<u8>>> {
            if self.ordered() {
                self.park().await;
            }
            self.inner.one(key).await
        }

        async fn put(&self, key: &str, value: Vec<u8>) -> crate::Result<()> {
            if self.ordered() {
                self.park().await;
            }
            self.inner.put(key, value).await
        }

        async fn delete(&self, key: &str) -> crate::Result<()> {
            if self.ordered() {
                self.park().await;
            }
            self.inner.delete(key).await
        }

        async fn batch(&self, ops: &[crate::store::StoreBatchOp]) -> crate::Result<()> {
            if self.ordered() {
                self.park().await;
            }
            self.inner.batch(ops).await
        }

        async fn scan_prefix(
            &self,
            key: &str,
            options: crate::store::ScanOptions,
        ) -> crate::Result<Vec<(String, Vec<u8>)>> {
            self.inner.scan_prefix(key, options).await
        }
    }

    async fn ids_of(col: &Arc<KvCollection<Doc>>, field: &str, value: JsonValue) -> Vec<String> {
        let page = col
            .query(&Query::new().filter(Filter::and().expr(Expr::eq(field, value))))
            .await
            .unwrap();
        ids(&page)
    }

    /// Two concurrent updates of one document must not compute their index
    /// drops from the same stored version: the second waits for the first
    /// batch (`lock_docs`), so the data row and its index rows always describe
    /// one version — the value the row holds is indexed and the value it
    /// dropped is not.
    #[tokio::test(flavor = "multi_thread")]
    async fn concurrent_updates_do_not_tear_data_and_index_rows() {
        let kv = Arc::new(GatedKv::default());
        let col = Arc::new(KvCollection::<Doc>::new("docs", kv.clone()));
        col.create(&doc("d1", "a", 1)).await.unwrap();

        kv.arm();
        let first = {
            let col = col.clone();
            tokio::spawn(async move { col.update(&doc("d1", "b", 1)).await })
        };
        kv.wait_entered(1).await;

        let second = {
            let col = col.clone();
            tokio::spawn(async move { col.update(&doc("d1", "a", 2)).await })
        };
        // Give the second update every chance to run: it must wait for the
        // document lock instead of reading the version the first one mutates.
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        assert_eq!(
            kv.entered(),
            1,
            "a concurrent update read the document version the first one is mutating"
        );

        kv.disarm();
        first.await.unwrap().unwrap();
        second.await.unwrap().unwrap();

        let row = col.find("d1").await.unwrap();
        assert_eq!(
            ids_of(&col, "state", json!(row.state)).await,
            vec!["d1"],
            "the value the stored row holds must stay indexed"
        );
        let dropped = if row.state == "a" { "b" } else { "a" };
        assert!(
            ids_of(&col, "state", json!(dropped)).await.is_empty(),
            "a value the stored row dropped must not stay indexed"
        );
        assert_eq!(
            ids_of(&col, "timestamp", json!(row.timestamp)).await,
            vec!["d1"],
            "every indexed field must agree with the stored row"
        );
    }

    /// A delete and an update racing on one document must end in one of the two
    /// clean states — the updated document with its indexes, or nothing at all.
    /// Never a data row whose index entries describe a value it no longer
    /// holds, nor index rows whose document is gone (a query would return a
    /// phantom id).
    #[tokio::test(flavor = "multi_thread")]
    async fn concurrent_delete_and_update_leave_no_orphan_index_rows() {
        let kv = Arc::new(GatedKv::default());
        let col = Arc::new(KvCollection::<Doc>::new("docs", kv.clone()));
        col.create(&doc("d1", "a", 1)).await.unwrap();

        kv.arm();
        let deleter = {
            let col = col.clone();
            tokio::spawn(async move { col.delete("d1").await })
        };
        kv.wait_entered(1).await;

        let updater = {
            let col = col.clone();
            tokio::spawn(async move { col.update(&doc("d1", "b", 2)).await })
        };
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        assert_eq!(
            kv.entered(),
            1,
            "the update read the document while the delete was mutating it"
        );

        kv.disarm();
        deleter.await.unwrap().unwrap();
        updater.await.unwrap().unwrap();

        match col.find("d1").await {
            Ok(row) => {
                assert_eq!(
                    ids_of(&col, "state", json!(row.state)).await,
                    vec!["d1"],
                    "the surviving row must be indexed under its own value"
                );
                assert_eq!(
                    ids_of(&col, "timestamp", json!(row.timestamp)).await,
                    vec!["d1"]
                );
                for (field, value) in [
                    ("state", json!("a")),
                    ("state", json!("b")),
                    ("timestamp", json!(1)),
                    ("timestamp", json!(2)),
                ] {
                    if field == "state" && value == json!(row.state) {
                        continue;
                    }
                    if field == "timestamp" && value == json!(row.timestamp) {
                        continue;
                    }
                    assert!(
                        ids_of(&col, field, value.clone()).await.is_empty(),
                        "index rows of a dropped value survived the delete/update race: {field}={value}"
                    );
                }
            }
            Err(_) => {
                for (field, value) in [
                    ("state", json!("a")),
                    ("state", json!("b")),
                    ("timestamp", json!(1)),
                    ("timestamp", json!(2)),
                ] {
                    assert!(
                        ids_of(&col, field, value.clone()).await.is_empty(),
                        "a deleted document left index rows behind: {field}={value}"
                    );
                }
            }
        }
    }

    /// `create` landing on an id that already holds a document — the engine's
    /// find-then-create upserts, or two racing creators — must drop the index
    /// rows of the version it replaces, exactly like `update`. Leaving them
    /// behind would answer queries for a value the stored row no longer holds.
    #[tokio::test(flavor = "multi_thread")]
    async fn concurrent_creates_do_not_leave_stale_index_rows() {
        let kv = Arc::new(GatedKv::default());
        let col = Arc::new(KvCollection::<Doc>::new("docs", kv.clone()));
        col.create(&doc("d1", "a", 1)).await.unwrap();

        kv.arm();
        let first = {
            let col = col.clone();
            tokio::spawn(async move { col.create(&doc("d1", "b", 2)).await })
        };
        kv.wait_entered(1).await;

        let second = {
            let col = col.clone();
            tokio::spawn(async move { col.create(&doc("d1", "a", 3)).await })
        };
        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        assert_eq!(
            kv.entered(),
            1,
            "a concurrent create read the document version the first one is replacing"
        );

        kv.disarm();
        first.await.unwrap().unwrap();
        second.await.unwrap().unwrap();

        let row = col.find("d1").await.unwrap();
        for (field, value) in [
            ("state", json!("a")),
            ("state", json!("b")),
            ("timestamp", json!(1)),
            ("timestamp", json!(2)),
            ("timestamp", json!(3)),
        ] {
            let expected = (field == "state" && value == json!(row.state))
                || (field == "timestamp" && value == json!(row.timestamp));
            assert_eq!(
                ids_of(&col, field, value.clone()).await,
                if expected { vec!["d1"] } else { Vec::new() },
                "index and data row disagree after concurrent creates: {field}={value}"
            );
        }
    }

    /// The lock registry holds one entry per document being mutated right now,
    /// not one per document the process has ever touched: an entry is dropped
    /// by the last holder to release it, so a long-running engine that writes
    /// unboundedly many ids does not grow a lock map with them.
    #[tokio::test(flavor = "multi_thread")]
    async fn doc_lock_registry_holds_only_in_flight_documents() {
        let kv = Arc::new(GatedKv::default());
        // A prefix no sibling test uses: the registry is process-wide and keyed
        // by the store key, so every collection over the same key shares one
        // lock (conservative for two stores with equal keys, and required when
        // two handles address one database).
        let col = Arc::new(KvCollection::<Doc>::new("doclockregdocs", kv.clone()));
        let ids: Vec<String> = (0..256).map(|i| format!("d{i}")).collect();
        let key = |id: &str| col.data_key(id);
        let held = |key: &str| super::DOC_LOCKS.entries.lock().contains_key(key);

        for id in &ids {
            col.create(&doc(id, "idle", 1)).await.unwrap();
        }
        assert!(
            ids.iter().all(|id| !held(&key(id))),
            "a released document lock must not stay registered"
        );

        // positive control: while a mutation is in flight its key IS registered
        kv.arm();
        let parked = {
            let col = col.clone();
            let id = ids[0].clone();
            tokio::spawn(async move { col.update(&doc(&id, "running", 2)).await })
        };
        kv.wait_entered(1).await;
        assert!(
            held(&key(&ids[0])),
            "the document being mutated must be registered"
        );
        kv.disarm();
        parked.await.unwrap().unwrap();

        assert!(
            ids.iter().all(|id| !held(&key(id))),
            "the registry must be back to empty once every mutation released its lock"
        );

        // Waiters hold the entry they queued on, so cleanup has to happen in
        // the LAST releaser of each key — including when many mutations of one
        // document overlap, and when the keys are all distinct.
        let mut tasks = Vec::new();
        for i in 0..32 {
            let col = col.clone();
            let ids = ids.clone();
            tasks.push(tokio::spawn(async move {
                for id in ids {
                    col.update(&doc(&id, "running", 3 + i)).await.unwrap();
                }
            }));
        }
        for task in tasks {
            task.await.unwrap();
        }
        assert!(
            ids.iter().all(|id| !held(&key(id))),
            "overlapping mutations must leave no entry behind"
        );
    }
}