kglite 0.16.15

//! Text-index lifecycle on a graph — the engine-side primitives behind every
//! binding's `build_text_index` / `drop_text_index`.
//!
//! Re-exported as [`kglite::api::text_indexes`](crate::api::text_indexes). The
//! query half needs no surface at all: ranking happens inside Cypher, so a
//! binding that can call `cypher_query` can already search.
//!
//! **Where the graph meets the index.**
//! [`TextIndex`](crate::graph::algorithms::text_index::TextIndex) is
//! deliberately petgraph-ignorant — a document is a caller-assigned `u32` slot.
//! This module owns the other half: which property is indexed, how a node
//! becomes a document, and when the index has to be torn down.
//!
//! **Slot identity: the node index itself.** A document's slot *is*
//! `NodeIndex::index()`, so there is no second mapping to keep in step with the
//! graph. That is a deliberate departure from
//! [`EmbeddingStore`](crate::graph::schema::EmbeddingStore)'s contiguous-slot
//! layout, which exists because its vectors live in one flat `Vec<f32>` that
//! must stay hole-free; the text index keeps its documents in a hash map, where
//! a sparse key space costs nothing. The mapping that does not exist is the
//! mapping that cannot go stale — and a stale node↔slot map is exactly the
//! ghost-hit bug class this lane has to avoid. `NodeIndex` is `u32`-backed in
//! every graph this crate builds, so the cast is total.
//!
//! **The key is the spelling, not the resolution.** An index is keyed
//! `(node_type, property)` with the property spelled the way the caller spelled
//! it, exactly as an embedding store is (see [`crate::graph::embeddings`]). The
//! *value* is read through the alias resolution a `MATCH` uses, so
//! `build_text_index("Person", "name")` on a type whose title column is `name`
//! indexes the titles — and the resolved field is recorded on the store so a
//! later refresh cannot read a different column than the build did.
//!
//! **Explicit build, incremental catch-up.** Building is opt-in. After that the
//! index does not follow writes *eagerly* — it records that they happened, and
//! folds them in at query entry when the outstanding delta is small enough
//! ([`crate::graph::index_freshness`] owns that bookkeeping; this module owns
//! the re-read). What a write costs an unindexed graph is one branch, and what
//! it costs an indexed one is a slot comparison — no tokenization on the ingest
//! path, ever. Folding is not free per document, though — a splice into every
//! posting list the document's terms appear in, so it grows with the corpus —
//! and past the measured crossover a catch-up rebuilds the index instead
//! ([`rebuild_beats_folding`]).
//!
//! Deletes are the exception and are *not* staleness — a slot freed by
//! `StableDiGraph` is handed to the next node created, so a document left
//! behind would be inherited and score as its new owner's content. Deletion
//! prunes immediately, at the delete site, and the *rollback* of a delete is
//! what puts the slot back in the dirty set.
//!
//! **Memory and mapped only.** The disk backend refuses: a heap-resident
//! inverted index over a Wikidata-scale disk graph is the RAM cliff that
//! backend exists to avoid, and disk does not persist the HNSW index either.

use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{RwLock, RwLockReadGuard};

use petgraph::graph::NodeIndex;

use crate::graph::algorithms::text_index::bm25::{PreparedQuery, ScoredDoc};
use crate::graph::algorithms::text_index::TextIndex;
use crate::graph::dir_graph::DirGraph;
use crate::graph::index_freshness::{FreshnessDelta, IndexFreshness};
use crate::graph::schema::InternedKey;
use crate::graph::storage::{GraphRead, StrField};

/// What a [`build_text_index`] call indexed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TextIndexReport {
    /// Documents in the index — nodes whose property held a string. An empty
    /// string counts: it indexes as an empty document and participates in the
    /// corpus statistics.
    pub indexed: usize,
    /// Nodes of the type that produced no document, because the property was
    /// absent or held a non-string value.
    pub skipped: usize,
    /// Distinct terms in the built vocabulary.
    pub terms: usize,
}

/// The index key for a `(node_type, property)` pair.
///
/// The one place the key is minted, so build, drop, lookup and `SHOW INDEXES`
/// agree on the spelling.
pub fn index_key(node_type: &str, property: &str) -> (String, String) {
    (node_type.to_string(), property.to_string())
}

/// One node type's BM25 index over one property, plus what it needs to stay
/// tied to the graph.
///
/// See the module docs for why the node↔document mapping is the identity.
///
/// **Why the index is behind a lock.** Catch-up happens at *query* entry, and a
/// query holds `&DirGraph` — `execute_read` cannot take `&mut`, and the Python
/// wheel hands out `Arc<DirGraph>`, so there is no route by which a read could
/// mutate the map this store lives in. The `RwLock` is that route, and it is
/// also what makes the double-checked refresh correct under the parallel
/// runtime. Lock order is documented on
/// [`crate::graph::index_freshness`]: this lock first, the freshness state
/// second, never the reverse.
#[derive(Debug)]
pub struct TextIndexStore {
    index: RwLock<TextIndex>,
    /// Bumped by every [`Self::refresh`] that claimed a delta.
    ///
    /// Term ids are recycled, so a [`PreparedQuery`] resolved before a refresh
    /// can name a *different* term after it — the query would then score
    /// documents it has no word in common with. A holder that cannot keep the
    /// read guard for its whole scoring pass (the Cypher scalar cannot: the
    /// executor is shared across rayon regions and therefore must be `Sync`,
    /// while `RwLockReadGuard` is `!Send`) stamps its prepared query with this
    /// number and re-prepares when it moves.
    ///
    /// Read it *after* acquiring the read guard, never before: a refresh bumps
    /// it while holding the write lock, so a guard held across the comparison
    /// is what makes the answer still true when the score is computed.
    generation: AtomicU64,
    /// What has changed since the last build or refresh.
    freshness: IndexFreshness,
    /// The alias-resolved field the build read — `"title"` where the caller
    /// named a type's title column, otherwise the property itself. Recorded so
    /// a refresh reads the same column even if the alias map moved under it,
    /// and so a `SET` can be told apart from a write to any other property
    /// without re-resolving per row.
    resolved_field: String,
    /// Nodes of the type the build produced no document for. Kept because it is
    /// the difference between "your corpus is 900 documents" and "100 of your
    /// nodes are invisible to search", which a document count alone hides.
    ///
    /// Restated by a rebuild and by nothing else: a *fold* moves documents in
    /// and out through the delta, and never revisits the nodes it saw no
    /// document for, so it cannot count them. Both routes into a rebuild
    /// restate it — [`build_text_index`] and the rebuild arm of
    /// [`Self::refresh`] — which is why it is behind an atomic: the arm runs
    /// on the read path, under `&self`.
    skipped: AtomicUsize,
}

impl Clone for TextIndexStore {
    /// Deep, never shared. `Clone` backs snapshots, transactions and
    /// `independent_copy`, and each of those must be able to delete a node
    /// without pruning the other's document — the fork independence the
    /// lifecycle tests pin.
    fn clone(&self) -> Self {
        Self {
            index: RwLock::new(self.index().clone()),
            generation: AtomicU64::new(self.generation()),
            freshness: self.freshness.clone(),
            resolved_field: self.resolved_field.clone(),
            skipped: AtomicUsize::new(self.skipped()),
        }
    }
}

/// A borrowed, consistent view of one text index.
///
/// Term ids are interned per index and are recycled when a term loses its last
/// posting, so a [`PreparedQuery`] is only meaningful against the index state
/// it was prepared from. Holding this guard across prepare-and-score is what
/// keeps a concurrent refresh from renumbering the dictionary underneath a
/// query, so a scoring loop that can hold it takes it once rather than once
/// per row.
///
/// A loop that *cannot* hold it — the Cypher `text_bm25` scalar cannot, because
/// its executor is shared across rayon regions and must be `Sync` while this
/// guard is `!Send` — re-acquires per row and compares
/// [`TextIndexStore::generation`] against the number its query was prepared
/// under, re-preparing when they differ.
pub struct TextIndexRead<'a>(RwLockReadGuard<'a, TextIndex>);

impl TextIndexRead<'_> {
    /// The whole index behind this view — what persistence writes. Everything
    /// else here is a query-shaped question; this is the one caller that wants
    /// the structure itself, and it wants it under the same guard so a
    /// concurrent refresh cannot renumber the dictionary mid-encode.
    pub(crate) fn index(&self) -> &TextIndex {
        &self.0
    }

    /// Tokenize and resolve a query string against this index's dictionary.
    pub fn prepare_query(&self, query: &str) -> PreparedQuery {
        self.0.prepare_query(query)
    }

    /// BM25 score of one node, or `None` when it has no document.
    ///
    /// The `None`/`Some(0.0)` split is the whole point: "this row is not in the
    /// index" and "this row is indexed and shares no term with the query" are
    /// different answers, and only the caller knows whether the first should
    /// surface as null.
    pub fn score(&self, node: NodeIndex, query: &PreparedQuery) -> Option<f64> {
        let slot = TextIndexStore::slot(node);
        self.0.contains_doc(slot).then(|| self.0.score(slot, query))
    }

    /// The `k` best-scoring nodes, best first.
    pub fn top_k(&self, query: &PreparedQuery, k: usize) -> Vec<(NodeIndex, f64)> {
        self.0
            .top_k(query, k)
            .into_iter()
            .map(|ScoredDoc { slot, score }| (NodeIndex::new(slot as usize), score))
            .collect()
    }

    /// Whether this node has a document.
    pub fn contains_node(&self, node: NodeIndex) -> bool {
        self.0.contains_doc(TextIndexStore::slot(node))
    }

    /// Documents in the corpus.
    pub fn documents(&self) -> usize {
        self.0.total_docs()
    }

    /// Whether the corpus is empty — the query-side short circuit.
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
}

impl TextIndexStore {
    /// The document slot for a node.
    ///
    /// `NodeIndex` is `u32`-backed (petgraph's `DefaultIx`) in every backend
    /// this crate constructs, so no graph can produce an index this truncates.
    #[inline]
    fn slot(node: NodeIndex) -> u32 {
        node.index() as u32
    }

    fn index(&self) -> RwLockReadGuard<'_, TextIndex> {
        self.index.read().unwrap_or_else(|e| e.into_inner())
    }

    /// A consistent read view. Hold it for as long as one query's
    /// prepare-and-score sequence lasts — see [`TextIndexRead`].
    pub fn read(&self) -> TextIndexRead<'_> {
        TextIndexRead(self.index())
    }

    /// How many times this index has been refreshed into a different shape.
    /// See the field docs for why a query-side cache has to watch it.
    pub fn generation(&self) -> u64 {
        self.generation.load(Ordering::Acquire)
    }

    /// The alias-resolved field this index reads.
    pub fn resolved_field(&self) -> &str {
        &self.resolved_field
    }

    /// The catch-up state to persist beside the index. An index that covers a
    /// prefix of its type has to record what it has yet to cover, or a reload
    /// presents a stale index as a current one.
    pub(crate) fn freshness_state(&self) -> &IndexFreshness {
        &self.freshness
    }

    /// Documents in the index.
    pub fn documents(&self) -> usize {
        self.index().total_docs()
    }

    /// Nodes of the type that carried no indexable string when the index was
    /// last built. See the field docs for why a fold cannot restate it.
    pub fn skipped(&self) -> usize {
        self.skipped.load(Ordering::Relaxed)
    }

    /// Distinct terms in the vocabulary.
    pub fn terms(&self) -> usize {
        self.index().vocabulary_len()
    }

    /// Approximate heap footprint of the index, in bytes.
    pub fn estimated_bytes(&self) -> usize {
        self.index().estimated_bytes()
    }

    /// Whether this node has a document.
    pub fn contains_node(&self, node: NodeIndex) -> bool {
        self.read().contains_node(node)
    }

    /// Documents the next [`TextIndexStore::refresh`] would re-read. O(1), and
    /// an upper bound — see [`crate::graph::index_freshness`].
    pub fn delta_size(&self, graph: &DirGraph) -> usize {
        self.freshness.delta_size(node_bound(graph))
    }

    /// Whether the graph has moved since this index last covered it.
    pub fn is_stale(&self, graph: &DirGraph) -> bool {
        self.freshness.is_stale(node_bound(graph))
    }

    /// The inline-refresh ceiling: a delta at or under this is cheap enough to
    /// fold in at query entry.
    pub fn auto_refresh_limit(&self) -> usize {
        self.freshness.limit()
    }

    /// Whether the outstanding delta is within the inline-refresh ceiling.
    /// `false` for a clean index — there is nothing to fold in.
    pub fn can_auto_refresh(&self, graph: &DirGraph) -> bool {
        self.freshness.within_limit(node_bound(graph))
    }

    /// Drop a node's document and every posting that mentions it. Returns
    /// whether it had one.
    ///
    /// **Every node deletion must reach this.** A document left on a freed
    /// `NodeIndex` is inherited by the next node created, which then scores as
    /// content it never had.
    pub fn remove_node(&mut self, node: NodeIndex) -> bool {
        self.index
            .get_mut()
            .unwrap_or_else(|e| e.into_inner())
            .remove_doc(Self::slot(node))
    }

    /// Mark a slot as needing a re-read on the next refresh.
    ///
    /// The rollback path's whole undo story: a delete pruned the document, and
    /// if the statement is reversed the node comes back with its text while the
    /// document does not. Marking the slot makes the next refresh restore it.
    pub(crate) fn note_slot_changed(&self, node: NodeIndex) {
        self.freshness.note_changed(Self::slot(node));
    }

    /// Tokenize and resolve a query string against this index's dictionary.
    ///
    /// Convenience over [`TextIndexStore::read`], which is what a per-row
    /// scoring loop should hold instead: the term ids in the returned query are
    /// only valid against the index state this call saw.
    pub fn prepare_query(&self, query: &str) -> PreparedQuery {
        self.read().prepare_query(query)
    }

    /// BM25 score of one node, or `None` when it has no document.
    pub fn score(&self, node: NodeIndex, query: &PreparedQuery) -> Option<f64> {
        self.read().score(node, query)
    }

    /// The `k` best-scoring nodes, best first.
    pub fn top_k(&self, query: &PreparedQuery, k: usize) -> Vec<(NodeIndex, f64)> {
        self.read().top_k(query, k)
    }

    /// Check every internal invariant of the underlying index.
    pub fn validate(&self) -> Result<(), String> {
        self.index().validate()
    }

    /// Fold every outstanding change into the index and mark it current.
    /// Returns how many slots were re-read — the delta for a fold, every
    /// member of the type for a rebuild.
    ///
    /// **Always refreshes when called** — the threshold is the *caller's*
    /// policy, asked through [`Self::can_auto_refresh`]. The one refusal is a
    /// read-only graph, where an index catching up would be the one write a
    /// read-only handle performed.
    ///
    /// Each slot is re-read through the field the *build* resolved, so an alias
    /// map that moved since cannot silently repoint the index at another
    /// column. A slot whose node is gone, has changed type, or no longer holds
    /// a string has its document removed — `add_doc` is an upsert, so a changed
    /// one is simply overwritten.
    ///
    /// **Cost: bounded by a rebuild, not by delta x corpus.** Folding one
    /// document splices into the postings list of each of its terms, and those
    /// lists are as long as their terms are common, so per-document fold cost
    /// grows with the corpus (measured: 0.075-0.17 ms at 20k documents,
    /// 0.36-0.60 ms at 100k). Past [`FOLD_SLOTS_PER_REBUILD`] documents that
    /// overtakes rebuilding the index outright, and this rebuilds instead —
    /// see [`rebuild_beats_folding`] for the measurement and the constant.
    /// Only the slots that would actually touch the index are counted, so a
    /// bulk load of an unrelated node type never buys a rebuild.
    ///
    /// The two arms produce the same index and the same scores — the bulk
    /// builder and `add_doc` are deliberately separate code paths, and the
    /// freshness tests assert them against each other — but they are not
    /// silent about which ran: the rebuild arm re-reads every node of the type
    /// and therefore restates [`Self::skipped`], which a fold cannot.
    pub fn refresh(&self, graph: &DirGraph, node_type: &str) -> usize {
        if graph.read_only {
            return 0;
        }
        // The index lock is taken *before* the delta is claimed: that ordering
        // is what makes a second reader either wait for this refresh or find
        // nothing left to do. See `index_freshness`'s lock-order note.
        let mut index = self.index.write().unwrap_or_else(|e| e.into_inner());
        let Some(delta) = self.freshness.take_delta(node_bound(graph)) else {
            return 0;
        };
        let field_key = InternedKey::from_str(&self.resolved_field);
        let type_key = InternedKey::from_str(node_type);

        let splices = delta
            .slots()
            .filter(|slot| {
                // A slot the fold would splice: a node of the indexed type is
                // an upsert, and an indexed slot that is no longer one is a
                // removal. Anything else costs one type lookup and no postings
                // work at all.
                graph.graph.node_type_of(NodeIndex::new(*slot as usize)) == Some(type_key)
                    || index.contains_doc(*slot)
            })
            .count();
        let seen = if rebuild_beats_folding(splices) {
            self.rebuild(graph, node_type, &mut index, field_key)
        } else {
            self.fold(graph, node_type, &delta, &mut index, type_key, field_key)
        };

        // Bumped for every claimed delta, not for every changed document: the
        // rebuild arm replaces the dictionary wholesale and can still report
        // fewer slots read than the fold would have (a type whose members are
        // gone rebuilds to an empty index and reports zero), so a bump
        // conditioned on the returned count would leave a prepared query
        // resolving against term ids that no longer exist. Bumped under the
        // write lock, so a reader holding the read guard cannot observe the new
        // dictionary against the old number.
        self.generation.fetch_add(1, Ordering::Release);
        debug_assert!(
            index.validate().is_ok(),
            "a refreshed text index must satisfy its own invariants: {:?}",
            index.validate()
        );
        seen
    }

    /// Re-read the delta's slots into `index`. Returns how many it read.
    fn fold(
        &self,
        graph: &DirGraph,
        node_type: &str,
        delta: &FreshnessDelta,
        index: &mut TextIndex,
        type_key: InternedKey,
        field_key: InternedKey,
    ) -> usize {
        let mut seen = 0usize;
        for slot in delta.slots() {
            seen += 1;
            let node = NodeIndex::new(slot as usize);
            let indexed = match graph.graph.node_type_of(node) {
                Some(key) if key == type_key => graph
                    .graph
                    .node_view(node)
                    .map(|view| {
                        match view.resolved_field_str(node_type, &self.resolved_field, field_key) {
                            StrField::Str(text) => {
                                index.add_doc(slot, text.as_ref());
                                true
                            }
                            StrField::NotString | StrField::Absent => false,
                        }
                    })
                    .unwrap_or(false),
                _ => false,
            };
            if !indexed {
                index.remove_doc(slot);
            }
        }
        seen
    }

    /// Replace `index` with one built from every node of the type. Returns how
    /// many nodes it read.
    ///
    /// The same bulk build [`build_text_index`] runs, reading the field this
    /// index resolved at *its* build rather than re-resolving — a refresh may
    /// not repoint the index at another column. The one thing it does not
    /// borrow is that function's empty-index error: a type whose every string
    /// has since been cleared is a legitimately empty index here, and it is
    /// exactly what folding the same delta would have produced.
    fn rebuild(
        &self,
        graph: &DirGraph,
        node_type: &str,
        index: &mut TextIndex,
        field_key: InternedKey,
    ) -> usize {
        let members = graph
            .type_indices
            .get(node_type)
            .map(|nodes| nodes.to_vec())
            .unwrap_or_default();
        let mut skipped = 0usize;
        *index = TextIndex::build(members.iter().filter_map(|node| {
            let view = graph.graph.node_view(*node)?;
            match view.resolved_field_str(node_type, &self.resolved_field, field_key) {
                StrField::Str(text) => Some((Self::slot(*node), text)),
                StrField::NotString | StrField::Absent => {
                    skipped += 1;
                    None
                }
            }
        }));
        self.skipped.store(skipped, Ordering::Relaxed);
        members.len()
    }
}

/// Slots one refresh may fold in before rebuilding the whole index is the
/// cheaper of the two.
///
/// Measured 2026-08-25 (release profile, three agreeing runs) — see
/// [`rebuild_beats_folding`] for the derivation.
pub(crate) const FOLD_SLOTS_PER_REBUILD: usize = 1500;

/// Whether rebuilding the index outright costs less than splicing `splices`
/// documents into it one at a time.
///
/// **The measurement (2026-08-25, release, three agreeing runs).** Folding is
/// not O(delta): [`TextIndex::add_doc`] inserts into each of the document's
/// per-term postings lists, and a list is as long as its term's document
/// frequency, so one folded document costs 0.075-0.17 ms over a 20k-document
/// corpus and 0.36-0.60 ms over a 100k one — five times the corpus, five times
/// the per-document cost. A full rebuild of the 100k corpus takes ~865 ms.
///
/// **Why the corpus is not a parameter.** Both sides scale linearly with the
/// corpus — folding through the postings lengths, rebuilding through the
/// documents it re-reads — so the corpus cancels and the break-even is an
/// absolute number of documents: 865 ms / 0.36-0.60 ms = 1440-2400 folds at
/// 100k, and 173 ms / 0.075-0.17 ms = 1020-2300 at 20k. The constant is the
/// middle of that band. A proportional rule ("delta above 2% of the corpus")
/// was rejected against the same table: it fires at 460 folds on a 20k corpus,
/// where folding measured five times cheaper than the rebuild it would buy.
///
/// It is a cost *model*, not a tuned constant — it picks between two paths that
/// produce the same index, so being wrong costs time and never an answer. Wrong
/// low, a refresh pays at most one extra rebuild's worth of folding (1500 x
/// 0.6 ms ~ 900 ms against 865 ms); wrong high, it pays a rebuild where folding
/// would have been up to 1.6x cheaper. The worst case that matters is the one
/// this bounds: without the switch, folding a 100k-document delta into a
/// 100k-document corpus costs ~10 hours.
fn rebuild_beats_folding(splices: usize) -> bool {
    splices > FOLD_SLOTS_PER_REBUILD
}

/// The graph's node-slot bound, as the document slot space sees it.
#[inline]
fn node_bound(graph: &DirGraph) -> u32 {
    GraphRead::node_bound(&graph.graph) as u32
}

/// A node of `node_type` was created at `node` — the recycled-slot check.
///
/// Reached only through
/// [`index_freshness::write_hooks`](crate::graph::index_freshness::write_hooks),
/// which has already established that this graph has at least one text index.
pub(crate) fn note_node_created(graph: &DirGraph, node: NodeIndex, node_type: &str) {
    let slot = TextIndexStore::slot(node);
    for ((indexed_type, _), store) in &graph.text_indexes {
        store
            .freshness
            .note_created(slot, indexed_type == node_type);
    }
}

/// A node's property was written. `field` is the alias-resolved field, or
/// `None` from a caller that wrote several and did not decompose them.
///
/// The field comparison is the discrimination that keeps an ordinary `SET` off
/// the dirty set: writing `n.updated_at` on a type whose `body` is indexed
/// changes nothing the index holds.
pub(crate) fn note_property_written(
    graph: &DirGraph,
    node: NodeIndex,
    node_type: &str,
    field: Option<&str>,
) {
    let slot = TextIndexStore::slot(node);
    for ((indexed_type, _), store) in &graph.text_indexes {
        if indexed_type != node_type {
            continue;
        }
        if field.is_none_or(|written| written == store.resolved_field) {
            store.freshness.note_changed(slot);
        }
    }
}

/// Build (or rebuild) a BM25 index over `property` for every node of
/// `node_type`.
///
/// Idempotent: a second call replaces the index wholesale, which is also how a
/// stale index is refreshed in one step. The property is read through the same
/// alias resolution a `MATCH` filter uses, so a type's id/title column can be
/// indexed under the name the loader gave it.
///
/// `auto_refresh_limit` bounds the delta a *query* will fold in inline before
/// it serves stale results instead — a count, not a duration: what a delta
/// costs depends on the corpus it folds into, bounded above by one rebuild
/// ([`rebuild_beats_folding`]). `None` keeps whatever the existing index
/// used, or
/// [`DEFAULT_AUTO_REFRESH_LIMIT`](crate::graph::index_freshness::DEFAULT_AUTO_REFRESH_LIMIT)
/// for a first build.
///
/// **What is skipped.** A node whose property is absent or holds a non-string
/// produces no document and therefore never scores — a stringified number is
/// not a document, and indexing one would let a text query rank rows whose
/// property is not text at all. An **empty string is indexed**, as an empty
/// document: it is a document with no terms, not a missing one, and it counts
/// towards the corpus statistics.
///
/// Errors when the node type is unknown, when the graph is disk-backed, or —
/// on a type that has nodes — when not one of them yielded a document, which is
/// what a misspelled property looks like. A type with no nodes yet builds an
/// empty index rather than erroring, so an index can be declared before ingest.
pub fn build_text_index(
    graph: &mut DirGraph,
    node_type: &str,
    property: &str,
    auto_refresh_limit: Option<usize>,
) -> Result<TextIndexReport, String> {
    if GraphRead::is_disk(&graph.graph) {
        return Err(format!(
            "build_text_index('{node_type}', '{property}') is not supported on a disk-backed \
             graph: the BM25 index is heap-resident, and building one over a graph sized for \
             the disk backend is the memory cliff that backend exists to avoid. Use the \
             default (in-memory) or 'mapped' storage mode."
        ));
    }
    if !graph.has_node_type(node_type) {
        return Err(format!(
            "Unknown node type '{node_type}'. build_text_index() indexes one node type's \
             property; list the graph's node types to see what exists."
        ));
    }

    let field = graph.resolve_alias(node_type, property).to_string();
    let key = InternedKey::from_str(&field);
    let nodes = graph
        .type_indices
        .get(node_type)
        .map(|members| members.to_vec())
        .unwrap_or_default();

    // Streamed into the bulk builder rather than collected first: a corpus's
    // worth of owned `String`s would double peak memory for the duration of
    // the build, and every one of them is dropped immediately after interning.
    let mut skipped = 0usize;
    let index = TextIndex::build(nodes.iter().filter_map(|node_idx| {
        let view = graph.graph.node_view(*node_idx)?;
        match view.resolved_field_str(node_type, &field, key) {
            StrField::Str(text) => Some((TextIndexStore::slot(*node_idx), text)),
            StrField::NotString | StrField::Absent => {
                skipped += 1;
                None
            }
        }
    }));

    if index.total_docs() == 0 && !nodes.is_empty() {
        return Err(format!(
            "No '{node_type}' node carries a string value for '{property}' — all {} were \
             absent or non-string, so there is nothing to index. Check the spelling, and note \
             that BM25 indexes text: a numeric or list-valued property is not indexable.",
            nodes.len()
        ));
    }

    debug_assert!(
        index.validate().is_ok(),
        "a freshly built text index must satisfy its own invariants: {:?}",
        index.validate()
    );
    let key_pair = index_key(node_type, property);
    // A rebuild keeps the ceiling its author set; only an explicit argument
    // moves it, so refreshing an index does not quietly restore the default.
    let limit = auto_refresh_limit.or_else(|| {
        graph
            .text_indexes
            .get(&key_pair)
            .map(|existing| existing.auto_refresh_limit())
    });
    let store = TextIndexStore {
        index: RwLock::new(index),
        generation: AtomicU64::new(0),
        freshness: IndexFreshness::covering(node_bound(graph), limit),
        resolved_field: field,
        skipped: AtomicUsize::new(skipped),
    };
    let report = TextIndexReport {
        indexed: store.documents(),
        skipped,
        terms: store.terms(),
    };
    graph.text_indexes.insert(key_pair, store);
    graph.bump_version();
    Ok(report)
}

/// Install an index restored from a `.kgl` section, with the freshness state,
/// resolved field and skipped count it was saved with.
///
/// Persistence-only. Every other route into `graph.text_indexes` goes through
/// [`build_text_index`], which reads the graph and therefore covers it by
/// construction; this one is handed an index it must take on trust, so the
/// decoder validates the payload before calling in here.
///
/// `skipped` is carried rather than recomputed on purpose: it is a *build-time*
/// count of nodes that produced no document, and nothing short of a rebuild can
/// restate it — resetting it to zero would turn "100 of your nodes are
/// invisible to search" into a claim that none are.
pub(crate) fn attach_persisted_text_index(
    graph: &mut DirGraph,
    node_type: &str,
    property: &str,
    index: TextIndex,
    freshness: IndexFreshness,
    resolved_field: String,
    skipped: usize,
) {
    graph.text_indexes.insert(
        index_key(node_type, property),
        TextIndexStore {
            index: RwLock::new(index),
            generation: AtomicU64::new(0),
            freshness,
            resolved_field,
            skipped: AtomicUsize::new(skipped),
        },
    );
}

/// Fold every outstanding change into the text index over
/// `(node_type, property)`, returning how many slots it re-read.
///
/// `None` when no such index exists. This is the refresh driver a query entry
/// calls once it has decided the delta is worth folding in — the decision
/// itself is [`TextIndexStore::can_auto_refresh`].
pub fn refresh_text_index(graph: &DirGraph, node_type: &str, property: &str) -> Option<usize> {
    let store = graph.text_indexes.get(&index_key(node_type, property))?;
    Some(store.refresh(graph, node_type))
}

/// Drop the text index for `(node_type, property)`. Returns whether one
/// existed.
pub fn drop_text_index(graph: &mut DirGraph, node_type: &str, property: &str) -> bool {
    let removed = graph
        .text_indexes
        .remove(&index_key(node_type, property))
        .is_some();
    if removed {
        graph.bump_version();
    }
    removed
}

/// The text index over `(node_type, property)`, if one is built.
///
/// The read-side counterpart of [`has_text_index`].
///
/// Scans rather than hashing, deliberately: the map is keyed by an owned
/// `(String, String)`, so a hash lookup from two `&str`s has to *mint* that key
/// — two allocations, on a path the Cypher scalar takes once per row. A graph
/// carries a handful of text indexes at most, and comparing a handful of short
/// string pairs costs less than one allocation.
pub fn text_index_store<'a>(
    graph: &'a DirGraph,
    node_type: &str,
    property: &str,
) -> Option<&'a TextIndexStore> {
    graph
        .text_indexes
        .iter()
        .find(|((indexed_type, indexed_property), _)| {
            indexed_type == node_type && indexed_property == property
        })
        .map(|(_, store)| store)
}

/// Whether a text index is built over `(node_type, property)`.
pub fn has_text_index(graph: &DirGraph, node_type: &str, property: &str) -> bool {
    graph
        .text_indexes
        .contains_key(&index_key(node_type, property))
}

/// Every text index on the graph, sorted by `(node_type, property)`.
///
/// The one enumeration order, so `SHOW INDEXES` and any binding-side listing
/// cannot disagree about it.
pub fn list_text_indexes(graph: &DirGraph) -> Vec<(&str, &str, &TextIndexStore)> {
    let mut out: Vec<(&str, &str, &TextIndexStore)> = graph
        .text_indexes
        .iter()
        .map(|((node_type, property), store)| (node_type.as_str(), property.as_str(), store))
        .collect();
    out.sort_unstable_by_key(|(node_type, property, _)| (*node_type, *property));
    out
}

#[cfg(test)]
#[path = "text_indexes_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "text_indexes_freshness_tests.rs"]
mod freshness_tests;