kevy-embedded 6.4.0

//! Embedded secondary-index API (server parity
//! minus FIELDS hydration, which exists to save wire round-trips the
//! embedded caller doesn't have — read fields with `hget` directly).
//!
//! Placement: each shard's `Inner` carries its slice of every index
//! (index-follows-key, same as the server), maintained inside
//! `commit_write` under the shard lock the write already holds — the
//! synchronous-derivation guarantee costs no extra locking. Key
//! extraction from the logged argv is EXACT (a precise multi-key
//! table, not the feed filter's fail-open heuristic): a missed update
//! would be index drift, which derived-by-construction forbids.
//!
//! Backfill: `idx_create` builds synchronously, shard by shard, each
//! under its own write lock (the hook holds the same lock, so there is
//! no race window per shard). No `Building` state embedded — create
//! returns when the index serves.

// The sidecar IS the catalog's persistence — `boot` reads it and a
// directory without one "boots empty". So a rename that fails loses
// the index definitions at the next start, after the command that
// created them has already replied OK. That is a gap, not a
// non-event, and it is written up as an open question rather than
// silently accepted here: .claude/OPEN-QUESTIONS-6.4.md §3.
#![expect(
    clippy::let_underscore_must_use,
    reason = "the catalog has no other home; see .claude/OPEN-QUESTIONS-6.4.md"
)]

use crate::{KevyError, KevyResult};
use std::io;
use std::sync::RwLock;

use kevy_index::{
    Catalog, Cursor, IndexKind, IndexSpec, IndexValue, Segment, SegmentStats, ValType,
};

use crate::store::{Store, lock_write};

pub(crate) use crate::ops_index_sync::{each_written_key_pub, on_commit, sync_segs};

/// One page of index hits plus the cursor to resume from.
pub type IndexPage = (Vec<(Vec<u8>, IndexValue)>, Option<Cursor>);

/// One field's highlight: its name and the `(start, end)` match spans.
#[cfg(feature = "text")]
pub type FieldSpans = (Vec<u8>, Vec<(u32, u32)>);
/// A highlighted MATCH hit: key, score, and per-field [`FieldSpans`].
#[cfg(feature = "text")]
pub type HighlightedHit = (Vec<u8>, f64, Vec<FieldSpans>);

// `idx_match_with`, its clause options and the span-mapping helper live
// in a child module to keep this file under the 500-LOC ceiling; the
// text-index specifics — declaring a multi-field one, and gathering the
// corpus statistics a global BM25 scores against — live in another.
#[cfg(feature = "text")]
#[path = "ops_index_highlight.rs"]
pub(crate) mod highlight;

// The clause-carrying scalar query (capacity arc G1) and the
// [`ValueFilter`] predicate shape it shares with MATCH — independent of
// the `text` feature: a range index filters fine without a tokenizer.
#[path = "ops_index_claused.rs"]
pub(crate) mod claused;

// The auto-declaration loop's observation face (refusal log + advice).
#[path = "ops_index_advise.rs"]
pub(crate) mod advise;

// The read-only admin surface (stats, enumeration).
#[path = "ops_index_admin.rs"]
mod admin;

#[cfg(feature = "text")]
#[path = "ops_index_text.rs"]
mod text;

// The embedded MATCH's cold seams (windowed tables' frozen buckets).
#[cfg(feature = "text")]
#[path = "ops_index_text_cold.rs"]
pub(crate) mod text_cold;

/// Sort merged `(value, key)` hits, cut to `limit`, and derive the
/// resume cursor. Shared by `Store::idx_query` and the transaction twin
/// on `AtomicAllShards`, which differ only in where the segments come
/// from — the pagination has to agree exactly or a cursor taken inside
/// a transaction would not resume outside one.
pub(crate) fn merge_page(mut all: Vec<(IndexValue, Vec<u8>)>, limit: usize) -> IndexPage {
    all.sort();
    all.truncate(limit);
    let next = if all.len() == limit {
        all.last().map(|(v, k)| Cursor { value: v.clone(), key: k.clone() })
    } else {
        None
    };
    (all.into_iter().map(|(v, k)| (k, v)).collect(), next)
}

/// Store-level index state: catalog + a version stamp the per-shard
/// segment lists sync against, and each declared path's usage cell
/// (the refusal log's dual — reclaim-face raw material).
#[derive(Debug, Default)]
pub(crate) struct IndexReg {
    pub(crate) catalog: RwLock<(u64, Catalog)>,
    pub(crate) usage:
        RwLock<std::collections::HashMap<Vec<u8>, std::sync::Arc<kevy_index::UsageCell>>>,
}

/// A shard's window runtime beside its segment — `()` stand-in on
/// wasm, where the cold tier compiles out.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) type WinRef<'a> = Option<&'a kevy_window::WindowRt>;
#[cfg(target_arch = "wasm32")]
pub(crate) type WinRef<'a> = Option<&'a core::convert::Infallible>;

/// Per-shard segment list, kept inside `Inner` (guarded by the shard
/// lock).
#[derive(Debug, Default)]
pub(crate) struct ShardSegs {
    pub(crate) version: u64,
    pub(crate) segs: Vec<(IndexSpec, Segment)>,
    /// Inverted segments for KIND text specs (parallel list —
    /// a spec appears in exactly one of the lists).
    #[cfg(feature = "text")]
    pub(crate) text: Vec<(IndexSpec, kevy_text::TextSegment)>,
    /// HNSW graphs for KIND ann specs.
    #[cfg(feature = "vector")]
    pub(crate) ann: Vec<(IndexSpec, kevy_vector::Hnsw)>,
    /// Aggregate segments for KIND agg specs.
    pub(crate) agg: Vec<(IndexSpec, kevy_index::AggSegment)>,
    /// Sliding-window runtimes, name-keyed — reconciled and slid by
    /// the reaper's window tick; empty under a manual reaper (nothing
    /// slides, everything stays hot, queries need no cold half).
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) windows: Vec<(Vec<u8>, kevy_window::WindowRt)>,
    /// A windowed table's text indexes' cold directories, name-keyed —
    /// reconciled and fed by the same window tick (the driver's
    /// eviction batch freezes out of every same-table text index).
    #[cfg(all(feature = "text", not(target_arch = "wasm32")))]
    pub(crate) cold_text: Vec<(Vec<u8>, kevy_window::TextColdDir)>,
    /// `reserved_bytes` generation cache: set by every
    /// segment-mutating chokepoint (`on_commit` applies, list
    /// rebuilds, FLUSH resets) via [`ShardSegs::mark_stats_dirty`];
    /// an idle tick reads the cached sum instead of walking every
    /// segment's stats. Compiled with the tier backend only — on
    /// targets without it (wasm) nothing reads the cache.
    #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
    pub(crate) stats_dirty: bool,
    #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
    pub(crate) reserved_cache: u64,
}

impl ShardSegs {
    /// The window runtime for `name`, if that index is a windowed
    /// table's window access path AND a tick has reconciled it (only
    /// the reaper's window tick populates the list, so a manual-reaper
    /// or memory-only store always answers `None` — nothing slid).
    #[cfg(not(target_arch = "wasm32"))]
    pub(crate) fn window_of(&self, name: &[u8]) -> Option<&kevy_window::WindowRt> {
        self.windows.iter().find(|(n, _)| n == name).map(|(_, w)| w)
    }

    /// Invalidate the `reserved_bytes` cache — a no-op on targets
    /// without the tier backend, so mutation chokepoints call it
    /// unconditionally.
    #[inline]
    pub(crate) fn mark_stats_dirty(&mut self) {
        #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
        {
            self.stats_dirty = true;
        }
    }
}

#[cfg(feature = "persist")]
const SIDECAR: &str = "index-catalog.meta";

impl Store {
    /// `IDX.CREATE` equivalent. Builds synchronously; errors on
    /// duplicate name / cap / bad spec.
    pub fn idx_create(
        &self,
        name: &[u8],
        prefix: &[u8],
        field: &[u8],
        ty: ValType,
        kind: IndexKind,
    ) -> KevyResult<()> {
        if prefix.is_empty() {
            return Err(KevyError::InvalidInput("empty prefix".into()));
        }
        #[cfg(not(feature = "text"))]
        if kind == IndexKind::Text {
            return Err(KevyError::Unsupported("text indexes need the `text` feature".into()));
        }
        #[cfg(not(feature = "vector"))]
        if kind == IndexKind::Ann {
            return Err(KevyError::Unsupported("vector indexes need the `vector` feature".into()));
        }
        let spec = IndexSpec {
            name: name.to_vec(),
            prefix: prefix.to_vec(),
            fields: vec![kevy_index::FieldSpec::new(field.to_vec())],
            ty,
            kind,
            max_bytes: 0,
            ann: None,
            group_by: None,
            with_positions: false,
            values: Vec::new(),
            composite: None,
        };
        self.register_spec(spec)
    }

    pub(crate) fn register_spec(&self, spec: IndexSpec) -> KevyResult<()> {
        // Tiering floor refusal: body in
        // `ops_index_sync::tier_floor_check` (500-LOC rule).
        #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
        crate::ops_index_sync::tier_floor_check(&self.shards)?;
        {
            let mut g =
                self.indexes.catalog.write().unwrap_or_else(std::sync::PoisonError::into_inner);
            let (ver, cat) = &mut *g;
            cat.create(spec).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
            *ver += 1;
        }
        self.persist_index_sidecar();
        self.advise_clear();
        self.usage_rekey();
        // Build every shard's slice now (each under its own lock).
        for shard in self.shards.iter() {
            let mut g = lock_write(shard);
            let inner = &mut *g;
            sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
        }
        Ok(())
    }

    /// Declare an ANN index (KIND ann, TYPE vector). `params.m`
    /// / `params.ef` of 0 select the defaults (16 / 200).
    #[cfg(feature = "vector")]
    pub fn idx_create_ann(
        &self,
        name: &[u8],
        prefix: &[u8],
        field: &[u8],
        params: kevy_index::AnnSpec,
    ) -> KevyResult<()> {
        if params.dim == 0 || params.distance > 2 {
            return Err(KevyError::InvalidInput("bad ann parameters".into()));
        }
        let spec = IndexSpec {
            name: name.to_vec(),
            prefix: prefix.to_vec(),
            fields: vec![kevy_index::FieldSpec::new(field.to_vec())],
            ty: ValType::Vector,
            kind: IndexKind::Ann,
            max_bytes: 0,
            ann: Some(kevy_index::AnnSpec {
                m: if params.m == 0 { 16 } else { params.m },
                ef: if params.ef == 0 { 200 } else { params.ef },
                ..params
            }),
            group_by: None,
            with_positions: false,
            values: Vec::new(),
            composite: None,
        };
        self.register_spec(spec)
    }

    /// `IDX.DROP` equivalent; `false` if absent. On a hit the catalog
    /// sidecar is re-persisted so the drop survives restart.
    pub fn idx_drop(&self, name: &[u8]) -> bool {
        let hit = {
            let mut g =
                self.indexes.catalog.write().unwrap_or_else(std::sync::PoisonError::into_inner);
            let (ver, cat) = &mut *g;
            let hit = cat.drop_index(name);
            if hit {
                *ver += 1;
            }
            hit
        };
        if hit {
            self.persist_index_sidecar();
            self.advise_clear();
            self.usage_rekey();
        }
        hit
    }

    /// `MATCH` — BM25-ranked hits merged across shards, scored against
    /// **global** corpus statistics so a hit's rank does not depend on
    /// which shard it landed on (see docs/text-search.md).
    ///
    /// Two query-time passes: the first sums each shard's `n_docs`,
    /// `total_len` and per-query-token `df` into one [`CorpusStats`]; the
    /// second scores every shard against it. Only the query's tokens'
    /// df is aggregated, not a whole-corpus table — the query narrows it.
    #[cfg(feature = "text")]
    pub fn idx_match(
        &self,
        name: &[u8],
        query: &[u8],
        limit: usize,
    ) -> KevyResult<Vec<(Vec<u8>, f64)>> {
        Ok(self
            .idx_match_with(name, query, limit, crate::MatchOpts::default())?
            .into_iter()
            .map(|(key, score, _)| (key, score))
            .collect())
    }

    /// Declare an aggregate index (KIND agg — write-time GROUP
    /// BY). `ty` must be numeric.
    pub fn idx_create_agg(
        &self,
        name: &[u8],
        prefix: &[u8],
        field: &[u8],
        ty: ValType,
        group_by: &[u8],
    ) -> KevyResult<()> {
        if !matches!(ty, ValType::I64 | ValType::F64) || group_by.is_empty() {
            return Err(KevyError::InvalidInput("agg requires numeric type + group field".into()));
        }
        let spec = IndexSpec {
            name: name.to_vec(),
            prefix: prefix.to_vec(),
            fields: vec![kevy_index::FieldSpec::new(field.to_vec())],
            ty,
            kind: IndexKind::Agg,
            max_bytes: 0,
            ann: None,
            group_by: Some(group_by.to_vec()),
            with_positions: false,
            values: Vec::new(),
            composite: None,
        };
        self.register_spec(spec)
    }

    /// One group's merged stats across shards.
    pub fn idx_group(&self, name: &[u8], group: &[u8]) -> KevyResult<kevy_index::GroupStats> {
        let mut merged = kevy_index::GroupStats { count: 0, sum: 0.0, min: None, max: None };
        let mut found = false;
        for shard in self.shards.iter() {
            let mut g = lock_write(shard);
            let inner = &mut *g;
            sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
            if let Some((_, a)) = inner.idx_segs.agg.iter().find(|(s, _)| s.name == name) {
                found = true;
                kevy_index::merge_group(&mut merged, &a.group(group));
            }
        }
        if !found {
            return Err(KevyError::NotFound("no such aggregate index".into()));
        }
        Ok(merged)
    }

    /// Top groups merged + ranked across shards.
    pub fn idx_groups(
        &self,
        name: &[u8],
        by: kevy_index::AggBy,
        limit: usize,
    ) -> KevyResult<Vec<(Vec<u8>, kevy_index::GroupStats)>> {
        let limit = limit.clamp(1, 1000);
        // HashMap merge (same O(rows×groups) trap the server reduce
        // had — hashing keeps it linear).
        let mut merged: std::collections::HashMap<Vec<u8>, kevy_index::GroupStats> =
            std::collections::HashMap::new();
        let mut found = false;
        for shard in self.shards.iter() {
            let mut g = lock_write(shard);
            let inner = &mut *g;
            sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
            if let Some((_, a)) = inner.idx_segs.agg.iter().find(|(s, _)| s.name == name) {
                found = true;
                for (gk, st) in a.all_groups() {
                    match merged.get_mut(&gk) {
                        Some(m) => kevy_index::merge_group(m, &st),
                        None => {
                            merged.insert(gk, st);
                        }
                    }
                }
            }
        }
        if !found {
            return Err(KevyError::NotFound("no such aggregate index".into()));
        }
        let mut ranked: Vec<(Vec<u8>, kevy_index::GroupStats)> = merged.into_iter().collect();
        kevy_index::sort_groups(&mut ranked, by);
        ranked.truncate(limit);
        Ok(ranked)
    }

    /// `KNN` — nearest neighbors merged ascending across shards.
    /// `ef` = query beam width (0 = engine default; recall knob).
    #[cfg(feature = "vector")]
    pub fn idx_knn(
        &self,
        name: &[u8],
        query: &[f32],
        k: usize,
        ef: usize,
    ) -> KevyResult<Vec<(Vec<u8>, f32)>> {
        let k = k.clamp(1, 1000);
        let mut all: Vec<(Vec<u8>, f32)> = Vec::new();
        let mut found = false;
        for shard in self.shards.iter() {
            let mut g = lock_write(shard);
            let inner = &mut *g;
            sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
            if let Some((_, graph)) = inner.idx_segs.ann.iter().find(|(s, _)| s.name == name) {
                found = true;
                all.extend(graph.knn(query, k, ef));
            }
        }
        if !found {
            return Err(KevyError::NotFound("no such vector index".into()));
        }
        all.sort_by(|a, b| a.1.total_cmp(&b.1).then_with(|| a.0.cmp(&b.0)));
        all.truncate(k);
        Ok(all)
    }

    /// Without `persist` there is no data dir — the catalog lives only
    /// in memory, so the sidecar halves are no-ops.
    #[cfg(not(feature = "persist"))]
    fn persist_index_sidecar(&self) {}

    #[cfg(not(feature = "persist"))]
    pub(crate) fn idx_boot(&self) {}

    fn for_each_segment(&self, name: &[u8], mut f: impl FnMut(&Segment)) -> KevyResult<()> {
        let mut found = false;
        for shard in self.shards.iter() {
            let mut g = lock_write(shard);
            let inner = &mut *g;
            sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
            if let Some((_, seg)) = inner.idx_segs.segs.iter().find(|(s, _)| s.name == name) {
                found = true;
                f(seg);
            }
        }
        if found { Ok(()) } else { Err(KevyError::NotFound("no such index".into())) }
    }

    #[cfg(feature = "persist")]
    fn persist_index_sidecar(&self) {
        let Some(dir) = &self.config.data_dir else { return };
        let g = self.indexes.catalog.read().unwrap_or_else(std::sync::PoisonError::into_inner);
        let tmp = dir.join("index-catalog.meta.tmp");
        if std::fs::write(&tmp, g.1.to_sidecar()).is_ok() {
            let _ = std::fs::rename(&tmp, dir.join(SIDECAR));
        }
    }

    /// Boot half — load a persisted catalog (indexes rebuild lazily on
    /// first touch via `sync_segs`).
    #[cfg(feature = "persist")]
    pub(crate) fn idx_boot(&self) {
        let Some(dir) = &self.config.data_dir else { return };
        if let Ok(text) = std::fs::read_to_string(dir.join(SIDECAR))
            && let Some(cat) = Catalog::from_sidecar(&text)
            && !cat.is_empty()
        {
            let mut g =
                self.indexes.catalog.write().unwrap_or_else(std::sync::PoisonError::into_inner);
            *g = (g.0 + 1, cat);
        }
    }
}