acts 0.22.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 serde::{Serialize, de::DeserializeOwned};
use serde_json::Value as JsonValue;
use std::{cmp::Ordering, collections::HashSet, fmt::Debug, marker::PhantomData, sync::Arc};

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

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

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

    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
            .get(&key)
            .await?
            .map(|data| serde_json::from_slice(&data).map_err(map_db_err))
            .transpose()
    }

    /// The mutations [`DbCollection::create`] would apply (data row + index
    /// rows), without applying them. Lets a caller fold several document
    /// writes — e.g. a model and its trigger rows on `deploy` — into one
    /// atomic [`KvStore::batch`].
    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)?;
        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.
    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)?;
        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.
    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>> {
        // Validate array-shaped operators up front: the error must not depend
        // on whether the field is indexed or which scan path is selected.
        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(),
                ));
            }
            _ => {}
        }
        // 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 = format!("{}{}id{}", self.prefix, KEY_SEP, KEY_SEP);
            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(|(_, bytes)| {
                    let v: JsonValue = serde_json::from_slice(bytes).ok()?;
                    let id = v.get("id")?.as_str()?.to_string();
                    if let Some(field_val) = v.get(&expr.key)
                        && expr.op(field_val, &expr.value)
                    {
                        return Some(id);
                    }
                    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
            }
        }
    }

    /// 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>> {
        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) => match filter.r#type {
                    FilterType::And => existing.intersection(&ids).cloned().collect(),
                    FilterType::Or => existing.union(&ids).cloned().collect(),
                },
            });
        }
        Ok(result.unwrap_or_default())
    }

    /// 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)?;
            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 are compared as f64; 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) {
        let fa = na.as_f64().unwrap_or_default();
        let fb = nb.as_f64().unwrap_or_default();
        if fa < fb {
            Ordering::Less
        } else if fa > fb {
            Ordering::Greater
        } else {
            Ordering::Equal
        }
    } 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.get(&key).await.map(|v| v.is_some())
    }

    async fn find(&self, id: &str) -> crate::Result<Self::Item> {
        let key = self.data_key(id);
        let data = self.kv.get(&key).await?.ok_or(ActError::Store(format!(
            "cannot find {} by '{}'",
            self.prefix, id
        )))?;
        let json: JsonValue = serde_json::from_slice(&data).map_err(map_db_err)?;
        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: HashSet<String> = if let Some(filter) = &q.filter {
            self.filter_ids(filter, indexed, q.get_order_by()).await?
        } else {
            // No filter – scan all data entries to collect all IDs
            let scan_key = format!("{}{}id{}", self.prefix, KEY_SEP, KEY_SEP);
            let options = ScanOptions::new(ScanOperation::Eq, scan_key.clone(), false);
            let entries = self.kv.scan_prefix(&scan_key, options).await?;
            entries
                .iter()
                .filter_map(|(_, bytes)| {
                    let v: JsonValue = serde_json::from_slice(bytes).ok()?;
                    v.get("id")?.as_str().map(|s| s.to_string())
                })
                .collect()
        };

        // ids always gather in ascending order: it is the page order when no
        // `order_by` is given, and the tie-break (a stable sort keeps the read
        // order) that makes offsets deterministic when sort keys repeat.
        let mut ids: Vec<String> = id_set.into_iter().collect();
        ids.sort();
        let count = ids.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 rows: Vec<T> = if order_by.is_empty() {
            let mut rows = Vec::new();
            for id in ids.into_iter().skip(q.offset).take(q.limit) {
                if let Some(json) = self.read_json(&id).await? {
                    rows.push(T::upcast(json)?);
                }
            }
            rows
        } else {
            let mut docs: Vec<JsonValue> = Vec::with_capacity(count);
            for id in &ids {
                if let Some(json) = self.read_json(id).await? {
                    docs.push(json);
                }
            }
            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> {
        // Data row and index rows are committed as one atomic batch, so a
        // mid-write failure can never leave a document without its indexes.
        let ops = self.create_ops(data)?;
        self.kv.batch(&ops).await?;
        Ok(true)
    }

    async fn update(&self, data: &Self::Item) -> crate::Result<bool> {
        // Stale index drops, the data row and the new index rows commit as
        // one atomic batch.
        let ops = self.update_ops(data).await?;
        self.kv.batch(&ops).await?;
        Ok(true)
    }

    async fn delete(&self, id: &str) -> crate::Result<bool> {
        // Index rows and the data row are removed as one atomic batch.
        let ops = self.delete_ops(id).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) {
                    if v1.is_f64() {
                        return v1.as_f64().unwrap() < v2.as_f64().unwrap_or_default();
                    } else if v1.is_i64() {
                        return v1.as_i64().unwrap() < v2.as_i64().unwrap_or_default();
                    } else if v1.is_u64() {
                        return v1.as_u64().unwrap() < v2.as_u64().unwrap_or_default();
                    }
                }
                false
            }
            ExprOp::LE => {
                if let (serde_json::Value::Number(v1), serde_json::Value::Number(v2)) = (l, r) {
                    if v1.is_f64() {
                        return v1.as_f64().unwrap() <= v2.as_f64().unwrap_or_default();
                    } else if v1.is_i64() {
                        return v1.as_i64().unwrap() <= v2.as_i64().unwrap_or_default();
                    } else if v1.is_u64() {
                        return v1.as_u64().unwrap() <= v2.as_u64().unwrap_or_default();
                    }
                }
                false
            }
            ExprOp::GT => {
                if let (serde_json::Value::Number(v1), serde_json::Value::Number(v2)) = (l, r) {
                    if v1.is_f64() {
                        return v1.as_f64().unwrap() > v2.as_f64().unwrap_or_default();
                    } else if v1.is_i64() {
                        return v1.as_i64().unwrap() > v2.as_i64().unwrap_or_default();
                    } else if v1.is_u64() {
                        return v1.as_u64().unwrap() > v2.as_u64().unwrap_or_default();
                    }
                }
                false
            }
            ExprOp::GE => {
                if let (serde_json::Value::Number(v1), serde_json::Value::Number(v2)) = (l, r) {
                    if v1.is_f64() {
                        return v1.as_f64().unwrap() >= v2.as_f64().unwrap_or_default();
                    } else if v1.is_i64() {
                        return v1.as_i64().unwrap() >= v2.as_i64().unwrap_or_default();
                    } else if v1.is_u64() {
                        return v1.as_u64().unwrap() >= v2.as_u64().unwrap_or_default();
                    }
                }
                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::{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);
    }

    // ========== 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_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.get(&legacy).await.unwrap().is_none(),
            "legacy key removed"
        );
        assert!(
            kv.get(&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()
    }

    #[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,
        puts: AtomicUsize,
        deletes: AtomicUsize,
    }

    #[async_trait::async_trait]
    impl KvStore for CountingKv {
        async fn get(&self, key: &str) -> crate::Result<Option<Vec<u8>>> {
            self.inner.get(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 scan_prefix(
            &self,
            key: &str,
            options: ScanOptions,
        ) -> crate::Result<Vec<(String, Vec<u8>)>> {
            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);
    }

    #[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");
        }
    }
}