hayson-engine 0.11.0

use std::collections::{HashMap, HashSet};
use std::io;

use serde_json::Value;

use crate::catalog::{CatalogIndexKind, CatalogPredicate, ExpandedIndex};
use crate::query::predicates::{test_predicate, ComparisonOp, Predicate};
use crate::schema::{HaysonRecord, IndexKind, TableSchema};
use crate::storage::{LogFile, LogOp, LogSet, ScanResult, Slot};

use super::composite_index::{CompositeExactIndex, CompositeSortedIndex, RegistryIndex, RegistryIndexKind};
use super::exact_index::ExactIndex;
use super::fulltext_index::{matches_terms, FulltextIndex, SearchMode};
use super::sorted_index::RangeBounds;
use super::tokenize::tokenize_value;

/// Fields of `schema` carrying their own engine-managed virtual-path index (see
/// `FieldSpec::has_virtual_path`) — every `image` field and every `markdown` one that did not
/// declare `addressable: false`.
fn virtual_path_fields_of(schema: &TableSchema) -> Vec<String> {
    schema
        .fields
        .iter()
        .filter(|(_, spec)| spec.has_virtual_path())
        .map(|(name, _)| name.clone())
        .collect()
}

fn virtual_path_of(record: &HaysonRecord, field: &str) -> Option<String> {
    record
        .get(field)?
        .as_object()?
        .get("virtualPath")?
        .as_str()
        .map(str::to_string)
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PrimaryEntry {
    pub offset: u64,
    pub length: u64,
    pub seq: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PartitionBytes {
    pub live_bytes: u64,
    pub dead_bytes: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RegistryStats {
    pub records: usize,
    pub live_bytes: u64,
    pub dead_bytes: u64,
}

struct HeldIndex {
    exact: Option<CompositeExactIndex>,
    sorted: Option<CompositeSortedIndex>,
}

fn sugar_index_name(table: &str, field: &str, kind: &str) -> String {
    format!("{table}_{field}_{kind}")
}

fn sugar_indexes_of(schema: &TableSchema) -> Vec<RegistryIndex> {
    let mut sugar = Vec::new();
    for (field, spec) in &schema.fields {
        if let Some(kind) = spec.index {
            let (kind_name, kind_value) = match kind {
                IndexKind::Exact => ("exact", RegistryIndexKind::Exact),
                IndexKind::Sorted => ("sorted", RegistryIndexKind::Sorted),
            };
            sugar.push(RegistryIndex {
                name: sugar_index_name(&schema.table, field, kind_name),
                fields: vec![field.clone()],
                kind: kind_value,
                unique: spec.unique,
            });
        }
        if spec.fulltext {
            sugar.push(RegistryIndex {
                name: sugar_index_name(&schema.table, field, "fulltext"),
                fields: vec![field.clone()],
                kind: RegistryIndexKind::Fulltext,
                unique: false,
            });
        }
    }
    sugar
}

fn projected_fields(
    schema: &TableSchema,
    indexes: &[RegistryIndex],
    where_filters: &HashMap<String, Vec<Predicate>>,
) -> HashSet<String> {
    let mut names = HashSet::new();
    names.insert(schema.primary_key.clone());
    for index in indexes {
        for field in &index.fields {
            names.insert(field.clone());
        }
    }
    for predicates in where_filters.values() {
        for predicate in predicates {
            names.insert(predicate.field.clone());
        }
    }
    for field in virtual_path_fields_of(schema) {
        names.insert(field);
    }
    names
}

fn registry_kind_of(kind: CatalogIndexKind) -> Option<RegistryIndexKind> {
    match kind {
        CatalogIndexKind::Exact => Some(RegistryIndexKind::Exact),
        CatalogIndexKind::Sorted => Some(RegistryIndexKind::Sorted),
        CatalogIndexKind::Fulltext => Some(RegistryIndexKind::Fulltext),
        // NOTE: ttl indexes are an expiry sweep, not a lookup structure -- nothing to hold here.
        CatalogIndexKind::Ttl => None,
    }
}

fn comparison_op_of(op: &str) -> Option<ComparisonOp> {
    match op {
        "=" => Some(ComparisonOp::Eq),
        "!=" => Some(ComparisonOp::Ne),
        ">" => Some(ComparisonOp::Gt),
        ">=" => Some(ComparisonOp::Gte),
        "<" => Some(ComparisonOp::Lt),
        "<=" => Some(ComparisonOp::Lte),
        "in" => Some(ComparisonOp::In),
        "contains" => Some(ComparisonOp::Contains),
        "startsWith" => Some(ComparisonOp::StartsWith),
        _ => None,
    }
}

fn predicates_of(clauses: &[CatalogPredicate]) -> Vec<Predicate> {
    clauses
        .iter()
        .filter_map(|clause| {
            comparison_op_of(&clause.op).map(|op| Predicate::new(clause.field.clone(), op, clause.value.clone()))
        })
        .collect()
}

fn record_matches_where(record: &HaysonRecord, predicates: &[Predicate]) -> bool {
    predicates.iter().all(|predicate| test_predicate(record, predicate))
}

/// Merges `schema`'s own field-modifier sugar with a catalog's `declared` list (typically
/// `catalog::expand_indexes`'s output, itself already sugar-plus-declared and deduplicated) into
/// the `RegistryIndex` list this registry holds, plus the `where`-scoped subset's filter
/// predicates, keyed by index name. `declared` wins on a name collision, but a collision is not
/// expected in practice: `catalog::expand_indexes` names its own sugar with the exact same
/// `sugar_index_name` scheme this module derives independently, so the two already agree.
/// Passing an empty `declared` (every caller with no catalog in hand) falls back to schema sugar
/// alone — the same set [`TableRegistry::new`] builds.
fn registry_indexes_of(
    schema: &TableSchema,
    declared: &[ExpandedIndex],
) -> (Vec<RegistryIndex>, HashMap<String, Vec<Predicate>>) {
    let mut indexes: Vec<RegistryIndex> = sugar_indexes_of(schema);
    let mut by_name: HashSet<String> = indexes.iter().map(|index| index.name.clone()).collect();
    let mut where_filters = HashMap::new();
    for expanded in declared {
        let Some(kind) = registry_kind_of(expanded.kind) else {
            continue;
        };
        if let Some(clauses) = &expanded.where_ {
            let predicates = predicates_of(clauses);
            if !predicates.is_empty() {
                where_filters.insert(expanded.name.clone(), predicates);
            }
        }
        let registry_index = RegistryIndex {
            name: expanded.name.clone(),
            fields: expanded.fields.clone(),
            kind,
            unique: expanded.unique,
        };
        if by_name.insert(expanded.name.clone()) {
            indexes.push(registry_index);
        } else if let Some(existing) = indexes.iter_mut().find(|index| index.name == expanded.name) {
            *existing = registry_index;
        }
    }
    (indexes, where_filters)
}

fn held_of(index: &RegistryIndex) -> HeldIndex {
    let exact = match index.kind {
        RegistryIndexKind::Exact | RegistryIndexKind::Sorted => Some(CompositeExactIndex::new(&index.fields)),
        RegistryIndexKind::Fulltext => None,
    };
    let sorted = match index.kind {
        RegistryIndexKind::Sorted => Some(CompositeSortedIndex::new(&index.fields)),
        _ => None,
    };
    HeldIndex { exact, sorted }
}

pub struct TableRegistry {
    pub schema: TableSchema,
    primary: HashMap<String, PrimaryEntry>,
    arrival: HashMap<String, u64>,
    next_arrival: u64,
    parts_of: HashMap<String, String>,
    projections: HashMap<String, HaysonRecord>,
    indexes: Vec<RegistryIndex>,
    held: HashMap<String, HeldIndex>,
    fulltext_indexes: Vec<RegistryIndex>,
    fulltext: FulltextIndex,
    by_field_exact: HashMap<String, String>,
    #[allow(dead_code)]
    by_field_sorted: HashMap<String, String>,
    projected: HashSet<String>,
    bytes: HashMap<String, (u64, u64)>,
    max_seq: u64,
    virtual_paths: HashMap<String, ExactIndex>,
    where_filters: HashMap<String, Vec<Predicate>>,
}

impl TableRegistry {
    pub fn new(schema: TableSchema) -> Self {
        let indexes = sugar_indexes_of(&schema);
        Self::with_indexes(schema, indexes)
    }

    pub fn with_indexes(schema: TableSchema, indexes: Vec<RegistryIndex>) -> Self {
        Self::with_indexes_and_filters(schema, indexes, HashMap::new())
    }

    /// Builds a registry from the catalog's fully expanded index list for this table
    /// (`catalog::expand_indexes`) instead of the schema-only sugar [`TableRegistry::new`] derives
    /// on its own — the entry point a composite or `where`-scoped declared index reaches this
    /// registry through, mirroring the TS engine's `Table.open({ indexes })`.
    pub fn with_declared_indexes(schema: TableSchema, declared: &[ExpandedIndex]) -> Self {
        let (indexes, where_filters) = registry_indexes_of(&schema, declared);
        Self::with_indexes_and_filters(schema, indexes, where_filters)
    }

    fn with_indexes_and_filters(
        schema: TableSchema,
        indexes: Vec<RegistryIndex>,
        where_filters: HashMap<String, Vec<Predicate>>,
    ) -> Self {
        let mut held = HashMap::new();
        let mut by_field_exact = HashMap::new();
        let mut by_field_sorted = HashMap::new();
        for index in &indexes {
            let entry = held_of(index);
            // NOTE: only a single-field index of the whole table answers a lookup phrased as one field.
            if index.fields.len() == 1 && entry.exact.is_some() {
                let field = index.fields[0].clone();
                by_field_exact
                    .entry(field.clone())
                    .or_insert_with(|| index.name.clone());
                if entry.sorted.is_some() {
                    by_field_sorted.entry(field).or_insert_with(|| index.name.clone());
                }
            }
            held.insert(index.name.clone(), entry);
        }
        let fulltext_indexes: Vec<RegistryIndex> = indexes
            .iter()
            .filter(|i| i.kind == RegistryIndexKind::Fulltext)
            .cloned()
            .collect();
        let projected = projected_fields(&schema, &indexes, &where_filters);
        let virtual_paths = virtual_path_fields_of(&schema)
            .into_iter()
            .map(|field| (field, ExactIndex::new()))
            .collect();

        TableRegistry {
            schema,
            primary: HashMap::new(),
            arrival: HashMap::new(),
            next_arrival: 0,
            parts_of: HashMap::new(),
            projections: HashMap::new(),
            indexes,
            held,
            fulltext_indexes,
            fulltext: FulltextIndex::new(),
            by_field_exact,
            by_field_sorted,
            projected,
            bytes: HashMap::new(),
            max_seq: 0,
            virtual_paths,
            where_filters,
        }
    }

    #[allow(dead_code)]
    pub fn build_from_log(schema: TableSchema, log: &LogFile) -> io::Result<Self> {
        let mut registry = TableRegistry::new(schema);
        let scan = log.scan(0)?;
        registry.apply_scan(scan);
        Ok(registry)
    }

    /// Same as [`TableRegistry::build_from_log`], but the registry is built with
    /// [`TableRegistry::with_declared_indexes`] instead of [`TableRegistry::new`].
    #[allow(dead_code)]
    pub fn build_from_log_with_declared(
        schema: TableSchema,
        log: &LogFile,
        declared: &[ExpandedIndex],
    ) -> io::Result<Self> {
        let mut registry = TableRegistry::with_declared_indexes(schema, declared);
        let scan = log.scan(0)?;
        registry.apply_scan(scan);
        Ok(registry)
    }

    /// Same as [`TableRegistry::build_from_log_with_declared`], for a table whose records are
    /// spread over one log file per partition value — every partition is scanned in name order and
    /// its entries are recorded against the partition they came from.
    ///
    /// Takes the set by `&mut` because a scan that meets a half-written tail also has to get rid of
    /// it ([`LogFile::discard_partial_tail`]): the registry may skip those bytes, but leaving them
    /// where the next append starts is what turns one interrupted write into an unreadable log.
    pub fn build_from_log_set_with_declared(
        schema: TableSchema,
        logs: &mut LogSet,
        declared: &[ExpandedIndex],
    ) -> io::Result<Self> {
        let mut registry = TableRegistry::with_declared_indexes(schema, declared);
        for part in logs.parts() {
            let Some(log) = logs.log_of_mut(&part) else { continue };
            let scan = log.scan(0)?;
            log.discard_partial_tail(&scan)?;
            registry.apply_scan_in(scan, &part);
        }
        Ok(registry)
    }

    #[allow(dead_code)]
    pub fn apply_scan(&mut self, scan: ScanResult) {
        self.apply_scan_in(scan, "");
    }

    pub fn apply_scan_in(&mut self, scan: ScanResult, part: &str) {
        for scanned in scan.entries {
            match scanned.entry.meta.op {
                LogOp::Put => {
                    let record = scanned.entry.data.unwrap_or_default();
                    self.apply_put_in(
                        &scanned.entry.meta.key,
                        &record,
                        Slot {
                            offset: scanned.slot.offset,
                            length: scanned.slot.length,
                        },
                        scanned.entry.meta.seq,
                        part,
                    );
                }
                LogOp::Del => {
                    self.apply_delete_in(
                        &scanned.entry.meta.key,
                        scanned.slot.length,
                        scanned.entry.meta.seq,
                        part,
                    );
                }
            }
        }
    }

    pub fn records(&self) -> usize {
        self.primary.len()
    }

    pub fn max_seq(&self) -> u64 {
        self.max_seq
    }

    pub fn stats(&self) -> RegistryStats {
        let mut live_bytes = 0;
        let mut dead_bytes = 0;
        for (live, dead) in self.bytes.values() {
            live_bytes += live;
            dead_bytes += dead;
        }
        RegistryStats {
            records: self.primary.len(),
            live_bytes,
            dead_bytes,
        }
    }

    pub fn stats_of(&self, part: &str) -> PartitionBytes {
        let (live, dead) = self.bytes.get(part).copied().unwrap_or((0, 0));
        PartitionBytes {
            live_bytes: live,
            dead_bytes: dead,
        }
    }

    fn tally(&mut self, part: &str) -> &mut (u64, u64) {
        self.bytes.entry(part.to_string()).or_insert((0, 0))
    }

    pub fn has_fulltext(&self) -> bool {
        !self.fulltext_indexes.is_empty()
    }

    pub fn indexes(&self) -> &[RegistryIndex] {
        &self.indexes
    }

    /// The fields a whole-table single-field `exact`/`sorted` index covers — the ones a unit of
    /// work's overlay can mirror. A partial index is skipped on purpose: it only holds the records
    /// its `where` admits, so an overlay built on it would answer "no such staged record" for a
    /// staged record that simply falls outside the filter.
    pub fn indexed_fields(&self) -> Vec<String> {
        let mut fields = HashSet::new();
        for index in &self.indexes {
            if index.fields.len() != 1 || self.where_of(index).is_some() {
                continue;
            }
            if index.kind != RegistryIndexKind::Exact && index.kind != RegistryIndexKind::Sorted {
                continue;
            }
            fields.insert(index.fields[0].clone());
        }
        fields.into_iter().collect()
    }

    pub fn fulltext(&self) -> &FulltextIndex {
        &self.fulltext
    }

    pub fn keys(&self) -> impl Iterator<Item = &String> {
        self.primary.keys()
    }

    #[allow(dead_code)]
    pub fn entries(&self) -> impl Iterator<Item = (&String, &PrimaryEntry)> {
        self.primary.iter()
    }

    pub fn slot_of(&self, key: &str) -> Option<PrimaryEntry> {
        self.primary.get(key).copied()
    }

    /// Where `key` sits in this table's insertion order: the rank it took when it first entered the
    /// registry, which an overwrite deliberately leaves alone. Mirrors the position a key holds in
    /// the TS engine's `Map`-backed primary registry, where `set` on an existing key keeps it.
    pub fn arrival_of(&self, key: &str) -> Option<u64> {
        self.arrival.get(key).copied()
    }

    /// Every live key, oldest arrival first.
    pub fn keys_in_arrival_order(&self) -> Vec<&String> {
        let mut keys: Vec<&String> = self.primary.keys().collect();
        keys.sort_unstable_by_key(|key| (self.arrival.get(*key).copied().unwrap_or(0), *key));
        keys
    }

    pub fn part_of(&self, key: &str) -> Option<&str> {
        self.parts_of.get(key).map(String::as_str)
    }

    pub fn projection_of(&self, key: &str) -> Option<&HaysonRecord> {
        self.projections.get(key)
    }

    pub fn is_projected(&self, field: &str) -> bool {
        self.projected.contains(field)
    }

    #[allow(dead_code)]
    pub fn sorted_index(&self, field: &str) -> Option<&CompositeSortedIndex> {
        self.held.get(self.by_field_sorted.get(field)?)?.sorted.as_ref()
    }

    pub fn lookup_exact(&self, field: &str, value: &Value) -> Option<&HashSet<String>> {
        self.held
            .get(self.by_field_exact.get(field)?)?
            .exact
            .as_ref()?
            .lookup(std::slice::from_ref(value))
    }

    /// The `where` clauses scoping `index`, or `None` for an index covering every record — what
    /// tells the planner whether an empty bucket of `index` proves absence or only says the record
    /// was never offered to it.
    pub fn where_of(&self, index: &RegistryIndex) -> Option<&[Predicate]> {
        self.where_filters.get(&index.name).map(Vec::as_slice)
    }

    pub fn candidates_of(&self, index: &RegistryIndex, values: &[Value]) -> Option<&HashSet<String>> {
        self.held.get(&index.name)?.exact.as_ref()?.lookup(values)
    }

    pub fn range_of(&self, index: &RegistryIndex, bounds: &RangeBounds) -> Option<Vec<String>> {
        Some(self.held.get(&index.name)?.sorted.as_ref()?.range(bounds))
    }

    pub fn tokens_of(&self, record: &HaysonRecord) -> Vec<String> {
        let mut tokens = Vec::new();
        for index in &self.fulltext_indexes {
            if let Some(field) = index.fields.first() {
                if let Some(value) = record.get(field) {
                    tokens.extend(tokenize_value(value));
                }
            }
        }
        tokens
    }

    // TODO: staged records join the result set but not the ranking -- BM25 rescoring is still open.
    pub fn search(&self, text: &str, mode: SearchMode) -> HashMap<String, f64> {
        self.fulltext.search(text, mode, None)
    }

    /// Whether `record` is one a `search(text, mode)` would answer with, judged on its own tokens
    /// rather than on the postings — the question a record this registry has not indexed yet needs
    /// asked, so a unit of work can hold its staged writes to the same search a committed record met.
    #[allow(dead_code)]
    pub fn matches_search(&self, record: &HaysonRecord, text: &str, mode: SearchMode) -> bool {
        matches_terms(&self.tokens_of(record), text, mode)
    }

    #[allow(dead_code)]
    pub fn unique_holder(&self, field: &str, value: &Value) -> Option<String> {
        let name = self.by_field_exact.get(field)?;
        let mut probe = HaysonRecord::new();
        probe.insert(field.to_string(), value.clone());
        self.held.get(name)?.exact.as_ref()?.holder_of(&probe)
    }

    pub fn unique_holder_of(&self, index: &RegistryIndex, record: &HaysonRecord) -> Option<String> {
        self.held.get(&index.name)?.exact.as_ref()?.holder_of(record)
    }

    /// Fields of this table carrying their own virtual-path index — an `image` field, or a
    /// `markdown` one that is addressable.
    pub fn virtual_path_fields(&self) -> impl Iterator<Item = &String> {
        self.virtual_paths.keys()
    }

    /// The key currently holding `virtual_path` in `field`'s virtual-path index, if any — the
    /// index is unique by construction, so at most one key can ever hold a given path.
    pub fn virtual_path_holder(&self, field: &str, virtual_path: &str) -> Option<String> {
        self.virtual_paths
            .get(field)?
            .holder_of(virtual_path)
            .map(str::to_string)
    }

    #[allow(dead_code)]
    pub fn apply_put(&mut self, key: &str, record: &HaysonRecord, slot: Slot, seq: u64) {
        self.apply_put_in(key, record, slot, seq, "");
    }

    pub fn apply_put_in(&mut self, key: &str, record: &HaysonRecord, slot: Slot, seq: u64, part: &str) {
        self.retire(key);
        self.primary.insert(
            key.to_string(),
            PrimaryEntry {
                offset: slot.offset,
                length: slot.length,
                seq,
            },
        );
        self.note_arrival(key);
        if part.is_empty() {
            self.parts_of.remove(key);
        } else {
            self.parts_of.insert(key.to_string(), part.to_string());
        }
        self.tally(part).0 += slot.length + 1;
        self.index_record(key, record);
        self.max_seq = self.max_seq.max(seq);
    }

    #[allow(dead_code)]
    pub fn apply_delete(&mut self, key: &str, tombstone_length: u64, seq: u64) {
        self.apply_delete_in(key, tombstone_length, seq, "");
    }

    pub fn apply_delete_in(&mut self, key: &str, tombstone_length: u64, seq: u64, part: &str) {
        if self.retire(key) {
            self.primary.remove(key);
            self.arrival.remove(key);
            self.parts_of.remove(key);
        }
        self.tally(part).1 += tombstone_length + 1;
        self.max_seq = self.max_seq.max(seq);
    }

    pub fn set_dead_bytes(&mut self, bytes: u64, part: &str) {
        self.tally(part).1 = bytes;
    }

    pub fn hydrate(&mut self, key: &str, projection: &HaysonRecord, entry: PrimaryEntry, part: &str) {
        self.primary.insert(key.to_string(), entry);
        self.note_arrival(key);
        if part.is_empty() {
            self.parts_of.remove(key);
        } else {
            self.parts_of.insert(key.to_string(), part.to_string());
        }
        self.tally(part).0 += entry.length + 1;
        self.index_record(key, projection);
        self.max_seq = self.max_seq.max(entry.seq);
    }

    /// Ranks `key` behind every key seen so far, but only the first time it is seen — an overwrite
    /// of a key already ranked keeps the rank it has, the way `Map.set` keeps a key's position.
    fn note_arrival(&mut self, key: &str) {
        if self.arrival.contains_key(key) {
            return;
        }
        self.arrival.insert(key.to_string(), self.next_arrival);
        self.next_arrival += 1;
    }

    fn retire(&mut self, key: &str) -> bool {
        let Some(previous) = self.primary.get(key).copied() else {
            return false;
        };
        let part = self.parts_of.get(key).cloned().unwrap_or_default();
        // NOTE: the dead bytes belong to the partition the old line sits in, not to the new one.
        let tally = self.tally(&part);
        tally.1 += previous.length + 1;
        tally.0 -= previous.length + 1;
        self.unindex(key);
        true
    }

    pub fn note_seq(&mut self, seq: u64) {
        self.max_seq = self.max_seq.max(seq);
    }

    pub fn remap(&mut self, slots: &HashMap<String, PrimaryEntry>, part: &str) {
        let tally = self.tally(part);
        tally.0 = 0;
        tally.1 = 0;
        // NOTE: sorted so a key compaction reintroduces gets a deterministic rank; ranked keys keep theirs.
        let mut keys: Vec<&String> = slots.keys().collect();
        keys.sort_unstable();
        for key in keys {
            let entry = slots[key];
            self.primary.insert(key.clone(), entry);
            self.note_arrival(key);
            self.tally(part).0 += entry.length + 1;
        }
    }

    fn index_record(&mut self, key: &str, record: &HaysonRecord) {
        let mut projection = HaysonRecord::new();
        for name in &self.projected {
            if let Some(value) = record.get(name) {
                projection.insert(name.clone(), value.clone());
            }
        }
        self.projections.insert(key.to_string(), projection);

        let where_filters = &self.where_filters;
        for (name, held) in self.held.iter_mut() {
            if where_filters
                .get(name)
                .is_some_and(|predicates| !record_matches_where(record, predicates))
            {
                continue;
            }
            if let Some(exact) = &mut held.exact {
                exact.add(record, key);
            }
            if let Some(sorted) = &mut held.sorted {
                sorted.insert(record, key);
            }
        }
        for (field, index) in &mut self.virtual_paths {
            if let Some(virtual_path) = virtual_path_of(record, field) {
                index.add(virtual_path, key.to_string());
            }
        }
        if self.has_fulltext() {
            let tokens = self.tokens_of(record);
            self.fulltext.index(key, &tokens);
        }
    }

    fn unindex(&mut self, key: &str) {
        if let Some(projection) = self.projections.remove(key) {
            let where_filters = &self.where_filters;
            for (name, held) in self.held.iter_mut() {
                if where_filters
                    .get(name)
                    .is_some_and(|predicates| !record_matches_where(&projection, predicates))
                {
                    continue;
                }
                if let Some(exact) = &mut held.exact {
                    exact.remove(&projection, key);
                }
                if let Some(sorted) = &mut held.sorted {
                    sorted.remove(&projection, key);
                }
            }
            for (field, index) in &mut self.virtual_paths {
                if let Some(virtual_path) = virtual_path_of(&projection, field) {
                    index.remove(&virtual_path, key);
                }
            }
        }
        self.fulltext.remove(key);
    }
}

#[cfg(test)]
mod tests {
    use serde_json::json;

    use super::*;
    use crate::schema::parse_schema;

    fn users_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string", "fulltext": true },
                "bio": { "type": "string", "fulltext": true },
                "age": { "type": "int", "index": "sorted" },
                "role": { "type": "string", "index": "exact" },
                "tags": { "type": "string[]", "index": "exact" },
                "email": { "type": "string", "index": "exact", "unique": true },
            },
        }))
        .unwrap()
    }

    fn record(pairs: &[(&str, Value)]) -> HaysonRecord {
        pairs.iter().map(|(k, v)| (k.to_string(), v.clone())).collect()
    }

    fn registry_with(records: &[HaysonRecord]) -> TableRegistry {
        let mut registry = TableRegistry::new(users_schema());
        let mut offset = 0u64;
        let mut seq = 0u64;
        for entry in records {
            let length = serde_json::to_string(entry).unwrap().len() as u64;
            seq += 1;
            let id = entry.get("id").and_then(Value::as_str).unwrap().to_string();
            registry.apply_put(&id, entry, Slot { offset, length }, seq);
            offset += length + 1;
        }
        registry
    }

    fn ania() -> HaysonRecord {
        record(&[
            ("id", json!("user_1")),
            ("name", json!("Ania Kowalska")),
            ("bio", json!("Backend developer")),
            ("age", json!(31)),
            ("role", json!("admin")),
            ("tags", json!(["backend", "node"])),
        ])
    }

    fn piotr() -> HaysonRecord {
        record(&[
            ("id", json!("user_2")),
            ("name", json!("Piotr Nowak")),
            ("bio", json!("Frontend developer")),
            ("age", json!(24)),
            ("role", json!("user")),
            ("tags", json!(["frontend"])),
        ])
    }

    fn maria() -> HaysonRecord {
        record(&[
            ("id", json!("user_3")),
            ("name", json!("Maria Wisniewska")),
            ("bio", json!("Go developer")),
            ("age", json!(41)),
            ("role", json!("user")),
            ("tags", json!(["backend", "go"])),
        ])
    }

    #[test]
    fn serves_exact_array_member_sorted_and_fulltext_lookups() {
        let registry = registry_with(&[ania(), piotr()]);

        let mut role_admin: Vec<&String> = registry.lookup_exact("role", &json!("admin")).unwrap().iter().collect();
        role_admin.sort();
        assert_eq!(role_admin, vec!["user_1"]);

        let mut tags_frontend: Vec<&String> = registry
            .lookup_exact("tags", &json!("frontend"))
            .unwrap()
            .iter()
            .collect();
        tags_frontend.sort();
        assert_eq!(tags_frontend, vec!["user_2"]);

        assert_eq!(
            registry.sorted_index("age").unwrap().range(&RangeBounds {
                gte: Some(vec![super::super::sorted_index::SortValue::Num(30.0)]),
                ..Default::default()
            }),
            vec!["user_1"]
        );

        let hits: Vec<String> = registry.search("kowalska", SearchMode::All).into_keys().collect();
        assert_eq!(hits, vec!["user_1"]);
        assert_eq!(registry.records(), 2);
    }

    #[test]
    fn does_not_confuse_a_number_with_its_string_form() {
        let registry = registry_with(&[ania()]);
        assert!(registry.lookup_exact("age", &json!("31")).is_none());
    }

    #[test]
    fn counts_an_overwritten_record_as_garbage_and_reindexes_it() {
        let mut registry = registry_with(&[ania()]);
        let before = registry.stats();
        let mut updated = ania();
        updated.insert("role".into(), json!("user"));
        updated.insert("age".into(), json!(32));
        let payload = serde_json::to_string(&updated).unwrap().len() as u64;
        registry.apply_put(
            "user_1",
            &updated,
            Slot {
                offset: 500,
                length: payload,
            },
            5,
        );

        assert_eq!(registry.stats().dead_bytes, before.live_bytes);
        assert!(registry.lookup_exact("role", &json!("admin")).is_none());
        let mut users: Vec<&String> = registry.lookup_exact("role", &json!("user")).unwrap().iter().collect();
        users.sort();
        assert_eq!(users, vec!["user_1"]);
    }

    #[test]
    fn drops_a_deleted_record_from_every_registry_and_counts_the_tombstone() {
        let mut registry = registry_with(&[ania(), piotr()]);
        let live_before = registry.stats().live_bytes;
        registry.apply_delete("user_1", 40, 9);

        assert_eq!(registry.records(), 1);
        assert!(registry.slot_of("user_1").is_none());
        assert!(registry.lookup_exact("role", &json!("admin")).is_none());
        assert!(registry.search("kowalska", SearchMode::All).is_empty());
        assert!(registry.stats().live_bytes < live_before);
        assert!(registry.stats().dead_bytes > 40);
    }

    fn arrival_order(registry: &TableRegistry) -> Vec<&str> {
        registry
            .keys_in_arrival_order()
            .into_iter()
            .map(String::as_str)
            .collect()
    }

    /// NOTE-8bmm: `Map.set` on a key already in the map keeps that key's position, so an overwrite
    /// in the TS engine never moves a record. This registry has to answer the same way.
    #[test]
    fn an_overwrite_leaves_a_key_where_it_first_arrived() {
        let mut registry = registry_with(&[ania(), piotr(), maria()]);
        let mut updated = ania();
        updated.insert("age".into(), json!(32));
        registry.apply_put(
            "user_1",
            &updated,
            Slot {
                offset: 900,
                length: 40,
            },
            9,
        );

        assert_eq!(arrival_order(&registry), vec!["user_1", "user_2", "user_3"]);
        assert_eq!(registry.arrival_of("user_1"), Some(0));
    }

    #[test]
    fn a_key_first_seen_after_an_overwrite_arrives_last() {
        let mut registry = registry_with(&[ania(), piotr()]);
        let mut updated = ania();
        updated.insert("age".into(), json!(32));
        registry.apply_put(
            "user_1",
            &updated,
            Slot {
                offset: 900,
                length: 40,
            },
            9,
        );
        registry.apply_put(
            "user_3",
            &maria(),
            Slot {
                offset: 940,
                length: 40,
            },
            10,
        );

        assert_eq!(arrival_order(&registry), vec!["user_1", "user_2", "user_3"]);
    }

    /// A delete drops the key from the TS `Map` outright, so writing it again is a first arrival --
    /// it goes to the back, unlike the overwrite of a key that never left.
    #[test]
    fn a_key_written_again_after_a_delete_arrives_afresh() {
        let mut registry = registry_with(&[ania(), piotr(), maria()]);
        registry.apply_delete("user_1", 40, 9);
        registry.apply_put(
            "user_1",
            &ania(),
            Slot {
                offset: 900,
                length: 40,
            },
            10,
        );

        assert_eq!(arrival_order(&registry), vec!["user_2", "user_3", "user_1"]);
    }

    #[test]
    fn compaction_leaves_every_surviving_key_where_it_arrived() {
        let mut registry = registry_with(&[ania(), piotr(), maria()]);
        let mut slots = HashMap::new();
        for (index, key) in ["user_3", "user_1", "user_2"].iter().enumerate() {
            slots.insert(
                (*key).to_string(),
                PrimaryEntry {
                    offset: index as u64 * 100,
                    length: 99,
                    seq: 1,
                },
            );
        }
        registry.remap(&slots, "");

        assert_eq!(arrival_order(&registry), vec!["user_1", "user_2", "user_3"]);
    }

    #[test]
    fn answers_the_unique_holder_of_a_value() {
        let mut with_email = ania();
        with_email.insert("email".into(), json!("ania@example.com"));
        let registry = registry_with(&[with_email]);

        assert_eq!(
            registry.unique_holder("email", &json!("ania@example.com")),
            Some("user_1".to_string())
        );
        assert_eq!(registry.unique_holder("email", &json!("other@example.com")), None);
    }

    /// Cross-check with `packages/hayson/test/fixtures/v1/fulltext-field` -- proves the registry
    /// built from a real scanned log matches the TS engine's `search()` result set.
    #[test]
    fn matches_the_ts_fixture_search_results_after_a_log_scan() {
        let mut root = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        root.push("../../packages/hayson/test/fixtures/v1/fulltext-field");
        // NOTE: a v1 fixture is a frozen format-1 file — its marker still sits under "hayson".
        let schema = parse_schema(&crate::schema::parse::with_current_schema_marker(
            serde_json::from_str(&std::fs::read_to_string(root.join("users.schema.json")).unwrap()).unwrap(),
        ))
        .unwrap();
        let log = LogFile::open(root.join("users.jsonl"), true).unwrap();
        let registry = TableRegistry::build_from_log(schema, &log).unwrap();

        let mut psy: Vec<String> = registry.search("psy", SearchMode::All).into_keys().collect();
        psy.sort();
        assert_eq!(psy, vec!["user_1".to_string()]);

        let mut koty: Vec<String> = registry.search("koty", SearchMode::All).into_keys().collect();
        koty.sort();
        assert_eq!(koty, vec!["user_1".to_string(), "user_2".to_string()]);
    }

    #[test]
    fn remaps_slots_after_compaction_and_forgets_the_garbage() {
        let mut registry = registry_with(&[ania(), piotr()]);
        registry.apply_delete("user_2", 40, 9);
        let mut slots = HashMap::new();
        slots.insert(
            "user_1".to_string(),
            PrimaryEntry {
                offset: 0,
                length: 99,
                seq: 1,
            },
        );
        registry.remap(&slots, "");

        let stats = registry.stats();
        assert_eq!(stats.records, 1);
        assert_eq!(stats.live_bytes, 100);
        assert_eq!(stats.dead_bytes, 0);
        assert_eq!(
            registry.slot_of("user_1"),
            Some(PrimaryEntry {
                offset: 0,
                length: 99,
                seq: 1
            })
        );
    }

    fn docs_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
            },
        }))
        .unwrap()
    }

    #[test]
    fn holds_the_virtual_path_of_a_markdown_field_uniquely() {
        let schema = docs_schema();
        let mut registry = TableRegistry::new(schema);
        let doc1 = record(&[
            ("id", json!("d1")),
            ("page", json!({ "content": "a", "virtualPath": "/docs/a.md" })),
        ]);
        registry.apply_put("d1", &doc1, Slot { offset: 0, length: 10 }, 1);

        assert_eq!(
            registry.virtual_path_holder("page", "/docs/a.md"),
            Some("d1".to_string())
        );
        assert_eq!(registry.virtual_path_holder("page", "/docs/b.md"), None);
    }

    #[test]
    fn keeps_a_non_addressable_markdown_field_out_of_the_virtual_path_index() {
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
                "notes": { "type": "markdown", "addressable": false },
            },
        }))
        .unwrap();
        let mut registry = TableRegistry::new(schema);
        let doc = record(&[
            ("id", json!("d1")),
            ("page", json!({ "content": "a", "virtualPath": "/docs/a.md" })),
            ("notes", json!({ "content": "b", "virtualPath": "/notes/a.md" })),
        ]);
        registry.apply_put("d1", &doc, Slot { offset: 0, length: 10 }, 1);

        let fields: Vec<&String> = registry.virtual_path_fields().collect();
        assert_eq!(fields, vec!["page"]);
        assert_eq!(registry.virtual_path_holder("notes", "/notes/a.md"), None);
    }

    #[test]
    fn frees_a_virtual_path_once_its_holder_is_deleted_or_overwritten() {
        let schema = docs_schema();
        let mut registry = TableRegistry::new(schema);
        let doc1 = record(&[
            ("id", json!("d1")),
            ("page", json!({ "content": "a", "virtualPath": "/docs/a.md" })),
        ]);
        registry.apply_put("d1", &doc1, Slot { offset: 0, length: 10 }, 1);
        registry.apply_delete("d1", 5, 2);
        assert_eq!(registry.virtual_path_holder("page", "/docs/a.md"), None);
    }
}