hayson-engine 0.8.0

use std::collections::HashMap;
use std::fmt;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

use serde_json::Value;

use crate::catalog::TriggerEvent;
use crate::query::planner::{QueryResult, QuerySpec};
use crate::table::{effect_document, matching, PreparedTxnEntry, Table, TableError, TriggerFiring, TxnOp};

use super::journal::{self, Journal, JournalEntry, JournalHeader, JOURNAL_VERSION};
use super::overlay::{StagedIndex, StagedOverlay};

static UNIT_COUNTER: AtomicU64 = AtomicU64::new(0);

/// The process half of a unit id. Same platform split, and the same trap, as `lock::platform::pid`
/// — kept local because `txn` does not depend on `lock`.
///
/// NOTE: std::process::id() PANICS on wasm32-wasip1 ("no pids on this platform") — never call it.
#[cfg(unix)]
fn process_tag() -> u32 {
    std::process::id()
}

#[cfg(not(unix))]
fn process_tag() -> u32 {
    0
}

fn generate_id() -> String {
    let counter = UNIT_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!(
        "{}-{}-{counter}",
        process_tag(),
        chrono::Utc::now().timestamp_nanos_opt().unwrap_or_default()
    )
}

#[derive(Debug)]
pub enum TxnError {
    UnknownTable(String),
    AlreadyCommitted,
    JournalPending(PathBuf),
    Table(String, TableError),
    Io(io::Error),
}

impl fmt::Display for TxnError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            TxnError::UnknownTable(name) => write!(f, "unit of work names unknown table \"{name}\""),
            TxnError::AlreadyCommitted => write!(f, "this unit of work was already committed — start a new one"),
            TxnError::JournalPending(path) => write!(
                f,
                "an unfinished unit of work is still journalled in \"{}\" — reopen the database to let it finish",
                path.display()
            ),
            TxnError::Table(name, error) => write!(f, "table \"{name}\": {error}"),
            TxnError::Io(error) => error.fmt(f),
        }
    }
}

impl std::error::Error for TxnError {}

impl From<io::Error> for TxnError {
    fn from(value: io::Error) -> Self {
        TxnError::Io(value)
    }
}

#[derive(Debug, Clone, PartialEq)]
pub struct UnitOfWorkResult {
    pub id: String,
    pub tables: Vec<String>,
    pub operations: usize,
}

/// A batch of writes across several tables that lands whole or not at all (ADR 0012). Operations
/// are collected with `put`/`patch`/`delete` and nothing touches a table until `commit()`, which
/// runs the six steps documented in `docs/architecture.md`: resolve, barrier, prepare, journal,
/// write, cleanup. `&mut Table` borrows held for the call are the barrier — no separate write
/// queue to drain, since nothing else can hold those tables mutably at the same time.
///
/// `commit()` borrows the unit rather than consuming it, so a refused commit can be retried on the
/// same unit once its cause is fixed — the same contract as the TS `UnitOfWork`. Staging is
/// append-only in both engines: fixing a refusal means fixing what it named, not unstaging an op.
#[derive(Default)]
pub struct UnitOfWork {
    order: Vec<String>,
    staged: HashMap<String, Vec<TxnOp>>,
    operations: usize,
    committed: bool,
}

/// Everything [`UnitOfWork::prepare`] worked out, carried from there to [`UnitOfWork::journal`]
/// and on to `apply_prepared`. Owns its lines, so it can be held across the release of whatever
/// lock guarded the tables it was prepared against — but only if that release keeps the caller the
/// sole writer, since the `_seq` numbers inside it are already burned and a write that slipped in
/// between would land ahead of them in the log.
pub struct PreparedUnit {
    id: String,
    touched: Vec<String>,
    entries: Vec<(String, Vec<PreparedTxnEntry>)>,
    operations: usize,
}

impl PreparedUnit {
    /// True when the unit had nothing staged: no journal is written and no table is touched.
    pub fn is_empty(&self) -> bool {
        self.operations == 0
    }

    /// The tables this unit will write to, its own in staging order first, then any a trigger
    /// effect added.
    pub fn tables(&self) -> &[String] {
        &self.touched
    }
}

/// What a key would read as once this unit lands, computed against `table`'s committed state plus
/// the ops staged so far for it (ADR-3h3r).
pub enum StagedView {
    Untouched,
    Deleted,
    Put(crate::schema::HaysonRecord),
}

impl UnitOfWork {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn operations(&self) -> usize {
        self.operations
    }

    pub fn tables(&self) -> &[String] {
        &self.order
    }

    pub fn put(&mut self, table: &str, document: Value) -> &mut Self {
        self.add(table, TxnOp::Put(document))
    }

    pub fn patch(&mut self, table: &str, key: impl Into<String>, changes: Value) -> &mut Self {
        self.add(table, TxnOp::Patch(key.into(), changes))
    }

    pub fn delete(&mut self, table: &str, key: impl Into<String>) -> &mut Self {
        self.add(table, TxnOp::Delete(key.into()))
    }

    fn add(&mut self, table: &str, op: TxnOp) -> &mut Self {
        if !self.staged.contains_key(table) {
            self.order.push(table.to_string());
        }
        self.staged.entry(table.to_string()).or_default().push(op);
        self.operations += 1;
        self
    }

    /// Reads a key through the unit: a value staged for `table` since the unit started, falling
    /// back to `table`'s committed state for a key the unit does not touch. `put` and `patch` ops
    /// are applied in staging order, `patch` merging over whatever the unit has staged so far (or
    /// the committed record, on the first touch) exactly as `Table::patch` would.
    pub fn preview(&self, table_name: &str, key: &str, table: &Table) -> Result<StagedView, TableError> {
        match self.staged.get(table_name) {
            Some(ops) => StagedOverlay::new(ops).resolve(key, table),
            None => Ok(StagedView::Untouched),
        }
    }

    pub fn get(
        &self,
        table_name: &str,
        key: &str,
        table: &Table,
    ) -> Result<Option<crate::schema::HaysonRecord>, TableError> {
        match self.preview(table_name, key, table)? {
            StagedView::Deleted => Ok(None),
            StagedView::Put(record) => Ok(Some(record)),
            StagedView::Untouched => Ok(table.get(key)?),
        }
    }

    /// Runs `spec` as the unit will leave `table`: the ops staged for `table_name` so far are
    /// resolved into a `StagedIndex`, the query is planned over the committed candidates the
    /// overlay has not taken over plus the staged records that can still match, and the answer is
    /// tested against staged records and committed ones alike. A `search` is scored over the
    /// overlaid corpus, so a staged record joins the ranking and not merely the result set.
    ///
    /// The plan carries a `PlanOverlay` reporting what the staged half contributed — how many
    /// staged records joined the candidates, how many committed ones the overlay suppressed, and
    /// whether the scores behind them were recomputed.
    pub fn query(&self, table_name: &str, table: &Table, spec: &QuerySpec) -> Result<QueryResult, TableError> {
        // NOTE: a unit with nothing staged still plans through an empty overlay, so `plan.overlay` is
        // there for every query a unit answers — a caller reads it to tell a unit query from a plain one.
        let overlay = StagedIndex::build(self.staged.get(table_name).map_or(&[][..], Vec::as_slice), table)?;
        table.query_with_overlay(spec, &overlay, &overlay.corpus_delta())
    }

    /// Runs every staged operation, or none of them, across `tables`.
    ///
    /// 1. resolve — an op naming a table missing from `tables` fails before anything else runs;
    /// 2. barrier — the `&mut Table` borrows this call holds for its whole body;
    /// 3. prepare — [`Table::prepare_txn_ops`] on every table, in staging order; the first
    ///    rejected op fails the call with nothing appended anywhere;
    /// 4. journal — every prepared line, across every table, written to `journal_path` in one
    ///    `atomic::write_atomic` call;
    /// 5. write — [`Table::apply_txn_entries`] per table, in staging order;
    /// 6. cleanup — the journal file removed.
    ///
    /// A refusal raised by steps 1–3 leaves the unit live and fully staged, so the caller can fix
    /// whatever the refusal named and call `commit` again on the same unit — nothing was written,
    /// and only monotonic `_seq` counters were burned. From step 4 on the unit is spent whatever
    /// happens, because a journal that reached the disk is finished by `recover_incomplete`, and a
    /// second run of the same operations would apply them twice.
    ///
    /// Steps 1–3, 4 and 5–6 are also callable one at a time as [`UnitOfWork::prepare`],
    /// [`UnitOfWork::journal`] and `apply_prepared`; this is those three back to back. A caller
    /// that holds the tables behind a lock uses the split so it can let go of that lock across
    /// step 4, which touches no table — see the concurrency section of `docs/architecture.md`.
    pub fn commit(
        &mut self,
        tables: &mut HashMap<String, Table>,
        journal_path: &Path,
    ) -> Result<UnitOfWorkResult, TxnError> {
        let prepared = self.prepare(tables)?;
        self.journal(&prepared, journal_path)?;
        apply_prepared(prepared, tables, journal_path)
    }

    /// Steps 1–3 of [`UnitOfWork::commit`]: resolve, prepare, and the triggers the prepared
    /// entries fire. Mutates only the `_seq` counters of the tables it touches; nothing is
    /// appended, no registry moves and no file is written, so a refusal here still leaves the unit
    /// staged and retriable.
    pub fn prepare(&mut self, tables: &mut HashMap<String, Table>) -> Result<PreparedUnit, TxnError> {
        if self.committed {
            return Err(TxnError::AlreadyCommitted);
        }
        let id = generate_id();
        if self.operations == 0 {
            self.committed = true;
            return Ok(PreparedUnit {
                id,
                touched: Vec::new(),
                entries: Vec::new(),
                operations: 0,
            });
        }

        for name in &self.order {
            if !tables.contains_key(name) {
                return Err(TxnError::UnknownTable(name.clone()));
            }
        }

        let mut entries: Vec<(String, Vec<PreparedTxnEntry>)> = Vec::with_capacity(self.order.len());
        for name in &self.order {
            let ops = self.staged.get(name).expect("resolved above");
            let table = tables.get_mut(name).expect("resolved above");
            let prepared = table
                .prepare_txn_ops(ops)
                .map_err(|error| TxnError::Table(name.clone(), error))?;
            entries.push((name.clone(), prepared));
        }

        let mut touched = self.order.clone();
        for (effect_table, effect_entries) in self.trigger_effect_entries(&entries, tables)? {
            if !touched.contains(&effect_table) {
                touched.push(effect_table.clone());
            }
            entries.push((effect_table, effect_entries));
        }

        Ok(PreparedUnit {
            id,
            touched,
            entries,
            operations: self.operations,
        })
    }

    /// Step 4 of [`UnitOfWork::commit`]: every prepared line, across every table, written to
    /// `journal_path` and fsynced in one `atomic::write_atomic` call. Reads no table and mutates
    /// none, which is what lets a caller run it without the lock that guards them — the writes are
    /// still invisible to a reader at this point, because nothing has been applied yet.
    ///
    /// Retires the unit: from the moment the journal is on disk the operations land whether this
    /// process finishes them or the next `open()` does, so a second run would apply them twice.
    pub fn journal(&mut self, prepared: &PreparedUnit, journal_path: &Path) -> Result<(), TxnError> {
        if prepared.is_empty() {
            return Ok(());
        }
        if journal_path.exists() {
            return Err(TxnError::JournalPending(journal_path.to_path_buf()));
        }

        let mut journal_entries = Vec::new();
        for (name, entries) in &prepared.entries {
            for entry in entries {
                journal_entries.push(JournalEntry {
                    t: name.clone(),
                    s: entry.seq(),
                    l: entry.line().to_string(),
                });
            }
        }

        // NOTE: spent from here on — a journal that lands is finished by recover_incomplete.
        self.committed = true;
        journal::write_journal(
            journal_path,
            &Journal {
                header: JournalHeader {
                    journal_version: JOURNAL_VERSION,
                    kind: "unit".to_string(),
                    id: prepared.id.clone(),
                    started_at: chrono::Utc::now().timestamp_millis(),
                    tables: prepared.touched.clone(),
                },
                entries: journal_entries,
            },
        )?;
        Ok(())
    }

    /// Computes every trigger effect the ops just prepared for `prepared` fire, `prepare_txn_ops`-
    /// validated against its own effect table but not yet appended anywhere — folded into the same
    /// `prepared`/journal/apply pass by the caller, so the effect lands in the very same unit of
    /// work as the write that fired it. Reads triggers off `prepared`'s ops only, never off an
    /// effect entry this same call produced, so v1's declared "no chaining" holds by construction.
    fn trigger_effect_entries(
        &self,
        prepared: &[(String, Vec<PreparedTxnEntry>)],
        tables: &mut HashMap<String, Table>,
    ) -> Result<Vec<(String, Vec<PreparedTxnEntry>)>, TxnError> {
        let mut documents_by_effect_table: HashMap<String, Vec<Value>> = HashMap::new();
        for (name, entries) in prepared {
            let triggers = tables.get(name).expect("resolved above").triggers();
            for entry in entries {
                let event = match entry.record() {
                    Some(_) => entry.upsert_event(),
                    None if entry.old().is_some() => TriggerEvent::Delete,
                    None => continue,
                };
                let firing = TriggerFiring {
                    event,
                    old: entry.old(),
                    new: entry.record(),
                };
                for trigger in matching(triggers, name, event) {
                    let document = effect_document(&trigger.effect, &firing);
                    documents_by_effect_table
                        .entry(trigger.effect.table.clone())
                        .or_default()
                        .push(document);
                }
            }
        }

        let mut effect_prepared = Vec::with_capacity(documents_by_effect_table.len());
        for (effect_table, documents) in documents_by_effect_table {
            let table = tables
                .get_mut(&effect_table)
                .ok_or_else(|| TxnError::UnknownTable(effect_table.clone()))?;
            let ops: Vec<TxnOp> = documents.into_iter().map(TxnOp::Put).collect();
            let entries = table
                .prepare_txn_ops(&ops)
                .map_err(|error| TxnError::Table(effect_table.clone(), error))?;
            effect_prepared.push((effect_table, entries));
        }
        Ok(effect_prepared)
    }
}

/// Steps 5–6 of [`UnitOfWork::commit`]: every prepared entry appended, table by table in staging
/// order, and the journal removed once they all are. `tables` must be the very map `prepared` was
/// prepared against — the entries carry the `_seq` numbers taken from it.
///
/// This is the step a reader must not see half of, so a caller that split the commit up holds its
/// tables exclusively for the whole of it.
pub fn apply_prepared(
    prepared: PreparedUnit,
    tables: &mut HashMap<String, Table>,
    journal_path: &Path,
) -> Result<UnitOfWorkResult, TxnError> {
    let PreparedUnit {
        id,
        touched,
        entries,
        operations,
    } = prepared;
    if operations == 0 {
        return Ok(UnitOfWorkResult {
            id,
            tables: Vec::new(),
            operations: 0,
        });
    }

    for (name, entries) in entries {
        let table = tables
            .get_mut(&name)
            .ok_or_else(|| TxnError::UnknownTable(name.clone()))?;
        table.apply_txn_entries(entries).map_err(TxnError::Io)?;
    }

    journal::remove_journal(journal_path);
    Ok(UnitOfWorkResult {
        id,
        tables: touched,
        operations,
    })
}

/// Finishes an unfinished journal found at `journal_path`, if any — the recovery half of
/// `commit()`'s step 4/5 split. `tables` must hold every table the journal names; a name it
/// cannot resolve aborts recovery without touching the journal, so the next `open()` retries it
/// rather than losing the record of what still needs finishing.
pub fn recover_incomplete(
    journal_path: &Path,
    tables: &mut HashMap<String, Table>,
) -> Result<Option<UnitOfWorkResult>, TxnError> {
    let Some(journal) = journal::read_journal(journal_path) else {
        return Ok(None);
    };

    let mut by_table: HashMap<&str, Vec<(u64, String)>> = HashMap::new();
    for entry in &journal.entries {
        by_table
            .entry(entry.t.as_str())
            .or_default()
            .push((entry.s, entry.l.clone()));
    }

    for name in &journal.header.tables {
        if !tables.contains_key(name) {
            return Err(TxnError::UnknownTable(name.clone()));
        }
    }

    let mut applied_ops = 0;
    for name in &journal.header.tables {
        let table = tables.get_mut(name).expect("checked above");
        if let Some(entries) = by_table.get(name.as_str()) {
            applied_ops += table
                .recover_txn_lines(entries)
                .map_err(|error| TxnError::Table(name.clone(), error))?;
        }
    }

    journal::remove_journal(journal_path);
    Ok(Some(UnitOfWorkResult {
        id: journal.header.id,
        tables: journal.header.tables,
        operations: applied_ops,
    }))
}

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

    use super::*;
    use crate::query::builder::QueryBuilder;
    use crate::query::planner::{OverlayStrategy, PlanOverlay};
    use crate::query::predicates::ComparisonOp;
    use crate::registry::fulltext_index::SearchMode;
    use crate::schema::{parse_schema, TableSchema};

    fn users_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string", "required": true },
            },
        }))
        .unwrap()
    }

    fn orders_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "orders",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "userId": { "type": "string", "required": true },
                "total": { "type": "int", "required": true },
            },
        }))
        .unwrap()
    }

    struct Fixture {
        _dir: tempfile::TempDir,
        users_path: PathBuf,
        orders_path: PathBuf,
        journal_path: PathBuf,
    }

    fn fixture() -> Fixture {
        let dir = tempdir().unwrap();
        let users_path = dir.path().join("users.jsonl");
        let orders_path = dir.path().join("orders.jsonl");
        let journal_path = dir.path().join(super::journal::JOURNAL_FILE);
        Fixture {
            _dir: dir,
            users_path,
            orders_path,
            journal_path,
        }
    }

    fn open_tables(fx: &Fixture) -> HashMap<String, Table> {
        let mut tables = HashMap::new();
        tables.insert(
            "users".to_string(),
            Table::open(users_schema(), &fx.users_path).unwrap(),
        );
        tables.insert(
            "orders".to_string(),
            Table::open(orders_schema(), &fx.orders_path).unwrap(),
        );
        tables
    }

    #[test]
    fn every_unit_id_is_its_own_even_when_two_are_minted_in_the_same_nanosecond() {
        let first = generate_id();
        let second = generate_id();

        assert_ne!(first, second, "the counter is what keeps two ids apart, not the clock");
        assert_eq!(first.split('-').count(), 3, "process tag, timestamp, counter");
    }

    #[test]
    fn commits_across_two_tables_and_both_see_the_change() {
        let fx = fixture();
        let mut tables = open_tables(&fx);

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        unit.put("orders", json!({ "id": "o1", "userId": "u1", "total": 10 }));

        let result = unit.commit(&mut tables, &fx.journal_path).unwrap();
        assert_eq!(result.operations, 2);
        assert_eq!(result.tables, vec!["users".to_string(), "orders".to_string()]);
        assert!(!fx.journal_path.exists());

        assert_eq!(
            tables.get("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        assert_eq!(
            tables.get("orders").unwrap().get("o1").unwrap().unwrap().get("total"),
            Some(&json!(10))
        );
    }

    #[test]
    fn a_unit_reading_and_committing_into_a_warn_table_never_sees_the_stray_field() {
        let fx = fixture();
        let mut tables = open_tables(&fx);
        tables
            .get_mut("users")
            .unwrap()
            .set_validation(crate::catalog::CatalogValidation::Warn);

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }));

        let previewed = unit.get("users", "u1", tables.get("users").unwrap()).unwrap().unwrap();
        assert_eq!(previewed.get("name"), Some(&json!("Ania")));
        assert_eq!(previewed.get("nickname"), None);

        unit.commit(&mut tables, &fx.journal_path).unwrap();

        assert_eq!(
            tables.get("users").unwrap().get("u1").unwrap().unwrap().get("nickname"),
            None
        );
        let written = std::fs::read_to_string(&fx.users_path).unwrap();
        assert!(!written.contains("nickname"), "{written}");
    }

    #[test]
    fn a_read_through_the_unit_sees_its_own_uncommitted_put() {
        let fx = fixture();
        let tables = open_tables(&fx);

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));

        let seen = unit.get("users", "u1", tables.get("users").unwrap()).unwrap();
        assert_eq!(seen.unwrap().get("name"), Some(&json!("Ania")));
        assert!(tables.get("users").unwrap().get("u1").unwrap().is_none());
    }

    /// The staged record sorts first, so it can only reach a page of two if the page was cut after
    /// the overlay corrected the baseline — cutting first would have paged the committed two alone.
    #[test]
    fn a_page_through_the_unit_is_cut_after_the_staged_writes_joined_the_order() {
        let fx = fixture();
        let mut tables = open_tables(&fx);
        let users = tables.get_mut("users").unwrap();
        users.put(&json!({ "id": "u1", "name": "Bea" }), false).unwrap();
        users.put(&json!({ "id": "u3", "name": "Cela" }), false).unwrap();
        let users = tables.get("users").unwrap();

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u2", "name": "Ala" }));

        let spec = QueryBuilder::new()
            .order_by("name", crate::query::planner::SortDirection::Asc)
            .limit(2)
            .spec();
        let through_unit = unit.query("users", users, &spec).unwrap();

        assert_eq!(through_unit.keys, vec!["u2".to_string(), "u1".to_string()]);
        assert_eq!(
            through_unit.matched, 3,
            "the page is two wide, all three records matched"
        );
    }

    fn name_eq(name: &str) -> QuerySpec {
        QueryBuilder::new()
            .r#where("name", ComparisonOp::Eq, json!(name))
            .spec()
    }

    #[test]
    fn a_query_through_the_unit_finds_an_uncommitted_put_the_table_itself_does_not_see_yet() {
        let fx = fixture();
        let tables = open_tables(&fx);
        let users = tables.get("users").unwrap();

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));

        let spec = name_eq("Ania");
        assert!(users.query(&spec).unwrap().keys.is_empty());

        let through_unit = unit.query("users", users, &spec).unwrap();
        assert_eq!(through_unit.keys, vec!["u1".to_string()]);
    }

    #[test]
    fn a_query_through_the_unit_excludes_an_uncommitted_delete_the_table_itself_still_finds() {
        let fx = fixture();
        let mut tables = open_tables(&fx);
        tables
            .get_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        let users = tables.get("users").unwrap();

        let mut unit = UnitOfWork::new();
        unit.delete("users", "u1");

        let spec = name_eq("Ania");
        assert_eq!(users.query(&spec).unwrap().keys, vec!["u1".to_string()]);

        let through_unit = unit.query("users", users, &spec).unwrap();
        assert!(through_unit.keys.is_empty());
    }

    #[test]
    fn a_query_through_the_unit_excludes_a_patch_that_makes_the_record_stop_matching() {
        let fx = fixture();
        let mut tables = open_tables(&fx);
        tables
            .get_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        let users = tables.get("users").unwrap();

        let mut unit = UnitOfWork::new();
        unit.patch("users", "u1", json!({ "name": "Zosia" }));

        let spec = name_eq("Ania");
        assert_eq!(users.query(&spec).unwrap().keys, vec!["u1".to_string()]);

        let through_unit = unit.query("users", users, &spec).unwrap();
        assert!(through_unit.keys.is_empty());
    }

    #[test]
    fn an_unknown_table_fails_the_whole_unit_before_anything_is_staged_to_disk() {
        let fx = fixture();
        let mut tables = open_tables(&fx);

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        unit.put("ghost", json!({ "id": "x" }));

        let error = unit.commit(&mut tables, &fx.journal_path).unwrap_err();
        assert!(matches!(error, TxnError::UnknownTable(name) if name == "ghost"));
        assert!(tables.get("users").unwrap().get("u1").unwrap().is_none());
        assert!(!fx.journal_path.exists());
    }

    #[test]
    fn an_invalid_op_in_one_table_rejects_the_whole_unit_with_nothing_written_anywhere() {
        let fx = fixture();
        let mut tables = open_tables(&fx);

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        // "total" is required by the orders schema — this op is invalid.
        unit.put("orders", json!({ "id": "o1", "userId": "u1" }));

        let error = unit.commit(&mut tables, &fx.journal_path).unwrap_err();
        assert!(matches!(error, TxnError::Table(name, _) if name == "orders"));
        assert!(tables.get("users").unwrap().get("u1").unwrap().is_none());
        assert!(tables.get("orders").unwrap().get("o1").unwrap().is_none());
        assert!(!fx.journal_path.exists());
    }

    #[test]
    fn a_refused_unit_keeps_its_staging_and_lands_once_the_cause_is_fixed() {
        let fx = fixture();
        let mut tables = open_tables(&fx);

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        // "o1" is not in orders yet, so this patch cannot be prepared.
        unit.patch("orders", "o1", json!({ "total": 20 }));

        let error = unit.commit(&mut tables, &fx.journal_path).unwrap_err();
        assert!(matches!(error, TxnError::Table(name, _) if name == "orders"));
        assert_eq!(unit.operations(), 2, "the refusal left the staging alone");
        assert_eq!(unit.tables(), ["users".to_string(), "orders".to_string()]);
        assert_eq!(
            unit.get("users", "u1", tables.get("users").unwrap())
                .unwrap()
                .unwrap()
                .get("name"),
            Some(&json!("Ania")),
            "and the unit still reads through its own staging"
        );

        tables
            .get_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o1", "userId": "u1", "total": 10 }), false)
            .unwrap();

        let result = unit.commit(&mut tables, &fx.journal_path).unwrap();
        assert_eq!(result.operations, 2);
        assert_eq!(
            tables.get("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        assert_eq!(
            tables.get("orders").unwrap().get("o1").unwrap().unwrap().get("total"),
            Some(&json!(20))
        );
        assert!(!fx.journal_path.exists());
    }

    fn people_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "people",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "email": { "type": "string", "required": true, "unique": true },
                "bio": { "type": "string", "fulltext": true },
            },
        }))
        .unwrap()
    }

    fn open_people(fx: &Fixture) -> HashMap<String, Table> {
        let mut tables = HashMap::new();
        let path = fx.users_path.with_file_name("people.jsonl");
        tables.insert("people".to_string(), Table::open(people_schema(), &path).unwrap());
        tables
    }

    fn is_unique_violation(error: &TxnError, holder: &str) -> bool {
        matches!(
            error,
            TxnError::Table(name, TableError::Unique(violation))
                if name == "people" && violation.field == "email" && violation.holder == holder
        )
    }

    #[test]
    fn two_puts_of_one_email_in_the_same_unit_are_refused_before_either_lands() {
        let fx = fixture();
        let mut tables = open_people(&fx);

        let mut unit = UnitOfWork::new();
        unit.put("people", json!({ "id": "p1", "email": "ala@example.com" }));
        unit.put("people", json!({ "id": "p2", "email": "ala@example.com" }));

        let error = unit.commit(&mut tables, &fx.journal_path).unwrap_err();
        assert!(
            is_unique_violation(&error, "p1"),
            "the first op of the unit is the holder: {error}"
        );
        assert!(
            tables.get("people").unwrap().get("p1").unwrap().is_none(),
            "nothing of the unit landed"
        );
        assert!(tables.get("people").unwrap().get("p2").unwrap().is_none());
        assert!(!fx.journal_path.exists(), "a refused prepare writes no journal");
    }

    #[test]
    fn a_patch_that_duplicates_a_put_made_earlier_in_the_same_unit_is_refused_too() {
        let fx = fixture();
        let mut tables = open_people(&fx);
        tables
            .get_mut("people")
            .unwrap()
            .put(&json!({ "id": "p2", "email": "bea@example.com" }), false)
            .unwrap();

        let mut unit = UnitOfWork::new();
        unit.put("people", json!({ "id": "p1", "email": "ala@example.com" }));
        unit.patch("people", "p2", json!({ "email": "ala@example.com" }));

        let error = unit.commit(&mut tables, &fx.journal_path).unwrap_err();
        assert!(is_unique_violation(&error, "p1"), "{error}");
        assert_eq!(
            tables.get("people").unwrap().get("p2").unwrap().unwrap().get("email"),
            Some(&json!("bea@example.com")),
            "the committed record kept the address the refused patch would have taken"
        );
    }

    /// The committed index still names `p1` as the holder while the unit is being prepared, but the
    /// same unit takes `p1` away first — so the value is free by the time anything lands.
    #[test]
    fn a_unit_may_hand_an_email_from_the_record_it_deletes_to_the_one_it_writes() {
        let fx = fixture();
        let mut tables = open_people(&fx);
        tables
            .get_mut("people")
            .unwrap()
            .put(&json!({ "id": "p1", "email": "ala@example.com" }), false)
            .unwrap();

        let mut unit = UnitOfWork::new();
        unit.delete("people", "p1");
        unit.put("people", json!({ "id": "p2", "email": "ala@example.com" }));

        unit.commit(&mut tables, &fx.journal_path).unwrap();
        assert!(tables.get("people").unwrap().get("p1").unwrap().is_none());
        assert_eq!(
            tables.get("people").unwrap().get("p2").unwrap().unwrap().get("email"),
            Some(&json!("ala@example.com"))
        );
    }

    #[test]
    fn a_patch_merges_over_the_put_the_same_unit_made_of_a_key_the_log_does_not_hold_yet() {
        let fx = fixture();
        let mut tables = open_people(&fx);

        let mut unit = UnitOfWork::new();
        unit.put(
            "people",
            json!({ "id": "p1", "email": "ala@example.com", "bio": "backend" }),
        );
        unit.patch("people", "p1", json!({ "bio": "backend developer" }));

        unit.commit(&mut tables, &fx.journal_path).unwrap();
        let landed = tables.get("people").unwrap().get("p1").unwrap().unwrap();
        assert_eq!(
            landed.get("bio"),
            Some(&json!("backend developer")),
            "the patch won over the put"
        );
        assert_eq!(
            landed.get("email"),
            Some(&json!("ala@example.com")),
            "a field the patch left alone survived the merge instead of being dropped"
        );
    }

    #[test]
    fn a_patch_of_a_key_the_same_unit_deleted_is_refused_as_absent() {
        let fx = fixture();
        let mut tables = open_people(&fx);
        tables
            .get_mut("people")
            .unwrap()
            .put(&json!({ "id": "p1", "email": "ala@example.com" }), false)
            .unwrap();

        let mut unit = UnitOfWork::new();
        unit.delete("people", "p1");
        unit.patch("people", "p1", json!({ "email": "bea@example.com" }));

        let error = unit.commit(&mut tables, &fx.journal_path).unwrap_err();
        assert!(
            matches!(&error, TxnError::Table(name, TableError::NotFound(missing)) if name == "people" && missing.key == "p1"),
            "the staged delete makes the key absent, it does not fall back to the committed record: {error}"
        );
        assert_eq!(
            tables.get("people").unwrap().get("p1").unwrap().unwrap().get("email"),
            Some(&json!("ala@example.com")),
            "the refused unit left the committed record untouched"
        );
    }

    #[test]
    fn a_search_through_the_unit_leaves_out_a_staged_record_the_term_is_not_in() {
        let fx = fixture();
        let mut tables = open_people(&fx);
        tables
            .get_mut("people")
            .unwrap()
            .put(
                &json!({ "id": "p1", "email": "ala@example.com", "bio": "backend developer" }),
                false,
            )
            .unwrap();
        let people = tables.get("people").unwrap();

        let mut unit = UnitOfWork::new();
        unit.put(
            "people",
            json!({ "id": "p2", "email": "bea@example.com", "bio": "frontend designer" }),
        );

        let backend = QueryBuilder::new().search("backend", SearchMode::All).spec();
        let designer = QueryBuilder::new().search("designer", SearchMode::All).spec();

        assert_eq!(
            unit.query("people", people, &backend).unwrap().keys,
            vec!["p1".to_string()],
            "the staged record says nothing about backends"
        );
        assert_eq!(
            unit.query("people", people, &designer).unwrap().keys,
            vec!["p2".to_string()],
            "and it is still found by a term it does carry"
        );
        assert!(
            people.query(&designer).unwrap().keys.is_empty(),
            "the committed state cannot see the staged record at all"
        );
    }

    #[test]
    fn a_search_through_the_unit_drops_a_record_a_staged_patch_takes_the_term_out_of() {
        let fx = fixture();
        let mut tables = open_people(&fx);
        tables
            .get_mut("people")
            .unwrap()
            .put(
                &json!({ "id": "p1", "email": "ala@example.com", "bio": "backend developer" }),
                false,
            )
            .unwrap();
        let people = tables.get("people").unwrap();

        let mut unit = UnitOfWork::new();
        unit.patch("people", "p1", json!({ "bio": "frontend designer" }));

        let backend = QueryBuilder::new().search("backend", SearchMode::All).spec();
        assert_eq!(people.query(&backend).unwrap().keys, vec!["p1".to_string()]);
        assert!(
            unit.query("people", people, &backend).unwrap().keys.is_empty(),
            "the patch takes the term away, so the unit must stop answering with the key"
        );
    }

    #[test]
    fn a_unit_answers_a_prefix_search_with_a_record_only_it_has_staged() {
        let fx = fixture();
        let tables = open_people(&fx);
        let people = tables.get("people").unwrap();

        let mut unit = UnitOfWork::new();
        unit.put(
            "people",
            json!({ "id": "p1", "email": "ala@example.com", "bio": "Backend architecture" }),
        );

        let hit = QueryBuilder::new().search("architec*", SearchMode::All).spec();
        let miss = QueryBuilder::new().search("architek* backend", SearchMode::All).spec();

        assert_eq!(unit.query("people", people, &hit).unwrap().keys, vec!["p1".to_string()]);
        assert!(
            unit.query("people", people, &miss).unwrap().keys.is_empty(),
            "\"architek\" is no prefix of \"architecture\", and All wants every term"
        );
    }

    #[test]
    fn a_unit_refused_for_a_bad_document_is_refused_again_because_staging_is_append_only() {
        let fx = fixture();
        let mut tables = open_tables(&fx);

        let mut unit = UnitOfWork::new();
        // "total" is required by the orders schema — no later op can take this one back.
        unit.put("orders", json!({ "id": "o1", "userId": "u1" }));
        unit.commit(&mut tables, &fx.journal_path).unwrap_err();

        unit.put("orders", json!({ "id": "o1", "userId": "u1", "total": 10 }));

        let error = unit.commit(&mut tables, &fx.journal_path).unwrap_err();
        assert!(matches!(error, TxnError::Table(name, _) if name == "orders"));
        assert!(tables.get("orders").unwrap().get("o1").unwrap().is_none());
        assert!(!fx.journal_path.exists());
    }

    #[test]
    fn a_unit_that_landed_refuses_a_second_commit_instead_of_writing_twice() {
        let fx = fixture();
        let mut tables = open_tables(&fx);

        let mut unit = UnitOfWork::new();
        unit.put("orders", json!({ "id": "o1", "userId": "u1", "total": 10 }));
        unit.commit(&mut tables, &fx.journal_path).unwrap();
        let after_first = tables.get("orders").unwrap().current_seq();

        let error = unit.commit(&mut tables, &fx.journal_path).unwrap_err();
        assert!(matches!(error, TxnError::AlreadyCommitted));
        assert_eq!(
            tables.get("orders").unwrap().current_seq(),
            after_first,
            "the refused second attempt did not even reserve a seq"
        );
    }

    #[test]
    fn nothing_to_recover_when_no_journal_was_ever_written() {
        let fx = fixture();
        let mut tables = open_tables(&fx);

        let recovered = recover_incomplete(&fx.journal_path, &mut tables).unwrap();
        assert!(recovered.is_none());
        assert!(tables.get("users").unwrap().get("u1").unwrap().is_none());
    }

    #[test]
    fn recover_incomplete_finishes_a_journal_left_by_a_crash_between_journal_and_write() {
        let fx = fixture();
        {
            let mut users = Table::open(users_schema(), &fx.users_path).unwrap();
            let mut orders = Table::open(orders_schema(), &fx.orders_path).unwrap();

            let user_entries = users
                .prepare_txn_ops(&[TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
                .unwrap();
            let order_entries = orders
                .prepare_txn_ops(&[TxnOp::Put(json!({ "id": "o1", "userId": "u1", "total": 10 }))])
                .unwrap();

            let mut entries = Vec::new();
            for entry in &user_entries {
                entries.push(JournalEntry {
                    t: "users".into(),
                    s: entry.seq(),
                    l: entry.line().to_string(),
                });
            }
            for entry in &order_entries {
                entries.push(JournalEntry {
                    t: "orders".into(),
                    s: entry.seq(),
                    l: entry.line().to_string(),
                });
            }

            journal::write_journal(
                &fx.journal_path,
                &Journal {
                    header: JournalHeader {
                        journal_version: JOURNAL_VERSION,
                        kind: "unit".into(),
                        id: "crash-test".into(),
                        started_at: 0,
                        tables: vec!["users".into(), "orders".into()],
                    },
                    entries,
                },
            )
            .unwrap();
            // Process dies here: prepare ran (users/orders in-memory seq is bumped and the journal is
            // durable), but apply_txn_entries was never called on either table.
        }

        assert!(fx.journal_path.exists());
        let mut tables = open_tables(&fx);
        assert!(tables.get("users").unwrap().get("u1").unwrap().is_none());
        assert!(tables.get("orders").unwrap().get("o1").unwrap().is_none());

        let result = recover_incomplete(&fx.journal_path, &mut tables)
            .unwrap()
            .expect("journal must be found");
        assert_eq!(result.id, "crash-test");
        assert_eq!(result.operations, 2);
        assert!(!fx.journal_path.exists());

        assert_eq!(
            tables.get("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        assert_eq!(
            tables.get("orders").unwrap().get("o1").unwrap().unwrap().get("total"),
            Some(&json!(10))
        );
    }

    #[test]
    fn recovering_twice_is_a_no_op_the_second_time() {
        let fx = fixture();
        {
            let mut users = Table::open(users_schema(), &fx.users_path).unwrap();
            let entries = users
                .prepare_txn_ops(&[TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
                .unwrap();
            let mut journal_entries = Vec::new();
            for entry in &entries {
                journal_entries.push(JournalEntry {
                    t: "users".into(),
                    s: entry.seq(),
                    l: entry.line().to_string(),
                });
            }
            journal::write_journal(
                &fx.journal_path,
                &Journal {
                    header: JournalHeader {
                        journal_version: JOURNAL_VERSION,
                        kind: "unit".into(),
                        id: "once".into(),
                        started_at: 0,
                        tables: vec!["users".into()],
                    },
                    entries: journal_entries,
                },
            )
            .unwrap();
        }

        let mut tables = open_tables(&fx);
        recover_incomplete(&fx.journal_path, &mut tables)
            .unwrap()
            .expect("first recovery finds it");
        let second = recover_incomplete(&fx.journal_path, &mut tables).unwrap();
        assert!(second.is_none());
        assert_eq!(
            tables.get("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
    }

    fn audit_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "audit",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "beforeName": { "type": "string" },
                "afterName": { "type": "string" },
            },
        }))
        .unwrap()
    }

    fn insert_audit_trigger() -> crate::catalog::TriggerEntry {
        crate::catalog::TriggerEntry {
            name: "users_insert_audit".into(),
            table: "users".into(),
            event: TriggerEvent::Insert,
            effect: crate::catalog::TriggerEffect {
                table: "audit".into(),
                fields: vec![
                    crate::catalog::TriggerFieldMapping {
                        target: "id".into(),
                        source: crate::catalog::TriggerValueSource::NewField("id".into()),
                    },
                    crate::catalog::TriggerFieldMapping {
                        target: "afterName".into(),
                        source: crate::catalog::TriggerValueSource::NewField("name".into()),
                    },
                ],
            },
        }
    }

    fn update_audit_trigger() -> crate::catalog::TriggerEntry {
        crate::catalog::TriggerEntry {
            name: "users_update_audit".into(),
            table: "users".into(),
            event: TriggerEvent::Update,
            effect: crate::catalog::TriggerEffect {
                table: "audit".into(),
                fields: vec![
                    crate::catalog::TriggerFieldMapping {
                        target: "id".into(),
                        source: crate::catalog::TriggerValueSource::NewField("id".into()),
                    },
                    crate::catalog::TriggerFieldMapping {
                        target: "beforeName".into(),
                        source: crate::catalog::TriggerValueSource::OldField("name".into()),
                    },
                    crate::catalog::TriggerFieldMapping {
                        target: "afterName".into(),
                        source: crate::catalog::TriggerValueSource::NewField("name".into()),
                    },
                ],
            },
        }
    }

    fn delete_audit_trigger() -> crate::catalog::TriggerEntry {
        crate::catalog::TriggerEntry {
            name: "users_delete_audit".into(),
            table: "users".into(),
            event: TriggerEvent::Delete,
            effect: crate::catalog::TriggerEffect {
                table: "audit".into(),
                fields: vec![
                    crate::catalog::TriggerFieldMapping {
                        target: "id".into(),
                        source: crate::catalog::TriggerValueSource::OldField("id".into()),
                    },
                    crate::catalog::TriggerFieldMapping {
                        target: "beforeName".into(),
                        source: crate::catalog::TriggerValueSource::OldField("name".into()),
                    },
                ],
            },
        }
    }

    #[test]
    fn a_delete_trigger_sees_the_record_the_same_unit_put_moments_earlier() {
        let dir = tempdir().unwrap();
        let mut tables = HashMap::new();
        tables.insert(
            "users".to_string(),
            Table::open(users_schema(), dir.path().join("users.jsonl")).unwrap(),
        );
        tables.insert(
            "audit".to_string(),
            Table::open(audit_schema(), dir.path().join("audit.jsonl")).unwrap(),
        );
        tables
            .get_mut("users")
            .unwrap()
            .set_triggers(vec![delete_audit_trigger()]);
        let journal_path = dir.path().join(super::journal::JOURNAL_FILE);

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        unit.delete("users", "u1");
        unit.commit(&mut tables, &journal_path).unwrap();

        assert!(tables.get("users").unwrap().get("u1").unwrap().is_none());
        let audit = tables
            .get("audit")
            .unwrap()
            .get("u1")
            .unwrap()
            .expect("the delete fired the trigger");
        assert_eq!(audit.get("beforeName"), Some(&json!("Ania")));
    }

    #[test]
    fn a_put_over_a_key_the_same_unit_deleted_fires_insert_not_update() {
        let dir = tempdir().unwrap();
        let mut tables = HashMap::new();
        tables.insert(
            "users".to_string(),
            Table::open(users_schema(), dir.path().join("users.jsonl")).unwrap(),
        );
        tables.insert(
            "audit".to_string(),
            Table::open(audit_schema(), dir.path().join("audit.jsonl")).unwrap(),
        );
        let journal_path = dir.path().join(super::journal::JOURNAL_FILE);

        let mut seed = UnitOfWork::new();
        seed.put("users", json!({ "id": "u1", "name": "Ania" }));
        seed.commit(&mut tables, &journal_path).unwrap();

        tables
            .get_mut("users")
            .unwrap()
            .set_triggers(vec![insert_audit_trigger(), update_audit_trigger()]);

        let mut unit = UnitOfWork::new();
        unit.delete("users", "u1");
        unit.put("users", json!({ "id": "u1", "name": "Zosia" }));
        unit.commit(&mut tables, &journal_path).unwrap();

        let audit = tables
            .get("audit")
            .unwrap()
            .get("u1")
            .unwrap()
            .expect("the put fired a trigger");
        assert_eq!(audit.get("afterName"), Some(&json!("Zosia")));
        assert_eq!(
            audit.get("beforeName"),
            None,
            "the key was dead when the put ran, so this is an insert"
        );
    }

    #[test]
    fn a_cross_table_insert_trigger_effect_lands_in_the_same_unit_of_work() {
        let dir = tempdir().unwrap();
        let mut tables = HashMap::new();
        tables.insert(
            "users".to_string(),
            Table::open(users_schema(), dir.path().join("users.jsonl")).unwrap(),
        );
        tables.insert(
            "audit".to_string(),
            Table::open(audit_schema(), dir.path().join("audit.jsonl")).unwrap(),
        );
        tables
            .get_mut("users")
            .unwrap()
            .set_triggers(vec![insert_audit_trigger()]);
        let journal_path = dir.path().join(super::journal::JOURNAL_FILE);

        let mut unit = UnitOfWork::new();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        let result = unit.commit(&mut tables, &journal_path).unwrap();

        assert!(result.tables.contains(&"audit".to_string()));
        assert!(!journal_path.exists());
        assert_eq!(
            tables
                .get("audit")
                .unwrap()
                .get("u1")
                .unwrap()
                .unwrap()
                .get("afterName"),
            Some(&json!("Ania"))
        );
    }

    #[test]
    fn an_update_trigger_sees_old_and_new_of_the_same_patch_at_once() {
        let dir = tempdir().unwrap();
        let mut tables = HashMap::new();
        tables.insert(
            "users".to_string(),
            Table::open(users_schema(), dir.path().join("users.jsonl")).unwrap(),
        );
        tables.insert(
            "audit".to_string(),
            Table::open(audit_schema(), dir.path().join("audit.jsonl")).unwrap(),
        );
        let journal_path = dir.path().join(super::journal::JOURNAL_FILE);

        let mut seed = UnitOfWork::new();
        seed.put("users", json!({ "id": "u1", "name": "Ania" }));
        seed.commit(&mut tables, &journal_path).unwrap();

        tables
            .get_mut("users")
            .unwrap()
            .set_triggers(vec![update_audit_trigger()]);

        let mut unit = UnitOfWork::new();
        unit.patch("users", "u1", json!({ "name": "Anna" }));
        unit.commit(&mut tables, &journal_path).unwrap();

        let audit = tables.get("audit").unwrap().get("u1").unwrap().unwrap();
        assert_eq!(audit.get("beforeName"), Some(&json!("Ania")));
        assert_eq!(audit.get("afterName"), Some(&json!("Anna")));
    }

    fn seeded_people(fx: &Fixture) -> HashMap<String, Table> {
        let mut tables = open_people(fx);
        let people = tables.get_mut("people").unwrap();
        people
            .put(
                &json!({ "id": "p1", "email": "ala@example.com", "bio": "backend developer" }),
                false,
            )
            .unwrap();
        people
            .put(
                &json!({ "id": "p2", "email": "bea@example.com", "bio": "frontend designer" }),
                false,
            )
            .unwrap();
        tables
    }

    fn overlay_of(unit: &UnitOfWork, people: &Table, spec: &QuerySpec) -> PlanOverlay {
        unit.query("people", people, spec)
            .unwrap()
            .plan
            .overlay
            .expect("a unit query always plans through an overlay")
    }

    #[test]
    fn a_staged_put_that_matches_counts_as_one_staged_candidate_and_suppresses_nothing() {
        let fx = fixture();
        let tables = seeded_people(&fx);
        let people = tables.get("people").unwrap();

        let mut unit = UnitOfWork::new();
        unit.put(
            "people",
            json!({ "id": "p3", "email": "cela@example.com", "bio": "backend developer" }),
        );

        let spec = QueryBuilder::new()
            .r#where("email", ComparisonOp::Eq, json!("cela@example.com"))
            .spec();
        let overlay = overlay_of(&unit, people, &spec);

        assert_eq!(overlay.staged, 1, "the staged record joined the candidates");
        assert_eq!(
            overlay.suppressed, 0,
            "no committed candidate was taken out — the index held none"
        );
        assert!(!overlay.rescored, "no search ran, so nothing was scored at all");
        assert_eq!(
            overlay.strategy,
            OverlayStrategy::Exact,
            "email is indexed, so the overlay narrowed on it"
        );
        assert_eq!(overlay.size, 1);
        assert_eq!(
            unit.query("people", people, &spec).unwrap().keys,
            vec!["p3".to_string()]
        );
    }

    #[test]
    fn a_staged_delete_suppresses_the_committed_candidate_the_index_still_holds() {
        let fx = fixture();
        let tables = seeded_people(&fx);
        let people = tables.get("people").unwrap();

        let mut unit = UnitOfWork::new();
        unit.delete("people", "p1");

        let spec = QueryBuilder::new()
            .r#where("email", ComparisonOp::Eq, json!("ala@example.com"))
            .spec();
        let overlay = overlay_of(&unit, people, &spec);

        assert_eq!(
            overlay.suppressed, 1,
            "the committed index still holds p1, the overlay takes it back out"
        );
        assert_eq!(overlay.staged, 0, "a delete stages no record to put back in");
        assert!(!overlay.rescored);
        assert_eq!(overlay.size, 0, "a deleted key is not a staged record");
        assert!(unit.query("people", people, &spec).unwrap().keys.is_empty());
    }

    #[test]
    fn a_staged_patch_suppresses_the_committed_candidate_and_re_stages_it_in_its_new_state() {
        let fx = fixture();
        let tables = seeded_people(&fx);
        let people = tables.get("people").unwrap();

        let mut unit = UnitOfWork::new();
        unit.patch("people", "p1", json!({ "email": "nowa@example.com" }));

        let by_old = QueryBuilder::new()
            .r#where("email", ComparisonOp::Eq, json!("ala@example.com"))
            .spec();
        let by_new = QueryBuilder::new()
            .r#where("email", ComparisonOp::Eq, json!("nowa@example.com"))
            .spec();

        let old = overlay_of(&unit, people, &by_old);
        assert_eq!(
            old.suppressed, 1,
            "the committed index still answers the old email, the overlay overrides it"
        );
        assert_eq!(old.staged, 0, "and the patched record no longer carries that value");
        assert!(unit.query("people", people, &by_old).unwrap().keys.is_empty());

        let new = overlay_of(&unit, people, &by_new);
        assert_eq!(
            new.staged, 1,
            "the overlay index answers a value the committed one never had"
        );
        assert_eq!(new.suppressed, 0);
        assert_eq!(
            unit.query("people", people, &by_new).unwrap().keys,
            vec!["p1".to_string()]
        );
    }

    #[test]
    fn a_search_through_a_unit_is_rescored_over_the_corpus_the_unit_will_leave() {
        let fx = fixture();
        let tables = seeded_people(&fx);
        let people = tables.get("people").unwrap();

        let mut unit = UnitOfWork::new();
        unit.put(
            "people",
            json!({ "id": "p3", "email": "cela@example.com", "bio": "backend architect" }),
        );
        unit.delete("people", "p1");

        let spec = QueryBuilder::new().search("backend", SearchMode::All).spec();
        let overlay = overlay_of(&unit, people, &spec);

        assert!(
            overlay.rescored,
            "the BM25 statistics were recomputed with p3 spliced in and p1 hidden"
        );
        assert_eq!(
            overlay.strategy,
            OverlayStrategy::Fulltext,
            "the staged half was filtered by the search text"
        );
        assert_eq!(
            overlay.staged, 1,
            "only p3 scores on \"backend\" among the staged records"
        );
        assert_eq!(
            overlay.suppressed, 1,
            "p3 scored, so it entered the committed path's key set and the overlay took it back for itself"
        );
        assert_eq!(
            unit.query("people", people, &spec).unwrap().keys,
            vec!["p3".to_string()],
            "the committed p1 is staged deleted, so only the staged p3 is left"
        );
        assert_eq!(
            people.query(&spec).unwrap().keys,
            vec!["p1".to_string()],
            "the committed state was never rescored and never saw p3"
        );
    }

    #[test]
    fn a_search_without_an_overlay_index_falls_back_to_checking_every_staged_record() {
        let fx = fixture();
        let tables = seeded_people(&fx);
        let people = tables.get("people").unwrap();

        let mut unit = UnitOfWork::new();
        unit.put(
            "people",
            json!({ "id": "p3", "email": "cela@example.com", "bio": "backend architect" }),
        );

        let spec = QueryBuilder::new()
            .r#where("bio", ComparisonOp::Contains, json!("architect"))
            .spec();
        let overlay = overlay_of(&unit, people, &spec);

        assert_eq!(
            overlay.strategy,
            OverlayStrategy::Scan,
            "no overlay index answers a `contains`"
        );
        assert_eq!(overlay.staged, 1, "so every staged record is a candidate");
        assert!(!overlay.rescored);
    }

    #[test]
    fn a_unit_with_nothing_staged_still_reports_an_empty_overlay() {
        let fx = fixture();
        let tables = seeded_people(&fx);
        let people = tables.get("people").unwrap();

        let spec = QueryBuilder::new()
            .r#where("email", ComparisonOp::Eq, json!("ala@example.com"))
            .spec();
        let overlay = overlay_of(&UnitOfWork::new(), people, &spec);

        assert_eq!(
            overlay,
            PlanOverlay {
                staged: 0,
                suppressed: 0,
                rescored: false,
                strategy: OverlayStrategy::Exact,
                size: 0
            },
            "an empty unit changes nothing, and says so rather than leaving the plan silent"
        );
        assert!(
            people.query(&spec).unwrap().plan.overlay.is_none(),
            "a plain table query has no overlay at all"
        );
    }
}