hayson-engine 0.11.0

use std::borrow::Cow;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt;
use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use serde_json::Value;

use crate::catalog::{
    CatalogCompaction, CatalogSyncMode, CatalogValidation, ExpandedIndex, LinkParsing, TriggerEntry, TriggerEvent,
};
use crate::checkpoint::{
    remove_snapshot, restore_or_rebuild_set_with_declared, write_snapshot_set, RestoreOutcome, Restored,
};
use crate::crypto::{
    assert_carried_seals, cypher_fields_of, envelope_key_id, open_record, seal_record, FieldCipher, MissingKeyPolicy,
};
use crate::errors::{
    HaysonCipherError, HaysonNotFoundError, HaysonPartitionError, HaysonPendingConflictError,
    HaysonRelationViolationError, HaysonSchemaError, HaysonUniqueViolationError, HaysonValidationError,
};
use crate::query::aggregate::{
    execute_aggregate, execute_grouped_aggregate, AggregateResult, AggregateSpec, GroupedAggregateResult,
};
use crate::query::planner::{
    execute_access, execute_query, plan_access_overlaid, LogSetRecordSource, QueryOverlay, QueryResult, QuerySpec,
    RecordSource, RecordSourceError,
};
use crate::queue::write_queue::{assert_partition_stable, assert_unique_in_unit, UnitShadow};
use crate::queue::{StageError, WriteOutcome, WriteQueue};
use crate::registry::fulltext_index::CorpusDelta;
use crate::registry::table_registry::TableRegistry;
use crate::schema::{
    broken_links, is_partitioned, partition_of, physical_path_of, primary_key_of, undeclared_fields, validate_record,
    without_undeclared, FieldType, HaysonRecord, TableSchema, ValidateOptions, UNPARTITIONED,
};
use crate::storage::{
    ensure_partitioned, parse_entry, partition_names_in, serialize_entry, LogEntry, LogFile, LogMeta, LogOp, LogProbe,
    LogSet, ScanResult, Slot,
};

use super::cache::{CachePolicy, RecordCache};
use super::trigger::{effect_document, matching, TriggerFiring};

#[derive(Debug)]
pub enum TableError {
    Validation(HaysonValidationError),
    Unique(HaysonUniqueViolationError),
    Conflict(HaysonPendingConflictError),
    NotFound(HaysonNotFoundError),
    Schema(HaysonSchemaError),
    Partition(HaysonPartitionError),
    Relation(HaysonRelationViolationError),
    /// Boxed — `HaysonCipherError` is the widest of these variants, and only the `cypher`-fields
    /// path ever builds one, so nothing else should pay for its size at rest.
    Cipher(Box<HaysonCipherError>),
    Io(io::Error),
}

impl fmt::Display for TableError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            TableError::Validation(e) => e.fmt(f),
            TableError::Unique(e) => e.fmt(f),
            TableError::Conflict(e) => e.fmt(f),
            TableError::NotFound(e) => e.fmt(f),
            TableError::Schema(e) => e.fmt(f),
            TableError::Partition(e) => e.fmt(f),
            TableError::Relation(e) => e.fmt(f),
            TableError::Cipher(e) => e.fmt(f),
            TableError::Io(e) => e.fmt(f),
        }
    }
}

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

impl From<RecordSourceError> for TableError {
    fn from(value: RecordSourceError) -> Self {
        match value {
            RecordSourceError::Cipher(error) => TableError::Cipher(error),
            RecordSourceError::Io(error) => TableError::Io(error),
        }
    }
}

impl From<StageError> for TableError {
    fn from(value: StageError) -> Self {
        match value {
            StageError::Validation(e) => TableError::Validation(e),
            StageError::Unique(e) => TableError::Unique(e),
            StageError::Conflict(e) => TableError::Conflict(e),
            StageError::Partition(e) => TableError::Partition(e),
        }
    }
}

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

impl From<HaysonValidationError> for TableError {
    fn from(value: HaysonValidationError) -> Self {
        TableError::Validation(value)
    }
}

impl From<HaysonPendingConflictError> for TableError {
    fn from(value: HaysonPendingConflictError) -> Self {
        TableError::Conflict(value)
    }
}

impl From<HaysonUniqueViolationError> for TableError {
    fn from(value: HaysonUniqueViolationError) -> Self {
        TableError::Unique(value)
    }
}

impl From<HaysonPartitionError> for TableError {
    fn from(value: HaysonPartitionError) -> Self {
        TableError::Partition(value)
    }
}

impl From<HaysonCipherError> for TableError {
    fn from(value: HaysonCipherError) -> Self {
        TableError::Cipher(Box::new(value))
    }
}

/// Lets the `io::Result` paths that now flush first (`compact`, `snapshot`, `query_following` and
/// friends) carry a flush failure without changing their signature. Only the `Io` variant survives
/// intact; the rest cannot be raised by a flush of writes that already passed staging, apart from a
/// trigger effect the schema refuses, which arrives here as its message.
impl From<TableError> for io::Error {
    fn from(value: TableError) -> Self {
        match value {
            TableError::Io(error) => error,
            other => io::Error::other(other.to_string()),
        }
    }
}

impl From<crate::queue::CommitError> for TableError {
    fn from(value: crate::queue::CommitError) -> Self {
        match value {
            crate::queue::CommitError::Io(error) => TableError::Io(error),
            crate::queue::CommitError::Cipher(error) => TableError::Cipher(Box::new(error)),
        }
    }
}

/// One markdown link of a record that resolves to nothing — the field the link was found in
/// against the virtual path it aimed at, kept apart rather than formatted into a sentence, so a
/// caller can group by field or retry the target without parsing text back out.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BrokenLink {
    pub field: String,
    pub path: String,
}

/// The result of [`Table::put_with_warnings`]/[`Table::patch_with_warnings`] — the written record
/// plus diagnostics that do not block the write (currently only broken markdown links found while
/// [`Table::link_parsing`] is `Eager`; empty for a `Lazy` table, which surfaces them only on
/// request via [`Table::broken_links_of`]). [`Table::put`]/[`Table::patch`] stay untouched and
/// keep returning a bare `HaysonRecord` on a separate method, rather than changing their own
/// signature, so no existing caller has to start handling warnings it never asked for.
///
/// `warnings` carries broken links and nothing else: the two other diagnostics this table answers
/// for — a dropped undeclared field and a write touching a `deprecated` one — have channels of
/// their own ([`Table::dropped_field_warnings`], [`Table::deprecated_field_warnings`]) because
/// neither can be recomputed from the stored record. That is why this field could take the
/// structured [`BrokenLink`] rather than staying a bag of formatted strings.
pub struct PutOutcome {
    pub record: HaysonRecord,
    pub warnings: Vec<BrokenLink>,
}

impl PutOutcome {
    fn into_record(self) -> HaysonRecord {
        self.record
    }
}

impl From<PutOutcome> for HaysonRecord {
    fn from(outcome: PutOutcome) -> Self {
        outcome.record
    }
}

/// Which of the three things [`Table::rotate_cypher_keys`] can do to a record it read applies.
enum RotationVerdict {
    /// Every sealed field already names the current write key.
    Current,
    /// At least one field is on another key, or in the clear, and the keyring holds what it needs.
    Rotate,
    /// A field names this key id, which the keyring does not hold.
    Missing(String),
}

/// What one run of [`Table::rotate_cypher_keys`] did. Counts records, not fields: a record with
/// two `cypher` fields on a retired key is rotated once and counted once.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct RotationResult {
    /// Records read, decrypted and written back under the current write key.
    pub rotated: usize,
    /// Records every `cypher` field of which was already sealed under the current write key —
    /// left exactly as they were, which is what makes a second run free.
    pub already_current: usize,
    /// Records left alone because a field of theirs names a key the keyring does not hold. Only
    /// ever non-zero under [`MissingKeyPolicy::Omit`]/[`MissingKeyPolicy::Raw`]; the default
    /// `Throw` refuses the whole run instead — see [`Table::rotate_cypher_keys`].
    pub skipped: usize,
}

/// What [`Table::refresh`] found on disk since this handle last looked. Modelled on
/// [`RestoreOutcome`], which reports the same kind of thing about an open: not whether it worked,
/// but which of its two ways it had to work.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RefreshOutcome {
    /// Every log file this handle holds is exactly where it left it — one `stat` per partition was
    /// read and nothing else.
    Unchanged,
    /// Lines another writer appended past this handle's watermark were scanned and applied on top
    /// of the registry it already had.
    CaughtUp { entries: usize },
    /// A log file was replaced underneath this handle — what the rewrite half of a foreign
    /// [`Table::compact`] does — so the registry was thrown away and rebuilt from the files as they
    /// are now. Slots recorded before such a rewrite name offsets that no longer mean anything,
    /// which is exactly why this case cannot be served by scanning onwards from a watermark.
    Reloaded,
}

impl RefreshOutcome {
    /// Whether anything changed at all — the boolean the TS `Table.refresh()` answers with.
    pub fn changed(self) -> bool {
        self != RefreshOutcome::Unchanged
    }
}

/// How often a handle catches up with foreign writes on its own, before serving a read — the port
/// of the TS `FollowPolicy` (`packages/hayson/src/table/table.ts`).
///
/// It exists for a readonly handle, which never writes and so has no other occasion to notice what
/// another process committed. A writable handle already sees its own writes and is the only writer
/// the directory allows, so it follows nothing by default.
///
/// `interval` is a floor on how often the check is paid for, not a schedule: nothing runs in the
/// background, and a handle nobody reads from never looks at the disk at all. `Duration::ZERO` —
/// the default a readonly open resolves to, matching the TS `DEFAULT_FOLLOW_INTERVAL_MS` of `0` —
/// means every read checks, which costs one `stat` per log file when nothing changed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FollowPolicy {
    pub interval: Duration,
}

impl Default for FollowPolicy {
    fn default() -> Self {
        FollowPolicy {
            interval: Duration::ZERO,
        }
    }
}

/// Where a key stands between [`Table::stage_put`]/[`Table::stage_patch`]/[`Table::stage_delete`]
/// and the [`Table::flush`] that commits them — one entry per key, holding the write that would win
/// if the flush ran now.
///
/// `old` is the committed record the key held when the *first* write of this staging window touched
/// it, not the one before the latest re-stage: the registry has already moved on by the time the
/// flush fires triggers, so the only chance to read it is here.
struct StagedWrite {
    /// Staging order, so a flush fires triggers in the order the writes were made rather than in
    /// whatever order the map iterates. A key staged again keeps the id of its first write.
    id: u64,
    /// `None` is a staged delete — a key whose reads answer "gone" before the tombstone is written.
    record: Option<HaysonRecord>,
    old: Option<HaysonRecord>,
    /// `false` for the buffer-replay path, which must not fire triggers a second time — see
    /// [`Table::put_from_buffer`].
    live: bool,
}

/// The counters behind [`Table::stats`], in the shape of the TS `TableStats`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TableStats {
    pub records: usize,
    pub live_bytes: u64,
    pub dead_bytes: u64,
    pub log_bytes: u64,
    /// Writes staged but not yet appended — `0` unless a caller staged through
    /// [`Table::stage_put`] and friends without flushing yet, since `put`/`patch`/`delete` stage and
    /// flush inside the same call.
    pub queue_depth: usize,
    /// The log's generation: how many times it has been rewritten under this registry, counting
    /// both this handle's own [`Table::compact`] and the rebuilds [`Table::refresh`] does after
    /// another process compacted underneath it. Persisted in the snapshot header and carried over
    /// on the next open, so it survives a restart rather than starting from `0` again.
    pub gen: u64,
    pub seq: u64,
}

/// One log line naming a record, as [`Table::history`] read it off the file — the version itself,
/// not the record as the live schema would project it.
#[derive(Debug, Clone, PartialEq)]
pub struct HistoryEntry {
    pub seq: u64,
    pub ts: i64,
    pub op: LogOp,
    /// The partition file this line was found in — [`UNPARTITIONED`] on a table with no
    /// `partitionBy`, and on one whose declared split has not run yet.
    pub partition: String,
    pub offset: u64,
    pub length: u64,
    /// The record exactly as the line stores it: `None` for a tombstone, and for a `put` still
    /// carrying whatever the schema of the day wrote — a sealed `cypher` envelope is not opened and
    /// a since-dropped field is not projected away.
    pub document: Option<HaysonRecord>,
}

/// Every version of one key, with the files that were read to find them — the answer of
/// [`Table::history`].
#[derive(Debug, Clone, PartialEq)]
pub struct TableHistory {
    /// Oldest first, ordered by `seq`, so a key that moved between partitions still reads as one
    /// history rather than as one list per file.
    pub entries: Vec<HistoryEntry>,
    /// The log files scanned, in partition-name order. Reported because the scan is the cost of the
    /// call, and on a partitioned table it says which file each version came from.
    pub scanned: Vec<PathBuf>,
}

fn value_as_key(value: &Value) -> String {
    match value {
        Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

/// Drops from a record every field the schema does not declare, so a `dropField` (and a log written
/// by an older schema) reads back as the current schema describes it. The log itself is never
/// rewritten — `dropField` leaves it byte for byte — so this is the only place the dropped field
/// stops existing for a reader.
///
/// Returns the record untouched when nothing has to go, which is every record of a table whose log
/// was only ever written under the schema it has now.
fn project_onto_schema(schema: &TableSchema, record: HaysonRecord) -> HaysonRecord {
    if record.keys().all(|field| schema.fields.contains_key(field)) {
        return record;
    }
    record
        .into_iter()
        .filter(|(field, _)| schema.fields.contains_key(field))
        .collect()
}

/// Default image-bytes directory for a table with no `options.imageDir` override: a sibling of its
/// log file, named after it, so opening `<dir>/<table>.jsonl` gives `<dir>/<table>.images/`.
fn default_image_dir(log_path: &Path, table: &str) -> PathBuf {
    let images = format!("{table}.images");
    match log_path.parent() {
        Some(parent) => parent.join(images),
        None => PathBuf::from(images),
    }
}

/// Ties `TableRegistry` (indexes), `WriteQueue` (validation, uniqueness, staging) and `LogFile`
/// (persistence) into one table API. Deliberately thin: put/patch/delete all delegate their
/// validation and unique-checking to `WriteQueue::stage_put`, they do not repeat it here.
pub struct Table {
    schema: TableSchema,
    registry: TableRegistry,
    logs: LogSet,
    queue: WriteQueue,
    /// The writes `queue` holds, keyed by the record they land on, so a read can answer from the
    /// staging window instead of the log — see [`StagedWrite`]. Emptied by [`Table::flush`].
    staged: HashMap<String, StagedWrite>,
    next_op_id: u64,
    seq: u64,
    snapshot_path: Option<PathBuf>,
    schema_hash: String,
    snapshot_every: u64,
    since_snapshot: u64,
    snapshot_current: bool,
    /// The thresholds [`Table::compact_if_due`] measures each partition against after a write.
    /// `Disabled` on every constructor, exactly as `snapshot_every` starts at `0`: a raw handle
    /// rewrites nothing it was not asked to. `Database::open` resolves the real policy from the
    /// catalog and hands it over through [`Table::set_compaction`].
    compaction: CatalogCompaction,
    /// How durable this table's appends are. `Commit` on every constructor — the safe default, and
    /// what every caller got before `options.sync` was read at all; `Database::open` resolves the
    /// declared mode from the catalog and hands it over through [`Table::set_sync`].
    sync: CatalogSyncMode,
    /// What a write naming a field the schema does not declare gets. `Strict` on every constructor
    /// — the refusal every caller has always had; `Database::open` resolves the declared mode from
    /// the table's catalog entry and hands it over through [`Table::set_validation`].
    validation: CatalogValidation,
    /// The fields the catalog marks `deprecated`, against their note. Empty on every constructor —
    /// the schema carries no such mark, so a raw handle has nothing to warn about;
    /// `Database::open` resolves the set from the table's catalog entry and hands it over through
    /// [`Table::set_deprecated_fields`].
    deprecated_fields: BTreeMap<String, String>,
    triggers: Vec<TriggerEntry>,
    image_dir: PathBuf,
    link_parsing: LinkParsing,
    cipher: Option<FieldCipher>,
    cypher_fields: Vec<String>,
    /// Kept past the open it was used for, because [`Table::refresh`] may have to rebuild the
    /// registry from scratch and a rebuilt one has to carry the same declared indexes as the
    /// original — dropping them here would silently retire every composite and `where`-scoped
    /// index the catalog declares the moment another process compacts.
    declared: Vec<ExpandedIndex>,
    gen: u64,
    /// `None` — the default every constructor starts from — is the TS `follow: false`: this handle
    /// only ever catches up when somebody calls [`Table::refresh`]. `Database::open` sets it.
    follow: Option<FollowPolicy>,
    /// When [`Table::follow_if_due`] last passed its interval gate. `None` until the first read, so
    /// the first one always checks however long the interval is.
    followed_at: Option<Instant>,
    /// Parsed documents [`Table::get`] may answer from instead of reading the log again.
    /// `Disabled` on every constructor, as `compaction` is: a raw handle caches nothing.
    /// `Database::open` hands over what `OpenOptions::cache` names, through [`Table::set_cache`].
    cache: RecordCache,
}

impl Table {
    /// Opens a table with no paired snapshot file — used for catalog operations (`drop_field`,
    /// `declare_index`) that reopen a table over its existing log right after rewriting the schema.
    /// `snapshot()`/`needs_snapshot()` are no-ops on a table opened this way, same as it never
    /// restores from a snapshot either.
    pub fn open(schema: TableSchema, path: impl AsRef<Path>) -> io::Result<Self> {
        Self::open_with_declared_indexes(schema, path, &[])
    }

    /// Same as [`Table::open`], but the registry also builds every composite or `where`-scoped
    /// index the catalog declares for this table (`catalog::expand_indexes`) alongside the sugar
    /// [`Table::open`] derives from the schema alone — called by `Database`, which owns the
    /// catalog `Table::open` itself has none of.
    pub fn open_with_declared_indexes(
        schema: TableSchema,
        path: impl AsRef<Path>,
        declared: &[ExpandedIndex],
    ) -> io::Result<Self> {
        let image_dir = default_image_dir(path.as_ref(), &schema.table);
        let cypher_fields = cypher_fields_of(&schema);
        let mut logs = open_logs(&schema, path.as_ref(), false)?;
        let registry = TableRegistry::build_from_log_set_with_declared(schema.clone(), &mut logs, declared)?;
        let seq = registry.max_seq();
        let queue = WriteQueue::new(schema.table.clone());
        Ok(Table {
            schema,
            registry,
            logs,
            queue,
            staged: HashMap::new(),
            next_op_id: 0,
            seq,
            snapshot_path: None,
            schema_hash: String::new(),
            snapshot_every: 0,
            since_snapshot: 0,
            snapshot_current: false,
            compaction: CatalogCompaction::Disabled,
            sync: CatalogSyncMode::Commit,
            validation: CatalogValidation::Strict,
            deprecated_fields: BTreeMap::new(),
            triggers: Vec::new(),
            image_dir,
            link_parsing: LinkParsing::Eager,
            cipher: None,
            cypher_fields,
            declared: declared.to_vec(),
            gen: 0,
            follow: None,
            followed_at: None,
            cache: RecordCache::new(CachePolicy::Disabled),
        })
    }

    /// Opens the log at `path` and restores its registry from `snapshot_path` via
    /// `checkpoint::restore_or_rebuild` instead of always scanning the whole log — used by
    /// `database::open()` at startup. Falls back to a full scan (`RestoreOutcome::Rebuilt`) exactly
    /// as `restore_or_rebuild` does when the snapshot is missing, stale, or paired with a different
    /// schema.
    ///
    /// `snapshot_every` mirrors the TS `snapshotEvery` table option (ADR/0047 line): after this many
    /// operations have been applied since the last snapshot, the next commit writes a fresh
    /// `snapshot_path` and resets the counter. `0` disables periodic snapshotting for this table.
    pub fn open_with_snapshot(
        schema: TableSchema,
        path: impl AsRef<Path>,
        snapshot_path: impl AsRef<Path>,
        schema_hash: &str,
        snapshot_every: u64,
    ) -> io::Result<(Self, RestoreOutcome)> {
        Self::open_with_snapshot_and_declared_indexes(
            schema,
            path,
            snapshot_path,
            schema_hash,
            snapshot_every,
            false,
            &[],
        )
    }

    /// Same as [`Table::open_with_snapshot`], but both the restored and the rebuilt registry are
    /// built with `declared` — the catalog's fully expanded index list (`catalog::expand_indexes`)
    /// — via `checkpoint::restore_or_rebuild_with_declared`.
    ///
    /// `readonly` is the open mode of the database this table belongs to, and it decides one thing
    /// here: whether a log whose tail was left half-written may be cut back to its last complete
    /// line (see `LogFile::discard_partial_tail`). A readonly handle only fixes
    /// its own idea of where the log ends and leaves the file alone.
    #[allow(clippy::too_many_arguments)]
    pub fn open_with_snapshot_and_declared_indexes(
        schema: TableSchema,
        path: impl AsRef<Path>,
        snapshot_path: impl AsRef<Path>,
        schema_hash: &str,
        snapshot_every: u64,
        readonly: bool,
        declared: &[ExpandedIndex],
    ) -> io::Result<(Self, RestoreOutcome)> {
        let image_dir = default_image_dir(path.as_ref(), &schema.table);
        let cypher_fields = cypher_fields_of(&schema);
        let mut logs = open_logs(&schema, path.as_ref(), readonly)?;
        let Restored { registry, outcome, gen } = restore_or_rebuild_set_with_declared(
            snapshot_path.as_ref(),
            schema.clone(),
            &mut logs,
            schema_hash,
            declared,
        )?;
        let seq = registry.max_seq();
        let queue = WriteQueue::new(schema.table.clone());
        let table = Table {
            schema,
            registry,
            logs,
            queue,
            staged: HashMap::new(),
            next_op_id: 0,
            seq,
            snapshot_path: Some(snapshot_path.as_ref().to_path_buf()),
            schema_hash: schema_hash.to_string(),
            snapshot_every,
            since_snapshot: 0,
            snapshot_current: outcome == RestoreOutcome::Restored,
            compaction: CatalogCompaction::Disabled,
            sync: CatalogSyncMode::Commit,
            validation: CatalogValidation::Strict,
            deprecated_fields: BTreeMap::new(),
            triggers: Vec::new(),
            image_dir,
            link_parsing: LinkParsing::Eager,
            cipher: None,
            cypher_fields,
            declared: declared.to_vec(),
            gen,
            follow: None,
            followed_at: None,
            cache: RecordCache::new(CachePolicy::Disabled),
        };
        Ok((table, outcome))
    }

    /// Attaches the catalog's trigger declarations naming this table as their source — called by
    /// `Database` after opening/reopening a table. A trigger whose `effect.table` differs from
    /// this table's own name only fires when the write reaches it through
    /// [`crate::txn::UnitOfWork::commit`], which holds every table of the database at once; the
    /// single-table `put`/`patch`/`delete` path below can only write into itself.
    pub fn set_triggers(&mut self, triggers: Vec<TriggerEntry>) {
        self.triggers = triggers;
    }

    pub fn triggers(&self) -> &[TriggerEntry] {
        &self.triggers
    }

    /// Overrides the directory `image` field bytes are written under — called by `Database` when
    /// `options.imageDir` names one, resolved relative to the database directory. Left at its
    /// `open`/`open_with_snapshot`-computed default (a `<table>.images/` sibling of the log file)
    /// otherwise.
    pub fn set_image_dir(&mut self, dir: PathBuf) {
        self.image_dir = dir;
    }

    pub fn image_dir(&self) -> &Path {
        &self.image_dir
    }

    /// Overrides when a markdown field's links get checked against this table's virtual-path
    /// index — called by `Database` from `options.linkParsing`, left at `Eager` otherwise.
    pub fn set_link_parsing(&mut self, mode: LinkParsing) {
        self.link_parsing = mode;
    }

    pub fn link_parsing(&self) -> LinkParsing {
        self.link_parsing
    }

    /// Overrides what a write naming a field the schema does not declare gets — called by
    /// `Database` from the table's `validation` entry, left at `Strict` otherwise. `Warn` drops
    /// such a field and lets the rest of the write through; nothing else about validation moves,
    /// so a bad type, a missing `required`, a primary key that is not there and a reserved `_` key
    /// are refused in either mode.
    pub fn set_validation(&mut self, mode: CatalogValidation) {
        self.validation = mode;
    }

    pub fn validation(&self) -> CatalogValidation {
        self.validation
    }

    /// `document` as this table takes it: unchanged on a `Strict` table, and without the fields the
    /// schema does not declare on a `Warn` one. Every write path runs its input through here
    /// before validating it, which is why nothing undeclared can reach the queue, the log or the
    /// registry — see [`Table::dropped_field_warnings`] for how a caller learns which fields went.
    pub(crate) fn accepted_document<'a>(&self, document: &'a Value) -> Cow<'a, Value> {
        match self.validation {
            CatalogValidation::Strict => Cow::Borrowed(document),
            CatalogValidation::Warn => without_undeclared(&self.schema, document),
        }
    }

    /// What a write of `document` has to say about fields the schema does not declare — empty on a
    /// `Strict` table, which refuses such a write instead of warning about it.
    ///
    /// A channel of its own, deliberately not folded into [`Table::eager_link_warnings_of`]: that
    /// one recomputes broken links from the stored record, and a dropped field is not in the
    /// stored record to be found again, so this has to be asked with the document that was
    /// written. Answers the same before and after the write, since it reads nothing but the schema.
    pub fn dropped_field_warnings(&self, document: &Value) -> Vec<String> {
        if self.validation != CatalogValidation::Warn {
            return Vec::new();
        }
        undeclared_fields(&self.schema, document)
            .into_iter()
            .map(|field| format!("field \"{field}\": is not declared in the schema, dropped from the write"))
            .collect()
    }

    /// Hands over the fields the catalog marks `deprecated`, each against its note — called by
    /// `Database` after opening/reopening a table, left empty otherwise. Nothing about a write
    /// changes: a deprecated field is declared, so it validates, lands and reads back exactly as
    /// any other; the mark is only ever reported, through [`Table::deprecated_field_warnings`].
    ///
    /// A note coming from the catalog is never empty — `catalog::parse_catalog` refuses
    /// `deprecated: ""` outright — but a caller building a handle by hand can hand one over, and
    /// [`Table::deprecated_field_warnings`] then names the field on its own.
    pub fn set_deprecated_fields(&mut self, fields: BTreeMap<String, String>) {
        self.deprecated_fields = fields;
    }

    pub fn deprecated_fields(&self) -> &BTreeMap<String, String> {
        &self.deprecated_fields
    }

    /// What a write of `document` has to say about fields the catalog marks `deprecated` — one
    /// message per marked field the document names, quoting the note where the catalog gives one.
    ///
    /// Independent of [`Table::validation`]: `deprecated` is a mark on the field, not a mode of the
    /// table, so a `strict` table warns here exactly as a `warn` one does. A channel of its own,
    /// deliberately not folded into [`Table::dropped_field_warnings`]: that one is about a field
    /// the schema does not declare and a `strict` table refuses the write for, this one about a
    /// field it does declare and every table takes. Answers the same before and after the write,
    /// since it reads nothing but the document and the mark.
    pub fn deprecated_field_warnings(&self, document: &Value) -> Vec<String> {
        if self.deprecated_fields.is_empty() {
            return Vec::new();
        }
        let Some(source) = document.as_object() else {
            return Vec::new();
        };
        source
            .keys()
            .filter_map(|field| self.deprecated_fields.get_key_value(field.as_str()))
            .map(|(field, note)| {
                if note.is_empty() {
                    format!("field \"{field}\": is deprecated")
                } else {
                    format!("field \"{field}\": is deprecated — {note}")
                }
            })
            .collect()
    }

    /// Arms automatic compaction with the thresholds `Database::open` resolved for this table —
    /// see [`Table::compact_if_due`] for what they are measured against and when.
    pub fn set_compaction(&mut self, compaction: CatalogCompaction) {
        self.compaction = compaction;
    }

    pub fn compaction(&self) -> CatalogCompaction {
        self.compaction
    }

    /// Arms this handle's read cache — the bounded LRU of parsed documents [`Table::get`] answers a
    /// repeated read of the same key from. Replaces whatever was cached so far, so re-arming a live
    /// table starts it from cold rather than from documents the old policy admitted.
    pub fn set_cache(&mut self, cache: CachePolicy) {
        self.cache = RecordCache::new(cache);
    }

    pub fn cache_policy(&self) -> CachePolicy {
        self.cache.policy()
    }

    /// Sets how durable this table's appends are — called by `Database` from `options.sync`, left
    /// at `Commit` otherwise. `Commit` `fsync`s the log on every flush; `Never` skips the `fsync`
    /// and leaves the bytes to the kernel, so a crash can lose appends this table already
    /// acknowledged. Nothing else changes: the writes, the registry and the snapshots are the same.
    pub fn set_sync(&mut self, sync: CatalogSyncMode) {
        self.sync = sync;
        self.logs.set_fsync(sync == CatalogSyncMode::Commit);
    }

    pub fn sync_mode(&self) -> CatalogSyncMode {
        self.sync
    }

    /// Sets the keyring `put`/`patch` seal `cypher` fields with and `get`/`query` open them with —
    /// called by `Database` from `options.cypher`, left at `None` otherwise, which is fine for a
    /// table whose schema declares no `cypher` field and an error the moment one tries to write or
    /// read one that does.
    pub fn set_cipher(&mut self, cipher: Option<FieldCipher>) {
        self.cipher = cipher;
        // NOTE: cached documents were opened with the old keyring — what they read as may change.
        self.cache.clear();
    }

    /// Writes the bytes of every `image` field `source` carries a `content` string for, content-
    /// addressed under [`Table::image_dir`] — skipped when a file already sits at that path, since
    /// the same hash means the same bytes (free dedup) and skipped for a field `source` does not
    /// mention (an untouched image field on a patch has nothing new to write). Reads `content` off
    /// `source` directly rather than the coerced record, which no longer carries it past validation.
    fn write_image_fields(&self, source: &Value) -> io::Result<()> {
        let Some(object) = source.as_object() else {
            return Ok(());
        };
        for (name, spec) in &self.schema.fields {
            if spec.field_type != FieldType::Image {
                continue;
            }
            let Some(content) = object
                .get(name)
                .and_then(|value| value.get("content"))
                .and_then(Value::as_str)
            else {
                continue;
            };
            let decoded = crate::schema::decode_image_content(content)
                .map_err(|reason| io::Error::new(io::ErrorKind::InvalidData, reason))?;
            self.write_image_bytes(&decoded)?;
        }
        Ok(())
    }

    /// Every markdown field of `record` scanned against the table's whole virtual-path index —
    /// a link resolves if any markdown/image field (this table's `virtual_path_fields()`) holds
    /// the target path, not only the field the link was found in.
    fn broken_markdown_links(&self, record: &HaysonRecord) -> Vec<BrokenLink> {
        let mut broken = Vec::new();
        for (name, spec) in &self.schema.fields {
            if spec.field_type != FieldType::Markdown {
                continue;
            }
            let Some(content) = record
                .get(name)
                .and_then(|value| value.get("content"))
                .and_then(Value::as_str)
            else {
                continue;
            };
            let field_broken = broken_links(content, |target| {
                self.registry
                    .virtual_path_fields()
                    .any(|field| self.registry.virtual_path_holder(field, target).is_some())
            });
            broken.extend(field_broken.into_iter().map(|target| BrokenLink {
                field: name.clone(),
                path: target,
            }));
        }
        broken
    }

    /// Diagnostics for a just-written `record` — empty unless [`Table::link_parsing`] is `Eager`,
    /// since a `Lazy` table only checks links on demand via [`Table::broken_links_of`].
    fn eager_link_warnings(&self, record: &HaysonRecord) -> Vec<BrokenLink> {
        if self.link_parsing != LinkParsing::Eager {
            return Vec::new();
        }
        self.broken_markdown_links(record)
    }

    /// The diagnostics [`PutOutcome::warnings`] would carry for the record now stored under `key` —
    /// empty unless [`Table::link_parsing`] is `Eager`, exactly as [`Table::put_with_warnings`].
    ///
    /// It exists for a binding that cannot use `put_with_warnings` because it stages and flushes in
    /// two separate calls across an FFI boundary and so never holds a [`PutOutcome`]; asking here
    /// once the write has landed gives the same answer, and keeps the `Eager` gate in one place
    /// rather than copied into every binding.
    pub fn eager_link_warnings_of(&self, key: &str) -> Result<Vec<BrokenLink>, TableError> {
        if self.link_parsing != LinkParsing::Eager {
            return Ok(Vec::new());
        }
        self.broken_links_of(key)
    }

    /// On-demand link check for a `Lazy` table (or any table) — reads `key`'s current record and
    /// reports every markdown link that does not resolve, regardless of [`Table::link_parsing`].
    pub fn broken_links_of(&self, key: &str) -> Result<Vec<BrokenLink>, TableError> {
        let record = self.get(key)?.ok_or_else(|| {
            TableError::NotFound(HaysonNotFoundError {
                table: self.schema.table.clone(),
                key: key.to_string(),
            })
        })?;
        Ok(self.broken_markdown_links(&record))
    }

    fn write_image_bytes(&self, bytes: &[u8]) -> io::Result<()> {
        let file_path = self.image_dir.join(physical_path_of(bytes));
        if file_path.exists() {
            return Ok(());
        }
        fs::create_dir_all(&self.image_dir)?;
        fs::write(file_path, bytes)
    }

    /// Writes `<table>.idx` for the registry's current state and resets the since-last-snapshot
    /// counter — mirrors the TS `Table.snapshot()`. A no-op on a table opened via [`Table::open`],
    /// which carries no `snapshot_path`.
    pub fn snapshot(&mut self) -> io::Result<()> {
        self.flush()?;
        self.write_snapshot_now()
    }

    /// [`Table::snapshot`] without the leading flush — for the callers already inside the write
    /// path, where the queue is drained but the staging window is still open for triggers.
    fn write_snapshot_now(&mut self) -> io::Result<()> {
        let Some(path) = self.snapshot_path.clone() else {
            return Ok(());
        };
        write_snapshot_set(
            &path,
            &self.registry,
            &self.schema.table,
            &self.schema_hash,
            self.gen,
            &self.logs,
        )?;
        self.since_snapshot = 0;
        self.snapshot_current = true;
        Ok(())
    }

    /// Whether this table has writes since its last snapshot that `close()` still needs to flush —
    /// always `false` on a table opened via [`Table::open`].
    pub fn needs_snapshot(&self) -> bool {
        self.snapshot_path.is_some() && !self.snapshot_current
    }

    /// Moves every record of this table onto the cipher's current write key, eagerly — the
    /// counterpart of the lazy rotation a record gets for free on its next organic write.
    ///
    /// A record is read, decrypted with whichever key sealed it and written back under the same
    /// primary key, so the plaintext is preserved and nothing new is created. Records already on
    /// the current key are not rewritten at all, which is what makes running this repeatedly cost
    /// one read per record and nothing else. A table declaring no `cypher` field answers zero
    /// without reading anything.
    ///
    /// A field naming a key the keyring does not hold follows the table's own
    /// [`MissingKeyPolicy`], for the reason a maintenance pass has to: under the default `Throw`
    /// the run is refused at that record, because silently leaving a record behind is exactly what
    /// an operator running a rotation is trying to rule out; under `Omit`/`Raw` it is counted in
    /// [`RotationResult::skipped`] and left byte for byte, since neither of those policies yields
    /// a plaintext that could be re-sealed — `Omit` drops the field and `Raw` hands back the
    /// envelope, and writing either one back would destroy the value.
    ///
    /// Records rotated before a refusal stay rotated: every write lands as a complete record, so a
    /// refused run is a partial one, not a broken one, and re-running it after adding the missing
    /// key finishes the job. No trigger fires — a rotation changes no value a trigger could be
    /// about.
    pub fn rotate_cypher_keys(&mut self) -> Result<RotationResult, TableError> {
        let mut result = RotationResult::default();
        if self.cypher_fields.is_empty() {
            return Ok(result);
        }
        if self.logs.is_readonly() {
            return Err(TableError::Io(io::Error::other(format!(
                "table \"{}\" is readonly — rotating cypher keys is not available",
                self.schema.table
            ))));
        }
        // NOTE: a staged write is not in the log yet, so it would be rotated as the record it replaces.
        self.flush()?;
        let Some(cipher) = self.cipher.clone() else {
            return Err(TableError::Cipher(Box::new(HaysonCipherError::new(
                self.schema.table.clone(),
                "",
                "cannot rotate cypher keys without a key",
            ))));
        };

        let keys: Vec<String> = self.keys().cloned().collect();
        for key in keys {
            let Some(sealed) = self.sealed_record(&key)? else {
                continue;
            };
            match self.rotation_verdict(&sealed, &cipher) {
                RotationVerdict::Current => result.already_current += 1,
                RotationVerdict::Missing(key_id) => match cipher.on_missing_key() {
                    MissingKeyPolicy::Throw => {
                        return Err(TableError::Cipher(Box::new(
                            HaysonCipherError::new(
                                self.schema.table.clone(),
                                "",
                                format!("record \"{key}\" was sealed with key \"{key_id}\", which was not given"),
                            )
                            .with_key_id(&key_id),
                        )))
                    }
                    MissingKeyPolicy::Omit | MissingKeyPolicy::Raw => result.skipped += 1,
                },
                RotationVerdict::Rotate => {
                    let opened = open_record(&self.schema, &key, &sealed, Some(&cipher), &self.cypher_fields)
                        .map_err(|error| TableError::Cipher(Box::new(error)))?;
                    let document = Value::Object(opened.into_iter().collect());
                    self.put_from_buffer(&document, false)?;
                    result.rotated += 1;
                }
            }
        }
        Ok(result)
    }

    /// `key` exactly as the log holds it — `cypher` fields still sealed, because the decision
    /// [`Table::rotate_cypher_keys`] makes is about the envelope, not about the value inside it.
    fn sealed_record(&self, key: &str) -> Result<Option<HaysonRecord>, TableError> {
        let mut source = LogSetRecordSource::new(&self.registry, &self.logs);
        source.load(key).map_err(TableError::from)
    }

    fn rotation_verdict(&self, sealed: &HaysonRecord, cipher: &FieldCipher) -> RotationVerdict {
        let mut rotate = false;
        for field in &self.cypher_fields {
            let Some(value) = sealed.get(field) else { continue };
            if value.is_null() {
                continue;
            }
            match envelope_key_id(value) {
                Some(key_id) if !cipher.knows(key_id) => return RotationVerdict::Missing(key_id.to_string()),
                Some(key_id) => rotate |= key_id != cipher.write_key_id(),
                // A value in the clear: the field was marked `cypher` after this record landed, so
                // rotating it is what seals it for the first time.
                None => rotate = true,
            }
        }
        if rotate {
            RotationVerdict::Rotate
        } else {
            RotationVerdict::Current
        }
    }

    /// Rewrites the log to keep only the live version of each key, then remaps the registry's
    /// slots onto the new offsets — see `compaction::compact_log`. A partitioned table
    /// compacts one partition file at a time and remaps each on its own, so a partition nothing was
    /// written to is left byte for byte as it was; the returned result is their sum.
    ///
    /// The snapshot on disk describes the log as it was before the rewrite, so it is deleted before
    /// the first swap and written afresh once every partition has landed — the port of the TS
    /// `beforeSwap: removeSnapshot(...)` plus the trailing `snapshot()`. Deleting first is what
    /// keeps a crash mid-compaction honest: whichever log survives, no index claims to pair with it.
    pub fn compact(&mut self) -> io::Result<crate::compaction::CompactionResult> {
        // NOTE: a staged write would be rewritten out of existence — it is not in the log yet.
        self.flush()?;
        self.cache.clear();
        let mut total = crate::compaction::CompactionResult {
            slots: HashMap::new(),
            records: 0,
            bytes_before: 0,
            bytes_after: 0,
            reclaimed_bytes: 0,
        };
        let mut swapped = false;
        for part in self.logs.parts() {
            if !swapped {
                if let Some(path) = self.snapshot_path.as_deref() {
                    remove_snapshot(path)?;
                }
                self.snapshot_current = false;
                swapped = true;
            }
            let Some(log) = self.logs.log_of_mut(&part) else {
                continue;
            };
            let result = crate::compaction::compact_log(log)?;
            self.registry.remap(&result.slots, &part);
            total.records += result.records;
            total.bytes_before += result.bytes_before;
            total.bytes_after += result.bytes_after;
            total.reclaimed_bytes += result.reclaimed_bytes;
            total.slots.extend(result.slots);
        }
        if swapped {
            // NOTE: bump before the snapshot -- the file has to carry the generation it describes.
            self.gen += 1;
            self.snapshot()?;
        }
        Ok(total)
    }

    /// Catches up with whatever another process did to this table's files since this handle last
    /// looked — the port of the TS `Table.refresh()`. Three things can have happened to a
    /// partition, and each is answered differently:
    ///
    /// * **nothing** — the file is the same size, on the same inode. One `stat` and no read at all,
    ///   which is what makes calling this on a read path affordable;
    /// * **an append** — the file grew on the same inode, so everything already applied is still at
    ///   the offset it was applied at and only the tail past this handle's watermark is scanned;
    /// * **a rewrite** — a different inode, or bytes gone from a file this engine only ever appends
    ///   to. That is what a foreign [`Table::compact`] leaves behind (it lands through
    ///   `atomic::write_atomic`, so the path gets a brand new file), and it invalidates every
    ///   offset this registry holds. There is no sound way to scan on from a watermark into a file
    ///   that was rewritten under it, so the whole registry is rebuilt from the files as they are.
    ///
    /// A partition file another process created is opened here too (`LogSet::discover`) and read in
    /// full, since none of it has ever been seen.
    pub fn refresh(&mut self) -> io::Result<RefreshOutcome> {
        let appeared = self.logs.discover()?;
        let mut entries = 0;

        for part in self.logs.parts() {
            if appeared.contains(&part) {
                // NOTE: a partition opened just now is unread in full, not merely behind by a tail.
                let Some(log) = self.logs.log_of(&part) else { continue };
                let scan = log.scan(0)?;
                entries += scan.entries.len();
                self.invalidate_scanned(&scan);
                self.registry.apply_scan_in(scan, &part);
                continue;
            }

            let Some(log) = self.logs.log_of_mut(&part) else {
                continue;
            };
            let Some(probe) = log.probe() else { continue };
            if rewritten(log, &probe) {
                self.reload()?;
                return Ok(RefreshOutcome::Reloaded);
            }
            if probe.size == log.size() {
                continue;
            }
            let from = log.size();
            log.adopt_size(probe.size);
            let scan = log.scan(from)?;
            // NOTE: rewind to the last complete line — a foreign append may be only half on disk.
            log.adopt_size(scan.end_offset);
            entries += scan.entries.len();
            self.invalidate_scanned(&scan);
            self.registry.apply_scan_in(scan, &part);
        }

        self.seq = self.seq.max(self.registry.max_seq());
        if entries == 0 {
            return Ok(RefreshOutcome::Unchanged);
        }
        self.snapshot_current = false;
        Ok(RefreshOutcome::CaughtUp { entries })
    }

    /// Drops the cached document of every key a foreign scan carries, before its entries reach the
    /// registry: those keys now point at another writer's line, and the cache holds what this
    /// handle read before that line existed.
    fn invalidate_scanned(&self, scan: &ScanResult) {
        for found in &scan.entries {
            self.cache.invalidate(&found.entry.meta.key);
        }
    }

    /// Sets how often this handle catches up with foreign writes on its own — `None` to follow
    /// nothing, which is where every constructor starts. Called by `Database::open`, which resolves
    /// the policy from `OpenOptions::follow` and `OpenOptions::readonly`.
    pub fn set_follow(&mut self, follow: Option<FollowPolicy>) {
        self.follow = follow;
    }

    pub fn follow(&self) -> Option<FollowPolicy> {
        self.follow
    }

    /// Catches this handle up when a follow policy is set and its interval has elapsed — the port
    /// of the TS `Table.#followIfDue`, and what the `*_following` read methods below run first.
    ///
    /// The interval gates the check, not the catching up: once the gate passes, [`Table::refresh`]
    /// decides whether anything actually happened, and it costs one `stat` per log file to find out
    /// that nothing did. The clock is moved on whenever the gate passes, so a table nobody else is
    /// writing to is probed at most once per interval no matter how many reads run in between.
    ///
    /// @returns what the catch-up found, or `None` when no policy is set or the interval has not
    ///   elapsed yet.
    pub fn follow_if_due(&mut self) -> io::Result<Option<RefreshOutcome>> {
        if !self.follow_is_due() {
            return Ok(None);
        }
        self.followed_at = Some(Instant::now());
        Ok(Some(self.refresh()?))
    }

    /// Whether [`Table::follow_if_due`] would reach for the files if it ran now: a policy is set and
    /// its interval has elapsed. Only the gate — a clock comparison against `followed_at`, no `stat`
    /// and no read — so this is the cheap half of the follow-aware read path.
    ///
    /// It is public because a binding that has to take an exclusive lock to catch up asks this
    /// first, under a shared one, and skips the exclusive lock entirely when the answer is `false`.
    /// A writable handle resolves to no policy at all, so for it the answer is always `false`.
    pub fn follow_is_due(&self) -> bool {
        let Some(policy) = self.follow else { return false };
        policy.interval.is_zero() || self.followed_at.is_none_or(|at| at.elapsed() >= policy.interval)
    }

    /// Whether anything staged is still waiting for [`Table::flush`] — the exact condition that
    /// makes that call more than a no-op, and what a read which cannot see the staging window has to
    /// drain before it can answer. [`Table::query`] and the aggregates plan over the registry rather
    /// than over the staged writes, which is why [`Table::query_following`] flushes and
    /// [`Table::get_following`] does not.
    pub fn has_staged_writes(&self) -> bool {
        self.queue.depth() > 0 || !self.staged.is_empty()
    }

    /// [`Table::get`] behind [`Table::follow_if_due`] — the read path the TS `Table.get` is, where
    /// the freshness check is part of the read rather than something a caller remembers to do.
    ///
    /// It is a second method rather than the behaviour of `get` itself because catching up needs
    /// `&mut self` and `get` deliberately does not: every read that runs inside an already-borrowed
    /// database (integrity checks, unit-of-work reads, `drop_field`'s residue scan) goes through the
    /// shared-borrow one, and none of those wants a foreign write landing underneath it mid-pass.
    pub fn get_following(&mut self, key: &str) -> Result<Option<HaysonRecord>, TableError> {
        self.follow_if_due()?;
        self.get(key)
    }

    /// [`Table::get_following`] for several keys, checking freshness once for the whole batch —
    /// the TS `Table.getMany`, which calls `#followIfDue` once and then reads key by key.
    pub fn get_many_following(&mut self, keys: &[String]) -> Result<Vec<Option<HaysonRecord>>, TableError> {
        self.follow_if_due()?;
        keys.iter().map(|key| self.get(key)).collect()
    }

    /// [`Table::query`] behind [`Table::follow_if_due`] — see [`Table::get_following`] for why the
    /// follow-aware read is a separate method.
    pub fn query_following(&mut self, spec: &QuerySpec) -> Result<QueryResult, TableError> {
        self.flush()?;
        self.follow_if_due()?;
        self.query(spec)
    }

    /// [`Table::aggregate`] behind [`Table::follow_if_due`].
    pub fn aggregate_following(
        &mut self,
        spec: &QuerySpec,
        aggregate: &AggregateSpec,
    ) -> Result<AggregateResult, TableError> {
        self.flush()?;
        self.follow_if_due()?;
        self.aggregate(spec, aggregate)
    }

    /// [`Table::aggregate_grouped`] behind [`Table::follow_if_due`].
    pub fn aggregate_grouped_following(
        &mut self,
        spec: &QuerySpec,
        aggregate: &AggregateSpec,
    ) -> Result<GroupedAggregateResult, TableError> {
        self.flush()?;
        self.follow_if_due()?;
        self.aggregate_grouped(spec, aggregate)
    }

    /// Throws this handle's registry away and builds it again from every log file, each reopened
    /// first so the descriptor points at the file now at that path rather than the one that was
    /// there when this handle opened. Used only by [`Table::refresh`], for the rewrite case.
    fn reload(&mut self) -> io::Result<()> {
        self.logs.discover()?;
        let mut registry = TableRegistry::with_declared_indexes(self.schema.clone(), &self.declared);
        for part in self.logs.parts() {
            let Some(log) = self.logs.log_of_mut(&part) else {
                continue;
            };
            log.reopen()?;
            let scan = log.scan(0)?;
            // NOTE: a half-written tail belongs to the writer that left it — only its end is taken.
            log.adopt_size(scan.end_offset);
            registry.apply_scan_in(scan, &part);
        }
        self.registry = registry;
        self.cache.clear();
        self.seq = self.seq.max(self.registry.max_seq());
        self.gen += 1;
        self.snapshot_current = false;
        Ok(())
    }

    /// Every key the table currently holds, in no particular order.
    pub fn keys(&self) -> impl Iterator<Item = &String> {
        self.registry.keys()
    }

    /// Whether the table holds a record under `key`, counting a write staged but not yet flushed —
    /// the TS `Table.has`, which answers from the write queue before it looks at the log.
    pub fn has(&self, key: &str) -> bool {
        match self.staged.get(key) {
            Some(staged) => staged.record.is_some(),
            None => self.registry.slot_of(key).is_some(),
        }
    }

    /// Whether any live, flushed record holds `value` on `field`. `field` has to be this table's
    /// primary key or a `unique` field — every caller reaches this through a relation's `to.field`,
    /// which the catalog's relation parser already restricts to one of those two (ADR-axje/0043) —
    /// anything else always answers `false`. O(1) via the registry: the primary-key case is
    /// [`Table::has`], the unique-field case is the same lookup `assert_unique` runs for the
    /// opposite question.
    ///
    /// NOTE: registry-only, unlike `has()` — a record staged but not yet flushed on a unique field
    /// is invisible here, matching the eventual-consistency window the write queue already has.
    pub fn contains_value(&self, field: &str, value: &Value) -> bool {
        if field == self.schema.primary_key {
            return match value {
                Value::String(s) => self.has(s),
                Value::Number(n) => self.has(&n.to_string()),
                _ => false,
            };
        }
        self.registry.unique_holder(field, value).is_some()
    }

    /// How many live records the table holds — the TS `Table.size`. Tombstoned keys and superseded
    /// versions are not among them; they are bytes in the log, not records in the registry.
    ///
    /// Staged writes are deliberately not counted, exactly as in the TS reference: this is the
    /// registry's own number, and a key first written in the current staging window joins it when
    /// the write lands. [`Table::keys`] leaves them out for the same reason.
    pub fn len(&self) -> usize {
        self.registry.records()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// The fields claimed by writes staged against `key` but not yet appended — the TS
    /// `Table.pending(key)`. Empty unless a caller staged through [`Table::stage_put`] and friends
    /// without flushing yet: `put`/`patch`/`delete` stage and flush inside the one call.
    pub fn pending(&self, key: &str) -> HashSet<String> {
        self.queue.pending(key)
    }

    /// The table's counters, in the shape of the TS `TableStats` — see [`TableStats`] for the
    /// member whose meaning is narrower here than there.
    pub fn stats(&self) -> TableStats {
        let stats = self.registry.stats();
        TableStats {
            records: stats.records,
            live_bytes: stats.live_bytes,
            dead_bytes: stats.dead_bytes,
            log_bytes: self.logs.total_size(),
            queue_depth: self.queue.depth(),
            gen: self.gen,
            seq: self.seq,
        }
    }

    fn note_writes(&mut self, count: usize) -> io::Result<()> {
        if count == 0 {
            return Ok(());
        }
        self.since_snapshot += count as u64;
        self.snapshot_current = false;
        self.compact_if_due()?;
        if self.snapshot_every > 0 && self.since_snapshot >= self.snapshot_every {
            self.write_snapshot_now()?;
        }
        Ok(())
    }

    /// Rewrites every partition whose garbage has grown past the policy [`Table::set_compaction`]
    /// armed — the port of the TS `Table.#maybeCompact`, run after each batch of writes lands.
    ///
    /// Both thresholds have to be met, and both are read straight off the registry's per-partition
    /// byte tallies (`TableRegistry::stats_of`), which every put and delete already maintains: this
    /// runs on every write, so it may not cost a single `stat` of its own. `min_dead_bytes` is the
    /// absolute floor that keeps a small log from being rewritten over and over, `dead_bytes_ratio`
    /// the share of the partition that has to be garbage before rewriting it is worth the I/O.
    ///
    /// A partitioned table is measured one partition at a time, so a partition nobody overwrote is
    /// left alone; the generation counter still moves once for the whole pass, not once per file,
    /// because that is what readers compare to notice the offsets moved under them.
    fn compact_if_due(&mut self) -> io::Result<()> {
        let CatalogCompaction::Enabled {
            dead_bytes_ratio,
            min_dead_bytes,
        } = self.compaction
        else {
            return Ok(());
        };
        let mut compacted = false;
        for part in self.logs.parts() {
            let bytes = self.registry.stats_of(&part);
            let total = bytes.live_bytes + bytes.dead_bytes;
            if total == 0 || bytes.dead_bytes < min_dead_bytes {
                continue;
            }
            if (bytes.dead_bytes as f64) / (total as f64) < dead_bytes_ratio {
                continue;
            }
            let Some(log) = self.logs.log_of_mut(&part) else {
                continue;
            };
            if !compacted {
                // NOTE: the snapshot on disk pairs with the log about to be replaced -- drop it first.
                if let Some(path) = self.snapshot_path.as_deref() {
                    remove_snapshot(path)?;
                }
                self.snapshot_current = false;
            }
            let result = crate::compaction::compact_log(log)?;
            self.registry.remap(&result.slots, &part);
            compacted = true;
        }
        if !compacted {
            return Ok(());
        }
        self.cache.clear();
        // NOTE: bump before the snapshot -- the file has to carry the generation it describes.
        self.gen += 1;
        self.write_snapshot_now()
    }

    pub fn name(&self) -> &str {
        &self.schema.table
    }

    pub fn schema(&self) -> &TableSchema {
        &self.schema
    }

    /// The expanded index set this handle was opened with and has served ever since — the answer
    /// to "what does this table actually index", as opposed to what the catalog on disk now says.
    /// A reload rebuilds its registry from exactly this list, so it stays true across `refresh`.
    pub fn declared_indexes(&self) -> &[ExpandedIndex] {
        &self.declared
    }

    /// Reads `key`, answering from the staging window first — the TS "read your own pending write".
    /// A staged record is handed back as it was validated, in plaintext, so a `cypher` field reads
    /// the same before and after the write that carries it lands.
    pub fn get(&self, key: &str) -> Result<Option<HaysonRecord>, TableError> {
        match self.staged.get(key) {
            Some(staged) => Ok(staged.record.clone()),
            None => self.committed(key),
        }
    }

    /// `key` as the log holds it, ignoring anything staged — what a staging window needs to record
    /// as the `old` a trigger will be fired with, and the only read that must not see itself.
    fn committed(&self, key: &str) -> Result<Option<HaysonRecord>, TableError> {
        if let Some(cached) = self.cache.get(key) {
            return Ok(Some(cached));
        }
        let found = self.with_record_source(|source| source.load(key))?;
        if let Some(record) = &found {
            self.cache.insert(key, record);
        }
        Ok(found)
    }

    /// Builds the [`LogRecordSource`] every read path (`get`, `query`, `aggregate`,
    /// `aggregate_grouped`) shares, wired to decrypt `cypher` fields and to project the record onto
    /// the live schema the moment it is parsed off the log — the single chokepoint that keeps
    /// `get`/`query`/`patch`'s read-merge all hydrating through the same hook, mirroring the TS
    /// `Table.#read`/`#open`.
    ///
    /// The hook now runs for every table, not only one with a `cypher` field: the log is append-only,
    /// so a line written before a `dropField` still carries the dropped field, and the schema is the
    /// only thing that says what a record has. See `packages/hayson/test/drop-field.test.ts`.
    fn with_record_source<T>(
        &self,
        f: impl FnOnce(&mut dyn RecordSource) -> Result<T, RecordSourceError>,
    ) -> Result<T, TableError> {
        let open = |key: &str, record: HaysonRecord| -> Result<HaysonRecord, RecordSourceError> {
            let opened = if self.cypher_fields.is_empty() {
                record
            } else {
                open_record(&self.schema, key, &record, self.cipher.as_ref(), &self.cypher_fields)?
            };
            Ok(project_onto_schema(&self.schema, opened))
        };
        let mut source = LogSetRecordSource::with_open(&self.registry, &self.logs, &open);
        f(&mut source).map_err(TableError::from)
    }

    /// Every log line naming `key`, oldest first — what [`Table::get`] answers with is the last of
    /// them.
    ///
    /// Reads the log files rather than the registry, because the registry holds exactly one slot per
    /// key (the live one) and nothing that points at a version already overwritten: the cost is a
    /// full scan of this table's logs, every partition of a partitioned table. Nothing here is
    /// hydrated the way a read is — see [`HistoryEntry::document`].
    ///
    /// Writes staged but not flushed are not on disk yet, so they are not in the answer either;
    /// [`Table::flush`] first if they should be.
    pub fn history(&self, key: &str) -> io::Result<TableHistory> {
        let mut entries = Vec::new();
        let mut scanned = Vec::new();
        for part in self.logs.parts() {
            let Some(log) = self.logs.log_of(&part) else { continue };
            scanned.push(log.path().to_path_buf());
            for found in log.scan(0)?.entries {
                if found.entry.meta.key != key {
                    continue;
                }
                entries.push(HistoryEntry {
                    seq: found.entry.meta.seq,
                    ts: found.entry.meta.ts,
                    op: found.entry.meta.op,
                    partition: part.clone(),
                    offset: found.slot.offset,
                    length: found.slot.length,
                    document: found.entry.data,
                });
            }
        }
        entries.sort_by_key(|entry| entry.seq);
        Ok(TableHistory { entries, scanned })
    }

    pub fn put(&mut self, document: &Value, force: bool) -> Result<HaysonRecord, TableError> {
        self.put_impl(document, force, true).map(PutOutcome::into_record)
    }

    /// [`Table::put`] for a record whose `cypher` fields named in `sealed` already hold an envelope
    /// this system produced — a backup being restored, or a copy moving a value between databases
    /// without ever holding it in the clear. Those fields reach the log exactly as given; every
    /// other field, `cypher` or not, takes the ordinary write path and is sealed.
    ///
    /// This is the **only** way to store a `cypher` field without sealing it, and it is deliberately
    /// not reachable from a document: an ordinary write seals a value that merely looks like an
    /// envelope just like any other, which is the whole point of the entry point existing. Naming a
    /// field here is a claim, checked as far as the keyring allows — see
    /// [`crate::crypto::assert_carried_seals`] for what is refused and for the one case that is
    /// taken on trust. Do not wire it to untrusted input: nothing below it can tell a caller's
    /// string from a genuine envelope once the key that would authenticate it is absent.
    ///
    /// The record returned is the one that was stored, so a carried field reads back as its
    /// envelope rather than as plaintext — unlike [`Table::put`], which is handed plaintext to
    /// begin with. A later [`Table::get`] opens it like any other stored value.
    pub fn put_sealed(&mut self, document: &Value, sealed: &[String], force: bool) -> Result<HaysonRecord, TableError> {
        if sealed.is_empty() {
            return self.put(document, force);
        }
        let accepted = self.accepted_document(document);
        let document = accepted.as_ref();
        let candidate = validate_record(&self.schema, document, ValidateOptions::default())?;
        let key = primary_key_of(&self.schema, &candidate)?;
        for field in sealed {
            if !self.cypher_fields.iter().any(|name| name == field) {
                return Err(TableError::Cipher(Box::new(HaysonCipherError::new(
                    self.schema.table.clone(),
                    field.clone(),
                    "cannot be carried over as already sealed — it is not declared \"cypher\"",
                ))));
            }
        }
        // NOTE: refuse before staging — a refusal from inside the commit would leave a claim held.
        assert_carried_seals(&self.schema, &key, &candidate, self.cipher.as_ref(), sealed)?;

        self.assert_partition_stable_against_staged(document)?;
        let key = self
            .queue
            .stage_put_carrying(&self.schema, &self.registry, document, sealed.to_vec(), force)?;
        let Some(record) = self.queue.last_put_record().cloned() else {
            unreachable!("stage_put_carrying queues exactly one put entry")
        };
        self.write_image_fields(document)?;
        self.stage(key, Some(record.clone()), true)?;
        self.flush()?;
        Ok(record)
    }

    /// Same as [`Table::put`], but the outcome also carries diagnostics — currently only broken
    /// markdown links found while [`Table::link_parsing`] is `Eager`. See [`PutOutcome`] for the
    /// shape and why `put` stays untouched alongside it.
    pub fn put_with_warnings(&mut self, document: &Value, force: bool) -> Result<PutOutcome, TableError> {
        self.put_impl(document, force, true)
    }

    /// Same as [`Table::put`] but never fires triggers — used only by
    /// `buffer::apply_clean_entries`, which replays an entry whose originating live write already
    /// fired them once; firing again here would double the effect.
    pub(crate) fn put_from_buffer(&mut self, document: &Value, force: bool) -> Result<HaysonRecord, TableError> {
        self.put_impl(document, force, false).map(PutOutcome::into_record)
    }

    fn put_impl(&mut self, document: &Value, force: bool, live: bool) -> Result<PutOutcome, TableError> {
        let record = self.stage_put_impl(document, force, live)?;
        self.flush()?;
        let warnings = self.eager_link_warnings(&record);
        Ok(PutOutcome { record, warnings })
    }

    /// The staging half of [`Table::put`]: validates, claims the record's fields against everything
    /// else in flight and queues the write — but appends nothing and fires no trigger, both of
    /// which belong to the [`Table::flush`] that commits it.
    ///
    /// The record is returned as it will be stored, defaults and derived fields included, so a
    /// caller that answers its own user before the write is durable has the same answer `put` would
    /// have given. Until the flush, [`Table::get`] and [`Table::has`] read it back and
    /// [`Table::pending`] names the fields it claimed.
    pub fn stage_put(&mut self, document: &Value, force: bool) -> Result<HaysonRecord, TableError> {
        self.stage_put_impl(document, force, true)
    }

    fn stage_put_impl(&mut self, document: &Value, force: bool, live: bool) -> Result<HaysonRecord, TableError> {
        let accepted = self.accepted_document(document);
        let document = accepted.as_ref();
        self.assert_partition_stable_against_staged(document)?;
        let key = self.queue.stage_put(&self.schema, &self.registry, document, force)?;
        let Some(record) = self.queue.last_put_record().cloned() else {
            unreachable!("stage_put queues exactly one put entry")
        };
        self.write_image_fields(document)?;
        self.stage(key, Some(record.clone()), live)?;
        Ok(record)
    }

    /// Records one staged write against `key`, carrying over the `old` and the position of an
    /// earlier write to the same key in this same window.
    fn stage(&mut self, key: String, record: Option<HaysonRecord>, live: bool) -> io::Result<()> {
        let (id, old) = match self.staged.get(&key) {
            Some(held) => (held.id, held.old.clone()),
            None => {
                self.next_op_id += 1;
                (self.next_op_id, self.committed(&key)?)
            }
        };
        self.staged.insert(key, StagedWrite { id, record, old, live });
        Ok(())
    }

    /// Refuses a staged write that would move a key out of the partition another staged write just
    /// put it in — the registry-backed half of this rule lives in `WriteQueue::stage_put`, which
    /// cannot see a partition that only exists in the staging window.
    ///
    /// Costs a second `validate_record` for the write it inspects, so it only runs where it can
    /// possibly refuse anything: a partitioned table with a non-empty window.
    fn assert_partition_stable_against_staged(&self, document: &Value) -> Result<(), TableError> {
        if self.staged.is_empty() || !is_partitioned(&self.schema) {
            return Ok(());
        }
        let record = validate_record(&self.schema, document, ValidateOptions::default())?;
        let key = primary_key_of(&self.schema, &record)?;
        let Some(StagedWrite { record: Some(held), .. }) = self.staged.get(&key) else {
            return Ok(());
        };
        let held_part = partition_of(&self.schema, held)?;
        let part = partition_of(&self.schema, &record)?;
        if held_part == part {
            return Ok(());
        }
        Err(TableError::Partition(HaysonPartitionError::new(
            self.schema.table.clone(),
            self.schema.partition_by.clone().unwrap_or_default(),
            format!(
                "key \"{key}\" already lives in partition \"{held_part}\" — delete it before writing it into \"{part}\""
            ),
        )))
    }

    /// Appends every write staged since the last flush, in staging order, then fires the triggers
    /// they earned — the port of the TS `WriteQueue.drain()`. A no-op on an empty window, so
    /// calling it before a read of committed state costs nothing when there is nothing in flight.
    ///
    /// Triggers fire once per key, with the record the key held when the window opened as their
    /// `old`: a key written twice before a flush is one change as far as an effect is concerned.
    pub fn flush(&mut self) -> Result<(), TableError> {
        if self.queue.depth() == 0 && self.staged.is_empty() {
            return Ok(());
        }
        self.commit()?;

        let mut window: Vec<StagedWrite> = self.staged.drain().map(|(_, staged)| staged).collect();
        window.sort_by_key(|staged| staged.id);
        for staged in window {
            if !staged.live || (staged.record.is_none() && staged.old.is_none()) {
                continue;
            }
            let event = match (&staged.old, &staged.record) {
                (_, None) => TriggerEvent::Delete,
                (Some(_), Some(_)) => TriggerEvent::Update,
                (None, Some(_)) => TriggerEvent::Insert,
            };
            self.fire_same_table_triggers(event, staged.old.as_ref(), staged.record.as_ref())?;
        }
        Ok(())
    }

    /// Read-merge-stage: reads the current record, validates `changes` as a partial record, merges
    /// it over the current one, then stages the merged document as a full `put` — mirrors the TS
    /// `Table.patch`, including the "primary key cannot be changed by patch" rule.
    pub fn patch(&mut self, key: &str, changes: &Value, force: bool) -> Result<HaysonRecord, TableError> {
        self.patch_impl(key, changes, force, true).map(PutOutcome::into_record)
    }

    /// Same as [`Table::patch`], but the outcome also carries diagnostics — see
    /// [`Table::put_with_warnings`] and [`PutOutcome`].
    pub fn patch_with_warnings(&mut self, key: &str, changes: &Value, force: bool) -> Result<PutOutcome, TableError> {
        self.patch_impl(key, changes, force, true)
    }

    /// See [`Table::put_from_buffer`] — same reasoning, for a buffered patch.
    pub(crate) fn patch_from_buffer(
        &mut self,
        key: &str,
        changes: &Value,
        force: bool,
    ) -> Result<HaysonRecord, TableError> {
        self.patch_impl(key, changes, force, false).map(PutOutcome::into_record)
    }

    fn patch_impl(&mut self, key: &str, changes: &Value, force: bool, live: bool) -> Result<PutOutcome, TableError> {
        let record = self.stage_patch_impl(key, changes, force, live)?;
        self.flush()?;
        let warnings = self.eager_link_warnings(&record);
        Ok(PutOutcome { record, warnings })
    }

    /// The staging half of [`Table::patch`] — see [`Table::stage_put`] for what staging does and
    /// does not do. The record it merges over is the staged one when this key already has a write
    /// in flight, so two patches in one window compose instead of the later one dropping the
    /// earlier.
    ///
    /// Only the fields `changes` names are claimed, not the whole merged record: a patch of `age`
    /// must not refuse a patch of `bio` on the same key.
    pub fn stage_patch(&mut self, key: &str, changes: &Value, force: bool) -> Result<HaysonRecord, TableError> {
        self.stage_patch_impl(key, changes, force, true)
    }

    fn stage_patch_impl(
        &mut self,
        key: &str,
        changes: &Value,
        force: bool,
        live: bool,
    ) -> Result<HaysonRecord, TableError> {
        let accepted = self.accepted_document(changes);
        let changes = accepted.as_ref();
        let (_, merged) = self.merged_patch(key, changes)?;
        let claim: Vec<String> = match changes.as_object() {
            Some(object) => object.keys().cloned().collect(),
            None => Vec::new(),
        };
        let document = Value::Object(merged.into_iter().collect());
        self.assert_partition_stable_against_staged(&document)?;
        let key = self
            .queue
            .stage_put_claiming(&self.schema, &self.registry, &document, &claim, force)?;
        let Some(record) = self.queue.last_put_record().cloned() else {
            unreachable!("stage_put_claiming queues exactly one put entry")
        };
        self.write_image_fields(changes)?;
        self.stage(key, Some(record.clone()), live)?;
        Ok(record)
    }

    fn merged_patch(&self, key: &str, changes: &Value) -> Result<(HaysonRecord, HaysonRecord), TableError> {
        let base = self.get(key)?;
        self.merged_patch_over(key, base, changes)
    }

    /// What a key holds as an operation in a unit of work is about to touch it: what the unit's own
    /// earlier operations left behind when they touched it, and the committed record otherwise.
    /// `Some(None)` in the shadow is a delete the unit staged earlier — absent, not "unshadowed".
    fn shadowed_state(&self, shadow: &UnitShadow, key: &str) -> Result<Option<HaysonRecord>, TableError> {
        match shadow.get(key) {
            Some(held) => Ok(held.clone()),
            None => Ok(self.get(key)?),
        }
    }

    /// The merge half of a patch, over a caller-chosen base rather than the committed record. A unit
    /// of work merges over what its own earlier operations left behind, so a key put and then
    /// patched inside one unit composes instead of being refused as absent.
    fn merged_patch_over(
        &self,
        key: &str,
        base: Option<HaysonRecord>,
        changes: &Value,
    ) -> Result<(HaysonRecord, HaysonRecord), TableError> {
        let current = base.ok_or_else(|| {
            TableError::NotFound(HaysonNotFoundError {
                table: self.schema.table.clone(),
                key: key.to_string(),
            })
        })?;
        let partial = validate_record(
            &self.schema,
            changes,
            ValidateOptions {
                partial: true,
                now: None,
            },
        )?;
        if let Some(new_key) = partial.get(&self.schema.primary_key) {
            if value_as_key(new_key) != key {
                return Err(TableError::Schema(HaysonSchemaError(format!(
                    "the primary key of \"{key}\" cannot be changed by patch"
                ))));
            }
        }

        let mut merged = current.clone();
        for (field, value) in &partial {
            merged.insert(field.clone(), value.clone());
        }
        Ok((current, merged))
    }

    pub fn delete(&mut self, key: &str, force: bool) -> Result<bool, TableError> {
        self.delete_impl(key, force, true)
    }

    /// See [`Table::put_from_buffer`] — same reasoning, for a buffered delete.
    pub(crate) fn delete_from_buffer(&mut self, key: &str, force: bool) -> Result<bool, TableError> {
        self.delete_impl(key, force, false)
    }

    fn delete_impl(&mut self, key: &str, force: bool, live: bool) -> Result<bool, TableError> {
        let existed = self.stage_delete_impl(key, force, live)?;
        self.flush()?;
        Ok(existed)
    }

    /// The staging half of [`Table::delete`] — see [`Table::stage_put`]. Answers whether a record
    /// was there to remove straight away, counting one staged in this same window, so the answer
    /// does not have to wait for the tombstone to be appended.
    pub fn stage_delete(&mut self, key: &str, force: bool) -> Result<bool, TableError> {
        self.stage_delete_impl(key, force, true)
    }

    fn stage_delete_impl(&mut self, key: &str, force: bool, live: bool) -> Result<bool, TableError> {
        let old = self.get(key)?;
        let part = self.partition_holding(key)?;
        // NOTE: an unknown key names no partition file, so its tombstone would have nowhere to land.
        if part.is_none() && is_partitioned(&self.schema) {
            return Ok(false);
        }
        self.queue
            .stage_delete(key, force, part.as_deref().unwrap_or(UNPARTITIONED))?;
        self.stage(key.to_string(), None, live)?;
        Ok(old.is_some())
    }

    /// The partition file `key` lives in — read off the staging window first, since a key first
    /// written there has no registry entry yet to name one.
    fn partition_holding(&self, key: &str) -> Result<Option<String>, TableError> {
        match self.staged.get(key) {
            Some(StagedWrite {
                record: Some(record), ..
            }) => Ok(Some(partition_of(&self.schema, record)?)),
            Some(_) => Ok(None),
            None => Ok(self.registry.part_of(key).map(str::to_string)),
        }
    }

    /// Runs every declared trigger whose event matches and whose effect targets this same table —
    /// a cross-table effect only fires through [`crate::txn::UnitOfWork::commit`], which is the
    /// only caller holding every table of the database at once. The effect write itself never
    /// fires triggers (v1 has no chaining, see `catalog::TriggerEntry`).
    fn fire_same_table_triggers(
        &mut self,
        event: TriggerEvent,
        old: Option<&HaysonRecord>,
        new: Option<&HaysonRecord>,
    ) -> Result<(), TableError> {
        let triggers = self.triggers.clone();
        let firing = TriggerFiring { event, old, new };
        for trigger in matching(&triggers, &self.schema.table, event) {
            if trigger.effect.table != self.schema.table {
                continue;
            }
            let document = effect_document(&trigger.effect, &firing);
            self.write_effect(&document)?;
        }
        Ok(())
    }

    fn write_effect(&mut self, document: &Value) -> Result<(), TableError> {
        self.queue.stage_put(&self.schema, &self.registry, document, true)?;
        self.commit()?;
        Ok(())
    }

    /// Refuses a `search` against a table no field of which is marked `fulltext`: the index that
    /// would answer it does not exist, so an empty answer would be a lie rather than a result.
    fn assert_searchable(&self, spec: &QuerySpec) -> Result<(), TableError> {
        if spec.search.is_none() || self.registry.has_fulltext() {
            return Ok(());
        }
        Err(TableError::Schema(HaysonSchemaError(format!(
            "table \"{}\" has no field marked \"fulltext\": true — search is not available",
            self.schema.table
        ))))
    }

    /// Runs `spec` against the committed log. Writes still in the staging window are **not** part
    /// of the answer: flushing needs `&mut self`, and every read that runs inside an already
    /// borrowed database (integrity checks, unit-of-work reads, `drop_field`'s residue scan) goes
    /// through this one. [`Table::query_following`] — what the bindings call — flushes first, and
    /// so is the one a caller staging its own writes wants.
    pub fn query(&self, spec: &QuerySpec) -> Result<QueryResult, TableError> {
        self.assert_searchable(spec)?;
        self.with_record_source(|source| execute_query(&self.registry, source, spec))
    }

    /// [`Table::query`] as a unit of work will leave the table: `overlay` supplies the staged
    /// records and the keys staged for deletion, and `delta` the same writes expressed as a corpus
    /// change, so a `search` is scored over the corpus the unit will leave rather than the
    /// committed one. The plan carries a `PlanOverlay` saying what the staged half contributed.
    pub fn query_with_overlay(
        &self,
        spec: &QuerySpec,
        overlay: &dyn QueryOverlay,
        delta: &CorpusDelta,
    ) -> Result<QueryResult, TableError> {
        self.assert_searchable(spec)?;
        let scores = spec
            .search
            .as_ref()
            .filter(|_| self.registry.has_fulltext())
            .map(|search| self.registry.fulltext().search(&search.text, search.mode, Some(delta)));
        let access = plan_access_overlaid(&self.registry, spec, scores, overlay);
        self.with_record_source(|source| execute_access(&self.registry, source, spec, access, Some(overlay)))
    }

    /// The single-field `exact`/`sorted` fields of this table — what a unit of work's overlay
    /// mirrors so an equality on one of them does not have to check every record it staged.
    pub fn indexed_fields(&self) -> Vec<String> {
        self.registry.indexed_fields()
    }

    pub fn has_fulltext(&self) -> bool {
        self.registry.has_fulltext()
    }

    /// The full-text tokens `record` would be indexed under, asked of a record the registry has not
    /// been told about yet — a unit of work's staged write, on its way into a [`CorpusDelta`].
    pub fn tokens_of(&self, record: &HaysonRecord) -> Vec<String> {
        self.registry.tokens_of(record)
    }

    pub fn aggregate(&self, spec: &QuerySpec, aggregate: &AggregateSpec) -> Result<AggregateResult, TableError> {
        self.assert_searchable(spec)?;
        self.with_record_source(|source| execute_aggregate(&self.registry, source, spec, aggregate))
    }

    /// Requires `aggregate.group_by` to be set — panics via `execute_grouped_aggregate` otherwise.
    pub fn aggregate_grouped(
        &self,
        spec: &QuerySpec,
        aggregate: &AggregateSpec,
    ) -> Result<GroupedAggregateResult, TableError> {
        self.assert_searchable(spec)?;
        self.with_record_source(|source| execute_grouped_aggregate(&self.registry, source, spec, aggregate))
    }

    fn commit(&mut self) -> Result<Vec<WriteOutcome>, TableError> {
        // NOTE: every put/patch/delete lands here — dropping the cached document belongs with it.
        for key in self.queue.queued_keys() {
            self.cache.invalidate(key);
        }
        let schema = &self.schema;
        let cipher = self.cipher.as_ref();
        let cypher_fields = &self.cypher_fields;
        let seal = |key: &str, record: &HaysonRecord, carried: &[String]| -> Result<HaysonRecord, HaysonCipherError> {
            seal_record(schema, key, record, cipher, cypher_fields, carried)
        };
        let outcomes = self
            .queue
            .commit_sealed(&mut self.logs, &mut self.registry, &mut self.seq, true, &seal)?;
        self.note_writes(outcomes.len())?;
        Ok(outcomes)
    }

    /// Highest `_seq` this table has reserved — after `Table::open`, this is the highest `_seq`
    /// actually durable in the log; a live table also counts `_seq` values reserved but not yet
    /// applied by [`Table::prepare_txn_ops`].
    pub fn current_seq(&self) -> u64 {
        self.seq
    }

    /// Step 3 ("prepare") of a multi-table unit of work: validates every op against this table's
    /// schema and unique constraints, reserves a `_seq` and serializes a log line for each — but
    /// appends nothing. Bypasses `WriteQueue`/`PendingMap` entirely (unlike `put`/`patch`/`delete`):
    /// the caller's exclusive `&mut Table` for the lifetime of the unit already rules out any other
    /// writer, so there is nothing left for a pending-claim to guard against.
    ///
    /// A rejected op aborts the whole call with nothing appended — but `_seq` values already
    /// reserved for earlier ops in this call stay burned, same as the TS reference.
    pub fn prepare_txn_ops(&mut self, ops: &[TxnOp]) -> Result<Vec<PreparedTxnEntry>, TableError> {
        // NOTE: a unit of work reserves _seq itself, so nothing may still be waiting to take one.
        self.flush()?;
        let mut prepared = Vec::with_capacity(ops.len());
        let now = chrono::Utc::now().timestamp_millis();
        // NOTE: a key put earlier in this same unit has no registry entry yet to name its partition.
        let mut staged_parts: HashMap<String, String> = HashMap::new();
        // NOTE: unique is judged against the ops prepared earlier in this same call, not just the log.
        let mut shadow = UnitShadow::new();
        for op in ops {
            let (key, old, record) = match op {
                TxnOp::Delete(key) => (key.clone(), self.shadowed_state(&shadow, key)?, None),
                TxnOp::Put(document) => {
                    let accepted = self.accepted_document(document);
                    let record = validate_record(&self.schema, accepted.as_ref(), ValidateOptions::default())?;
                    let key = primary_key_of(&self.schema, &record)?;
                    assert_unique_in_unit(&self.schema, &self.registry, &key, &record, &shadow)?;
                    let old = self.shadowed_state(&shadow, &key)?;
                    (key, old, Some(record))
                }
                TxnOp::Patch(key, changes) => {
                    let accepted = self.accepted_document(changes);
                    let base = self.shadowed_state(&shadow, key)?;
                    let (old, merged) = self.merged_patch_over(key, base, accepted.as_ref())?;
                    assert_unique_in_unit(&self.schema, &self.registry, key, &merged, &shadow)?;
                    (key.clone(), Some(old), Some(merged))
                }
            };

            let part = match &record {
                Some(record) => {
                    let part = partition_of(&self.schema, record)?;
                    assert_partition_stable(&self.schema, &self.registry, &key, &part)?;
                    if let Some(held) = staged_parts.get(&key) {
                        if held != &part {
                            return Err(TableError::Partition(HaysonPartitionError::new(
                                self.schema.table.clone(),
                                self.schema.partition_by.clone().unwrap_or_default(),
                                format!("key \"{key}\" already lives in partition \"{held}\" — delete it before writing it into \"{part}\""),
                            )));
                        }
                    }
                    staged_parts.insert(key.clone(), part.clone());
                    part
                }
                None => {
                    let held = staged_parts
                        .get(&key)
                        .cloned()
                        .or_else(|| self.registry.part_of(&key).map(str::to_string));
                    match held {
                        Some(part) => part,
                        // NOTE: an unknown key names no partition file, so its tombstone is dropped.
                        None if is_partitioned(&self.schema) => continue,
                        None => UNPARTITIONED.to_string(),
                    }
                }
            };

            shadow.insert(key.clone(), record.clone());
            self.seq += 1;
            let meta = LogMeta {
                key: key.clone(),
                op: if record.is_some() { LogOp::Put } else { LogOp::Del },
                seq: self.seq,
                ts: now,
            };
            // NOTE: a unit of work carries no already-sealed value — every op of one comes from a document.
            let sealed = record
                .as_ref()
                .map(|rec| seal_record(&self.schema, &key, rec, self.cipher.as_ref(), &self.cypher_fields, &[]))
                .transpose()?;
            let line = serialize_entry(&meta, sealed.as_ref());
            prepared.push(PreparedTxnEntry {
                key,
                seq: self.seq,
                line,
                old,
                record,
                part,
            });
        }
        Ok(prepared)
    }

    /// Step 5 ("write") of a multi-table unit of work: appends the lines a prior
    /// `prepare_txn_ops` call already serialized, in order, in one `append` per partition they
    /// touch — then `fsync`s and updates the registry so the table reflects them immediately.
    pub fn apply_txn_entries(&mut self, entries: Vec<PreparedTxnEntry>) -> io::Result<()> {
        if entries.is_empty() {
            return Ok(());
        }
        let applied = entries.len();
        let mut slots: Vec<Option<Slot>> = vec![None; entries.len()];
        for part in ordered_parts(&entries) {
            let positions: Vec<usize> = entries
                .iter()
                .enumerate()
                .filter(|(_, entry)| entry.part == part)
                .map(|(at, _)| at)
                .collect();
            let lines: Vec<String> = positions.iter().map(|at| entries[*at].line.clone()).collect();
            for (at, slot) in positions.into_iter().zip(self.logs.append(&part, &lines)?) {
                slots[at] = Some(slot);
            }
        }
        self.logs.sync()?;
        for (entry, slot) in entries.into_iter().zip(slots) {
            let Some(slot) = slot else { continue };
            self.cache.invalidate(&entry.key);
            match entry.record {
                Some(record) => self
                    .registry
                    .apply_put_in(&entry.key, &record, slot, entry.seq, &entry.part),
                None => self
                    .registry
                    .apply_delete_in(&entry.key, slot.length, entry.seq, &entry.part),
            }
        }
        self.note_writes(applied)
    }

    /// Finishes an unfinished journal at reopen: appends every already-serialized `(seq, line)`
    /// pair whose `seq` is above what this table's log already holds, in order, then `fsync`s and
    /// updates the registry — the same "seq higher than the log" rule the TS engine uses. Returns
    /// how many lines were actually appended (0 when this table already has everything).
    pub fn recover_txn_lines(&mut self, entries: &[(u64, String)]) -> Result<usize, TableError> {
        let missing: Vec<&(u64, String)> = entries.iter().filter(|(seq, _)| *seq > self.seq).collect();
        if missing.is_empty() {
            return Ok(0);
        }
        let applied = missing.len();
        // NOTE: parse before appending — a journalled line's partition is only in the line itself.
        let mut pending: Vec<(String, LogEntry, String)> = Vec::with_capacity(applied);
        for (_, line) in &missing {
            let path = self.logs.path_of(UNPARTITIONED).to_string_lossy().into_owned();
            let parsed = parse_entry(line, &path, 0)
                .map_err(|error| TableError::Io(io::Error::new(io::ErrorKind::InvalidData, error)))?;
            let part = match (&parsed.meta.op, &parsed.data) {
                (LogOp::Put, Some(record)) => partition_of(&self.schema, record)?,
                (LogOp::Put, None) => {
                    return Err(TableError::Io(io::Error::new(
                        io::ErrorKind::InvalidData,
                        format!("journalled put for \"{}\" carries no record", parsed.meta.key),
                    )))
                }
                (LogOp::Del, _) => self
                    .registry
                    .part_of(&parsed.meta.key)
                    .unwrap_or(UNPARTITIONED)
                    .to_string(),
            };
            pending.push((part, parsed, line.clone()));
        }

        for (part, parsed, line) in pending {
            let slot = self.logs.append(&part, std::slice::from_ref(&line))?[0];
            self.cache.invalidate(&parsed.meta.key);
            match parsed.data {
                Some(record) => self
                    .registry
                    .apply_put_in(&parsed.meta.key, &record, slot, parsed.meta.seq, &part),
                None => self
                    .registry
                    .apply_delete_in(&parsed.meta.key, slot.length, parsed.meta.seq, &part),
            }
            self.seq = self.seq.max(parsed.meta.seq);
        }
        self.logs.sync()?;
        // NOTE: mirrors TS #redo — recovery always resnapshots, never leans on the op threshold.
        self.snapshot_current = false;
        self.snapshot()?;
        Ok(applied)
    }
}

/// Whether the file behind `log` is no longer the one this handle attached to: a different inode
/// (an atomic rewrite — how `compaction::compact_log` lands), or bytes gone from a file this engine
/// only ever appends to. Where the platform exposes no inode (wasi) only the shrink is visible, so
/// a foreign rewrite landing on exactly the same size passes as "nothing happened" there.
fn rewritten(log: &LogFile, probe: &LogProbe) -> bool {
    match (log.inode(), probe.inode) {
        (Some(held), Some(found)) if held != found => true,
        _ => probe.size < log.size(),
    }
}

/// The partitions `entries` name, in first-appearance order — the order their appends run in, so a
/// unit of work touching two partitions writes them in the order it staged them.
fn ordered_parts(entries: &[PreparedTxnEntry]) -> Vec<String> {
    let mut parts: Vec<String> = Vec::new();
    for entry in entries {
        if !parts.iter().any(|part| part == &entry.part) {
            parts.push(entry.part.clone());
        }
    }
    parts
}

/// The directory a partitioned table's log files live in: `<dir>/<table>/`, beside the
/// `<dir>/<table>.jsonl` an unpartitioned table would use.
fn partition_dir_of(log_path: &Path, table: &str) -> PathBuf {
    match log_path.parent() {
        Some(parent) => parent.join(table),
        None => PathBuf::from(table),
    }
}

/// The log files of a table: the single file at `path` when its schema declares no `partitionBy`,
/// and `<dir>/<table>/<value>.jsonl` when it does — in which case a flat log left over from before
/// the schema declared partitions is moved into that layout first (`storage::ensure_partitioned`).
///
/// A readonly open never runs that move. Between a catalog that already declares `partitionBy` and
/// the move that has not run yet, the records still live in the flat log, and this handle takes them
/// as it finds them: no rename, no deletion, no partition directory brought into being. The choice
/// mirrors `database::present_logs_of`, which decides on the same terms which logs such an open is
/// about to take on — a partition directory holding logs wins, an empty one falls back to the flat
/// log beside it.
fn open_logs(schema: &TableSchema, path: &Path, readonly: bool) -> io::Result<LogSet> {
    if !is_partitioned(schema) {
        return LogSet::flat(path, readonly);
    }
    let partition_dir = partition_dir_of(path, &schema.table);
    if readonly {
        if partition_names_in(&partition_dir).is_empty() && path.exists() {
            return LogSet::flat(path, true);
        }
        return LogSet::partitioned(partition_dir, true);
    }
    ensure_partitioned(path, &partition_dir, schema)?;
    LogSet::partitioned(partition_dir, readonly)
}

/// One operation staged in a multi-table unit of work, ready for [`Table::prepare_txn_ops`].
pub enum TxnOp {
    Put(Value),
    Patch(String, Value),
    Delete(String),
}

/// A validated, `_seq`-assigned, serialized log line produced by [`Table::prepare_txn_ops`] —
/// nothing on disk yet, but everything needed to journal it and, later, to append it.
pub struct PreparedTxnEntry {
    key: String,
    seq: u64,
    line: String,
    old: Option<HaysonRecord>,
    record: Option<HaysonRecord>,
    /// The partition file this line lands in — [`UNPARTITIONED`] on an unpartitioned table.
    part: String,
}

impl PreparedTxnEntry {
    pub fn seq(&self) -> u64 {
        self.seq
    }

    pub fn line(&self) -> &str {
        &self.line
    }

    pub fn old(&self) -> Option<&HaysonRecord> {
        self.old.as_ref()
    }

    pub fn record(&self) -> Option<&HaysonRecord> {
        self.record.as_ref()
    }

    /// `Insert` when no `old` record existed, `Update` otherwise — a `Delete` is never produced
    /// here, since a caller telling insert/update apart from delete already knows which `TxnOp`
    /// it staged; use it against `TxnOp::Put`/`TxnOp::Patch` results only.
    pub fn upsert_event(&self) -> crate::catalog::TriggerEvent {
        if self.old.is_some() {
            crate::catalog::TriggerEvent::Update
        } else {
            crate::catalog::TriggerEvent::Insert
        }
    }
}

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

    use super::*;
    use crate::query::builder::QueryBuilder;
    use crate::query::predicates::ComparisonOp;
    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", "required": true },
                "age": { "type": "int", "index": "sorted" },
                "email": { "type": "string", "unique": true },
            },
        }))
        .unwrap()
    }

    fn open_table() -> (tempfile::TempDir, Table) {
        let dir = tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        let table = Table::open(users_schema(), &path).unwrap();
        (dir, table)
    }

    #[test]
    fn a_table_set_to_sync_never_flushes_its_writes_without_a_single_fsync() {
        let (_dir, mut table) = open_table();
        table.set_sync(CatalogSyncMode::Never);

        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();
        table
            .put(&json!({ "id": "u2", "name": "Piotr", "age": 41 }), false)
            .unwrap();
        table.flush().unwrap();

        assert_eq!(table.sync_mode(), CatalogSyncMode::Never);
        assert_eq!(table.logs.log_of(UNPARTITIONED).unwrap().fsyncs(), 0);
        assert_eq!(table.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
        assert_eq!(table.get("u2").unwrap().unwrap().get("name"), Some(&json!("Piotr")));
    }

    #[test]
    fn a_table_nobody_configured_still_fsyncs_every_flush() {
        let (_dir, mut table) = open_table();

        assert_eq!(table.sync_mode(), CatalogSyncMode::Commit);
        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();
        table.flush().unwrap();

        assert_eq!(table.logs.log_of(UNPARTITIONED).unwrap().fsyncs(), 1);
    }

    #[test]
    fn history_hands_back_every_version_of_one_key_oldest_first() {
        let (dir, mut table) = open_table();
        table
            .put(
                &json!({ "id": "u1", "name": "Ania", "age": 30, "email": "a@example.test" }),
                false,
            )
            .unwrap();
        table
            .put(
                &json!({ "id": "u2", "name": "Piotr", "age": 41, "email": "p@example.test" }),
                false,
            )
            .unwrap();
        table.patch("u1", &json!({ "age": 31 }), false).unwrap();
        table.delete("u1", false).unwrap();
        table
            .put(
                &json!({ "id": "u1", "name": "Ania", "age": 32, "email": "a@example.test" }),
                false,
            )
            .unwrap();

        let history = table.history("u1").unwrap();

        assert_eq!(history.scanned, vec![dir.path().join("users.jsonl")]);
        let shape: Vec<(u64, LogOp, Option<&Value>)> = history
            .entries
            .iter()
            .map(|entry| (entry.seq, entry.op, entry.document.as_ref().and_then(|d| d.get("age"))))
            .collect();
        assert_eq!(
            shape,
            vec![
                (1, LogOp::Put, Some(&json!(30))),
                (3, LogOp::Put, Some(&json!(31))),
                (4, LogOp::Del, None),
                (5, LogOp::Put, Some(&json!(32))),
            ]
        );
        assert!(history.entries.iter().all(|entry| entry.partition == UNPARTITIONED));
        assert!(history.entries.iter().all(|entry| entry.ts > 0));
        assert_eq!(table.history("u2").unwrap().entries.len(), 1);
        assert!(table.history("nobody").unwrap().entries.is_empty());
    }

    #[test]
    fn history_reads_the_line_the_offsets_it_reports_point_at() {
        let (_dir, mut table) = open_table();
        table
            .put(
                &json!({ "id": "u1", "name": "Ania", "age": 30, "email": "a@example.test" }),
                false,
            )
            .unwrap();
        table.patch("u1", &json!({ "age": 31 }), false).unwrap();

        let history = table.history("u1").unwrap();
        let log = table.logs.log_of(UNPARTITIONED).unwrap();

        for entry in &history.entries {
            let slot = Slot {
                offset: entry.offset,
                length: entry.length,
            };
            let line: Value = serde_json::from_str(&log.read_slice(&slot).unwrap()).unwrap();
            assert_eq!(line.get("_seq"), Some(&json!(entry.seq)));
        }
    }

    #[test]
    fn history_does_not_see_a_write_that_is_still_staged() {
        let (_dir, mut table) = open_table();
        table
            .stage_put(
                &json!({ "id": "u1", "name": "Ania", "age": 30, "email": "a@example.test" }),
                false,
            )
            .unwrap();

        assert!(table.history("u1").unwrap().entries.is_empty());

        table.flush().unwrap();

        assert_eq!(table.history("u1").unwrap().entries.len(), 1);
    }

    #[test]
    fn a_table_nobody_configured_refuses_a_write_naming_an_undeclared_field() {
        let (_dir, mut table) = open_table();
        assert_eq!(table.validation(), CatalogValidation::Strict);

        let error = table
            .put(&json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }), false)
            .unwrap_err();

        assert!(error.to_string().contains("nickname"), "{error}");
        assert!(error.to_string().contains("is not declared in the schema"), "{error}");
        assert_eq!(table.get("u1").unwrap(), None);
    }

    #[test]
    fn a_warn_table_takes_the_write_drops_the_undeclared_field_and_says_which() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        let mut table = Table::open(users_schema(), &path).unwrap();
        table.set_validation(CatalogValidation::Warn);
        let document = json!({ "id": "u1", "name": "Ania", "age": 30, "nickname": "Anka" });

        let record = table.put(&document, false).unwrap();

        assert_eq!(record.get("name"), Some(&json!("Ania")));
        assert_eq!(record.get("nickname"), None);
        assert_eq!(table.get("u1").unwrap().unwrap().get("nickname"), None);
        assert_eq!(
            table.dropped_field_warnings(&document),
            vec!["field \"nickname\": is not declared in the schema, dropped from the write".to_string()]
        );
        let written = fs::read_to_string(&path).unwrap();
        assert!(!written.contains("nickname"), "{written}");
        assert!(!written.contains("Anka"), "{written}");
    }

    #[test]
    fn a_warn_table_still_refuses_a_bad_type_a_missing_required_field_and_a_reserved_key() {
        let (_dir, mut table) = open_table();
        table.set_validation(CatalogValidation::Warn);

        let bad_type = table
            .put(&json!({ "id": "u1", "name": "Ania", "age": "thirty" }), false)
            .unwrap_err();
        let missing = table.put(&json!({ "id": "u2" }), false).unwrap_err();
        let reserved = table
            .put(&json!({ "id": "u3", "name": "Ola", "_seq": 7 }), false)
            .unwrap_err();

        assert!(bad_type.to_string().contains("expected an integer"), "{bad_type}");
        assert!(missing.to_string().contains("is required"), "{missing}");
        assert!(reserved.to_string().contains("reserved"), "{reserved}");
    }

    #[test]
    fn a_warn_patch_merges_the_declared_changes_and_drops_the_undeclared_one() {
        let (_dir, mut table) = open_table();
        table.set_validation(CatalogValidation::Warn);
        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();

        let record = table
            .patch("u1", &json!({ "age": 31, "nickname": "Anka" }), false)
            .unwrap();

        assert_eq!(record.get("age"), Some(&json!(31)));
        assert_eq!(record.get("nickname"), None);
        assert_eq!(table.get("u1").unwrap().unwrap().get("age"), Some(&json!(31)));
    }

    #[test]
    fn a_warn_table_prepares_a_unit_of_work_line_without_the_undeclared_field() {
        let (_dir, mut table) = open_table();
        table.set_validation(CatalogValidation::Warn);

        let prepared = table
            .prepare_txn_ops(&[TxnOp::Put(json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }))])
            .unwrap();

        assert_eq!(prepared.len(), 1);
        assert!(!prepared[0].line.contains("nickname"), "{}", prepared[0].line);
        assert_eq!(prepared[0].record.as_ref().unwrap().get("nickname"), None);
    }

    #[test]
    fn dropped_field_warnings_answer_nothing_on_a_strict_table_or_a_document_with_no_stray_field() {
        let (_dir, mut table) = open_table();
        let document = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });

        assert!(table.dropped_field_warnings(&document).is_empty());

        table.set_validation(CatalogValidation::Warn);
        assert!(table
            .dropped_field_warnings(&json!({ "id": "u1", "name": "Ania" }))
            .is_empty());
        assert_eq!(table.dropped_field_warnings(&document).len(), 1);
    }

    fn deprecating(age: &str) -> BTreeMap<String, String> {
        BTreeMap::from([("age".to_string(), age.to_string())])
    }

    #[test]
    fn deprecated_field_warnings_name_the_marked_field_the_document_touches_and_quote_its_note() {
        let (_dir, mut table) = open_table();
        table.set_deprecated_fields(deprecating("use \"bornAt\" instead"));

        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();

        assert_eq!(
            table.deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania", "age": 30 })),
            vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()]
        );
        assert_eq!(table.get("u1").unwrap().unwrap().get("age"), Some(&json!(30)));
    }

    #[test]
    fn deprecated_field_warnings_stay_quiet_for_a_document_touching_no_marked_field() {
        let (_dir, mut table) = open_table();
        table.set_deprecated_fields(deprecating("use \"bornAt\" instead"));

        assert!(table
            .deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania" }))
            .is_empty());
        assert!(table.deprecated_field_warnings(&json!({ "name": "Ola" })).is_empty());
    }

    /// `parse_catalog` refuses `deprecated: ""`, so only a hand-built handle can get here — the
    /// guard exists because `set_deprecated_fields` is public and cannot assume its caller.
    #[test]
    fn a_field_handed_over_with_an_empty_note_warns_with_the_field_alone() {
        let (_dir, mut table) = open_table();
        table.set_deprecated_fields(deprecating(""));

        assert_eq!(
            table.deprecated_field_warnings(&json!({ "id": "u1", "age": 30 })),
            vec!["field \"age\": is deprecated".to_string()]
        );
    }

    #[test]
    fn deprecated_field_warnings_answer_the_same_in_either_validation_mode_and_on_every_write() {
        let (_dir, mut table) = open_table();
        table.set_deprecated_fields(deprecating("use \"bornAt\" instead"));
        let document = json!({ "id": "u1", "name": "Ania", "age": 30 });
        let expected = vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()];

        assert_eq!(table.validation(), CatalogValidation::Strict);
        table.put(&document, false).unwrap();
        assert_eq!(table.deprecated_field_warnings(&document), expected);

        table.set_validation(CatalogValidation::Warn);
        table.put(&document, false).unwrap();
        assert_eq!(table.deprecated_field_warnings(&document), expected);
        assert!(table.dropped_field_warnings(&document).is_empty());
    }

    #[test]
    fn a_table_nobody_marked_answers_no_deprecation_warning_at_all() {
        let (_dir, table) = open_table();

        assert!(table.deprecated_fields().is_empty());
        assert!(table
            .deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania", "age": 30 }))
            .is_empty());
    }

    #[test]
    fn what_a_sync_never_table_wrote_is_still_there_after_reopening_it() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        {
            let mut table = Table::open(users_schema(), &path).unwrap();
            table.set_sync(CatalogSyncMode::Never);
            table
                .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
                .unwrap();
            table.flush().unwrap();
        }

        let reopened = Table::open(users_schema(), &path).unwrap();
        assert_eq!(reopened.get("u1").unwrap().unwrap().get("age"), Some(&json!(30)));
    }

    #[test]
    fn put_then_get_round_trips_the_record() {
        let (_dir, mut table) = open_table();
        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();

        let fetched = table.get("u1").unwrap().unwrap();
        assert_eq!(fetched.get("name"), Some(&json!("Ania")));
        assert_eq!(fetched.get("age"), Some(&json!(30)));
    }

    #[test]
    fn get_of_an_unknown_key_is_none() {
        let (_dir, table) = open_table();
        assert_eq!(table.get("missing").unwrap(), None);
    }

    #[test]
    fn put_delete_get_leaves_nothing_behind() {
        let (_dir, mut table) = open_table();
        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        let existed = table.delete("u1", false).unwrap();
        assert!(existed);
        assert_eq!(table.get("u1").unwrap(), None);
    }

    #[test]
    fn delete_of_an_unknown_key_reports_it_did_not_exist() {
        let (_dir, mut table) = open_table();
        let existed = table.delete("ghost", false).unwrap();
        assert!(!existed);
    }

    #[test]
    fn patch_merges_only_the_given_fields() {
        let (_dir, mut table) = open_table();
        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();

        let patched = table.patch("u1", &json!({ "age": 31 }), false).unwrap();
        assert_eq!(patched.get("name"), Some(&json!("Ania")));
        assert_eq!(patched.get("age"), Some(&json!(31)));

        let fetched = table.get("u1").unwrap().unwrap();
        assert_eq!(fetched, patched);
    }

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

    #[test]
    fn a_live_put_fires_a_same_table_trigger_right_after_the_write_it_fired_on() {
        let (_dir, mut table) = open_table();
        table.set_triggers(vec![self_audit_trigger()]);

        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();

        assert!(table.get("u1").unwrap().is_some());
        let effect = table.get("audit-u1").unwrap().unwrap();
        assert_eq!(effect.get("name"), Some(&json!("Ania")));
    }

    #[test]
    fn a_trigger_only_fires_for_the_event_it_declares() {
        let (_dir, mut table) = open_table();
        table.set_triggers(vec![self_audit_trigger()]);

        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        table.patch("u1", &json!({ "name": "Anna" }), false).unwrap();

        // the insert-only trigger did not re-fire on the update.
        assert_eq!(
            table.get("audit-u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
    }

    #[test]
    fn patch_of_an_unknown_key_is_an_error() {
        let (_dir, mut table) = open_table();
        let err = table.patch("ghost", &json!({ "age": 1 }), false).unwrap_err();
        assert!(matches!(err, TableError::NotFound(_)));
    }

    #[test]
    fn patch_cannot_change_the_primary_key() {
        let (_dir, mut table) = open_table();
        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        let err = table.patch("u1", &json!({ "id": "u2" }), false).unwrap_err();
        assert!(matches!(err, TableError::Schema(_)));
    }

    #[test]
    fn repeated_puts_on_the_same_key_leave_only_the_latest_version() {
        let (_dir, mut table) = open_table();
        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        table.put(&json!({ "id": "u1", "name": "Zosia" }), false).unwrap();
        table.put(&json!({ "id": "u1", "name": "Kasia" }), false).unwrap();

        let fetched = table.get("u1").unwrap().unwrap();
        assert_eq!(fetched.get("name"), Some(&json!("Kasia")));
    }

    /// No field of `users_schema` is marked `fulltext`, so nothing could have scored the text — an
    /// empty answer would claim the table holds no match, which is a different statement.
    #[test]
    fn a_search_against_a_table_with_no_fulltext_field_is_refused_rather_than_answered_empty() {
        let (_dir, mut table) = open_table();
        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();

        let spec = QueryBuilder::new()
            .search("anything", crate::registry::fulltext_index::SearchMode::All)
            .spec();
        let refusal = table.query(&spec).unwrap_err();

        assert!(
            matches!(&refusal, TableError::Schema(_)),
            "a table with no fulltext field is a schema refusal, not an io failure: {refusal:?}"
        );
        assert!(
            refusal.to_string().contains("has no field marked \"fulltext\""),
            "the message is the one the TS package raised: {refusal}"
        );
    }

    #[test]
    fn query_delegates_to_the_planner_and_matches_it_directly() {
        let (_dir, mut table) = open_table();
        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();
        table
            .put(&json!({ "id": "u2", "name": "Piotr", "age": 24 }), false)
            .unwrap();

        let spec = QueryBuilder::new().r#where("age", ComparisonOp::Gt, json!(25)).spec();
        let via_table = table.query(&spec).unwrap();

        let mut source = LogSetRecordSource::new(&table.registry, &table.logs);
        let direct = execute_query(&table.registry, &mut source, &spec).unwrap();

        assert_eq!(via_table, direct);
        assert_eq!(via_table.keys, vec!["u1".to_string()]);
    }

    #[test]
    fn aggregate_delegates_to_the_planner_and_matches_it_directly() {
        let (_dir, mut table) = open_table();
        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();
        table
            .put(&json!({ "id": "u2", "name": "Piotr", "age": 24 }), false)
            .unwrap();

        let (spec, aggregate) = QueryBuilder::new().count().sum("age").avg("age").build_aggregate();
        let via_table = table.aggregate(&spec, &aggregate).unwrap();

        let mut source = LogSetRecordSource::new(&table.registry, &table.logs);
        let direct =
            crate::query::aggregate::execute_aggregate(&table.registry, &mut source, &spec, &aggregate).unwrap();

        assert_eq!(via_table, direct);
        assert_eq!(via_table.count, 2);
        assert_eq!(via_table.sum["age"], 54.0);
        assert_eq!(via_table.avg["age"], Some(27.0));
    }

    fn open_table_with_snapshot_every(snapshot_every: u64) -> (tempfile::TempDir, Table, std::path::PathBuf) {
        let dir = tempdir().unwrap();
        let log_path = dir.path().join("users.jsonl");
        let snapshot_path = dir.path().join("users.idx");
        let (table, _) =
            Table::open_with_snapshot(users_schema(), &log_path, &snapshot_path, "schema-hash", snapshot_every)
                .unwrap();
        (dir, table, snapshot_path)
    }

    #[test]
    fn writes_a_snapshot_once_the_operation_threshold_is_crossed() {
        let (_dir, mut table, snapshot_path) = open_table_with_snapshot_every(3);
        assert!(!snapshot_path.exists());

        for n in 0..3 {
            table
                .put(
                    &json!({ "id": format!("u{n}"), "name": "Ania", "age": 30, "email": format!("u{n}@x.com") }),
                    false,
                )
                .unwrap();
        }

        let snapshot = crate::checkpoint::read_snapshot(&snapshot_path).unwrap().unwrap();
        assert_eq!(snapshot.header.watermark, table.logs.total_size());
        assert_eq!(snapshot.entries.len(), 3);
        assert!(!table.needs_snapshot());
    }

    #[test]
    fn a_snapshot_written_mid_stream_does_not_disturb_later_reads_and_writes() {
        let (_dir, mut table, snapshot_path) = open_table_with_snapshot_every(2);

        table
            .put(
                &json!({ "id": "u1", "name": "Ania", "age": 30, "email": "u1@x.com" }),
                false,
            )
            .unwrap();
        table
            .put(
                &json!({ "id": "u2", "name": "Piotr", "age": 24, "email": "u2@x.com" }),
                false,
            )
            .unwrap();
        assert!(snapshot_path.exists());

        table
            .put(
                &json!({ "id": "u3", "name": "Kasia", "age": 40, "email": "u3@x.com" }),
                false,
            )
            .unwrap();
        let existed = table.delete("u1", false).unwrap();
        assert!(existed);

        assert_eq!(table.get("u1").unwrap(), None);
        assert_eq!(table.get("u3").unwrap().unwrap().get("name"), Some(&json!("Kasia")));
    }

    #[test]
    fn compaction_replaces_the_snapshot_with_one_that_pairs_with_the_rewritten_log() {
        let (_dir, mut table, snapshot_path) = open_table_with_snapshot_every(0);
        for n in 0..4 {
            table
                .put(
                    &json!({ "id": format!("u{n}"), "name": "Ania", "age": 30, "email": format!("u{n}@x.com") }),
                    false,
                )
                .unwrap();
        }
        table.delete("u0", false).unwrap();
        table.delete("u1", false).unwrap();
        table.snapshot().unwrap();
        let stale = crate::checkpoint::read_snapshot(&snapshot_path).unwrap().unwrap();
        assert!(
            stale.header.dead_bytes > 0,
            "the seed really did leave dead bytes behind"
        );

        let result = table.compact().unwrap();

        assert!(result.reclaimed_bytes > 0, "the compaction really did shrink the log");
        let fresh = crate::checkpoint::read_snapshot(&snapshot_path).unwrap().unwrap();
        assert_eq!(fresh.header.watermark, table.logs.total_size());
        assert_eq!(
            fresh.header.watermark,
            fs::metadata(table.logs.log_of("").unwrap().path()).unwrap().len()
        );
        assert!(fresh.header.watermark < stale.header.watermark);
        assert_eq!(fresh.header.dead_bytes, 0);
        assert_eq!(fresh.entries.len(), 2);
        assert!(!table.needs_snapshot());
    }

    #[test]
    fn the_generation_counts_compactions_and_survives_a_restart_through_the_snapshot_header() {
        let dir = tempdir().unwrap();
        let log_path = dir.path().join("users.jsonl");
        let snapshot_path = dir.path().join("users.idx");

        {
            let (mut table, _) =
                Table::open_with_snapshot(users_schema(), &log_path, &snapshot_path, "schema-hash", 0).unwrap();
            assert_eq!(
                table.stats().gen,
                0,
                "a table opened over a log with no snapshot starts at zero"
            );
            for n in 0..3 {
                table
                    .put(
                        &json!({ "id": format!("u{n}"), "name": "Ania", "age": 30, "email": format!("u{n}@x.com") }),
                        false,
                    )
                    .unwrap();
            }
            table.delete("u0", false).unwrap();
            table.compact().unwrap();
            assert_eq!(table.stats().gen, 1);
            table.compact().unwrap();
            assert_eq!(table.stats().gen, 2);
            assert_eq!(
                crate::checkpoint::read_snapshot(&snapshot_path)
                    .unwrap()
                    .unwrap()
                    .header
                    .gen,
                2,
                "the snapshot carries the generation it describes"
            );
        }

        let (mut reopened, outcome) =
            Table::open_with_snapshot(users_schema(), &log_path, &snapshot_path, "schema-hash", 0).unwrap();
        assert_eq!(outcome, RestoreOutcome::Restored);
        assert_eq!(
            reopened.stats().gen,
            2,
            "the restart carries the generation over instead of starting again"
        );
        reopened.compact().unwrap();
        assert_eq!(reopened.stats().gen, 3);
        assert_eq!(
            crate::checkpoint::read_snapshot(&snapshot_path)
                .unwrap()
                .unwrap()
                .header
                .gen,
            3
        );
    }

    /// The snapshot header grew `gen` after this engine had already written snapshots without it,
    /// so a file from before that has to stay readable — and the counter has to start somewhere
    /// sane rather than throwing the whole cache away over a missing hint.
    #[test]
    fn a_snapshot_written_before_the_header_carried_a_generation_still_restores_and_starts_at_zero() {
        let dir = tempdir().unwrap();
        let log_path = dir.path().join("users.jsonl");
        let snapshot_path = dir.path().join("users.idx");

        {
            let (mut table, _) =
                Table::open_with_snapshot(users_schema(), &log_path, &snapshot_path, "schema-hash", 0).unwrap();
            for n in 0..3 {
                table
                    .put(
                        &json!({ "id": format!("u{n}"), "name": "Ania", "age": 30, "email": format!("u{n}@x.com") }),
                        false,
                    )
                    .unwrap();
            }
            table.compact().unwrap();
            assert_eq!(table.stats().gen, 1);
        }

        let written = fs::read_to_string(&snapshot_path).unwrap();
        let (header_line, entry_lines) = written.split_once('\n').unwrap();
        let mut header: serde_json::Map<String, Value> = serde_json::from_str(header_line).unwrap();
        assert!(
            header.remove("gen").is_some(),
            "the header this engine writes carries the field to remove"
        );
        fs::write(
            &snapshot_path,
            format!("{}\n{entry_lines}", serde_json::to_string(&header).unwrap()),
        )
        .unwrap();

        let (mut reopened, outcome) =
            Table::open_with_snapshot(users_schema(), &log_path, &snapshot_path, "schema-hash", 0).unwrap();
        assert_eq!(
            outcome,
            RestoreOutcome::Restored,
            "an old snapshot is still a usable cache"
        );
        assert_eq!(reopened.len(), 3);
        assert_eq!(reopened.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
        assert_eq!(reopened.stats().gen, 0);
        reopened.compact().unwrap();
        assert_eq!(
            reopened.stats().gen,
            1,
            "counting resumes from the default, and lands in the rewritten header"
        );
        assert_eq!(
            crate::checkpoint::read_snapshot(&snapshot_path)
                .unwrap()
                .unwrap()
                .header
                .gen,
            1
        );
    }

    #[test]
    fn reads_agree_before_and_after_a_restart_that_follows_a_compaction() {
        let dir = tempdir().unwrap();
        let log_path = dir.path().join("users.jsonl");
        let snapshot_path = dir.path().join("users.idx");
        let spec = QueryBuilder::new()
            .r#where("name", ComparisonOp::Eq, json!("Ania"))
            .spec();

        let (before, after_compaction) = {
            let (mut table, _) =
                Table::open_with_snapshot(users_schema(), &log_path, &snapshot_path, "schema-hash", 0).unwrap();
            for n in 0..5 {
                table
                    .put(
                        &json!({ "id": format!("u{n}"), "name": "Ania", "age": 30, "email": format!("u{n}@x.com") }),
                        false,
                    )
                    .unwrap();
            }
            table.patch("u2", &json!({ "age": 31 }), false).unwrap();
            table.delete("u4", false).unwrap();
            table.snapshot().unwrap();
            let before = table.query(&spec).unwrap().keys;

            table.compact().unwrap();
            let after = table.query(&spec).unwrap().keys;
            (before, after)
        };

        let (reopened, outcome) =
            Table::open_with_snapshot(users_schema(), &log_path, &snapshot_path, "schema-hash", 0).unwrap();
        assert_eq!(
            outcome,
            RestoreOutcome::Restored,
            "the snapshot compaction wrote pairs with the new log"
        );
        assert_eq!(
            after_compaction, before,
            "compaction changed neither the set nor the order of live keys"
        );
        assert_eq!(reopened.query(&spec).unwrap().keys, before);
        assert_eq!(reopened.len(), 4);
        assert_eq!(reopened.get("u2").unwrap().unwrap().get("age"), Some(&json!(31)));
        assert_eq!(reopened.get("u4").unwrap(), None);
    }

    #[test]
    fn zero_disables_periodic_snapshotting() {
        let (_dir, mut table, snapshot_path) = open_table_with_snapshot_every(0);
        for n in 0..10 {
            table
                .put(
                    &json!({ "id": format!("u{n}"), "name": "Ania", "age": 30, "email": format!("u{n}@x.com") }),
                    false,
                )
                .unwrap();
        }
        assert!(!snapshot_path.exists());
        assert!(table.needs_snapshot());
    }

    #[test]
    fn a_table_opened_without_a_snapshot_path_never_writes_one() {
        let (_dir, mut table) = open_table();
        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        table.snapshot().unwrap();
        assert!(!table.needs_snapshot());
    }

    /// Two handles over the same log file, as two processes of one directory would have — nothing
    /// at the `Table` level keeps them apart, the writer lock lives a layer up in `Database`.
    fn two_handles() -> (tempfile::TempDir, Table, Table) {
        let dir = tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        let writer = Table::open(users_schema(), &path).unwrap();
        let follower = Table::open(users_schema(), &path).unwrap();
        (dir, writer, follower)
    }

    #[test]
    fn refresh_on_a_log_nobody_touched_reports_no_change_and_leaves_the_registry_alone() {
        let (_dir, mut table) = open_table();
        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();

        let first = table.refresh().unwrap();
        let second = table.refresh().unwrap();

        assert_eq!(first, RefreshOutcome::Unchanged);
        assert_eq!(
            second,
            RefreshOutcome::Unchanged,
            "a handle is not behind its own writes"
        );
        assert!(!first.changed());
        assert_eq!(table.len(), 1);
        assert_eq!(table.get("u1").unwrap().unwrap().get("age"), Some(&json!(30)));
    }

    #[test]
    fn refresh_applies_the_records_another_handle_appended_to_the_same_log() {
        let (_dir, mut writer, mut follower) = two_handles();
        writer.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        assert_eq!(
            follower.get("u1").unwrap(),
            None,
            "a handle sees nothing until it looks again"
        );

        let outcome = follower.refresh().unwrap();

        assert_eq!(outcome, RefreshOutcome::CaughtUp { entries: 1 });
        assert_eq!(follower.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
        assert_eq!(follower.len(), 1);
        assert_eq!(follower.stats().seq, writer.stats().seq);
    }

    #[test]
    fn refresh_applies_a_foreign_delete_as_well_as_a_foreign_write() {
        let (_dir, mut writer, mut follower) = two_handles();
        writer.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        writer.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
        follower.refresh().unwrap();
        writer.delete("u1", false).unwrap();

        assert_eq!(follower.refresh().unwrap(), RefreshOutcome::CaughtUp { entries: 1 });
        assert!(!follower.has("u1"));
        assert!(follower.has("u2"));
        assert_eq!(follower.len(), 1);
    }

    #[test]
    fn refresh_rebuilds_after_another_handle_compacted_the_log_under_it() {
        let (_dir, mut writer, mut follower) = two_handles();
        writer
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();
        writer.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
        writer.patch("u1", &json!({ "age": 31 }), false).unwrap();
        writer.delete("u2", false).unwrap();
        follower.refresh().unwrap();
        let before = follower.stats().log_bytes;

        writer.compact().unwrap();

        let outcome = follower.refresh().unwrap();
        assert_eq!(outcome, RefreshOutcome::Reloaded);
        assert!(
            follower.stats().log_bytes < before,
            "the compaction really did shrink the file"
        );
        assert_eq!(follower.len(), 1);
        assert_eq!(follower.get("u1").unwrap().unwrap().get("age"), Some(&json!(31)));
        assert_eq!(follower.get("u2").unwrap(), None);
    }

    /// The case a size watermark alone cannot see: after the compaction the writer appends enough
    /// to push the rewritten file past where the follower's watermark stood, so the file looks like
    /// it merely grew. Only the inode says otherwise — which is why this test is unix-only; wasi
    /// exposes no inode and `rewritten()` degrades to the shrink it can see.
    #[cfg(unix)]
    #[test]
    fn refresh_catches_a_foreign_compaction_even_when_the_log_ends_up_longer_than_before() {
        let (_dir, mut writer, mut follower) = two_handles();
        for round in 0..6 {
            writer
                .put(&json!({ "id": "u1", "name": format!("round {round}") }), false)
                .unwrap();
        }
        follower.refresh().unwrap();
        let watermark = follower.stats().log_bytes;

        writer.compact().unwrap();
        for id in ["u2", "u3", "u4", "u5", "u6"] {
            writer
                .put(&json!({ "id": id, "name": "after the compaction" }), false)
                .unwrap();
        }
        assert!(
            writer.stats().log_bytes > watermark,
            "the log grew past where the follower stood"
        );

        let outcome = follower.refresh().unwrap();

        assert_eq!(outcome, RefreshOutcome::Reloaded);
        assert_eq!(follower.len(), 6);
        assert_eq!(
            follower.get("u1").unwrap().unwrap().get("name"),
            Some(&json!("round 5"))
        );
        assert_eq!(
            follower.get("u6").unwrap().unwrap().get("name"),
            Some(&json!("after the compaction"))
        );
    }

    #[test]
    fn a_reload_keeps_serving_queries_from_the_declared_indexes_it_was_built_with() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        let declared = vec![ExpandedIndex {
            name: "users_name_email".into(),
            fields: vec!["name".into(), "email".into()],
            kind: crate::catalog::CatalogIndexKind::Exact,
            unique: false,
            where_: None,
            expire_after_seconds: None,
        }];
        let mut writer = Table::open_with_declared_indexes(users_schema(), &path, &declared).unwrap();
        let mut follower = Table::open_with_declared_indexes(users_schema(), &path, &declared).unwrap();
        writer
            .put(&json!({ "id": "u1", "name": "Ania", "email": "a@x.pl" }), false)
            .unwrap();
        follower.refresh().unwrap();
        writer.compact().unwrap();

        assert_eq!(follower.refresh().unwrap(), RefreshOutcome::Reloaded);
        let spec = QueryBuilder::new()
            .r#where("name", ComparisonOp::Eq, json!("Ania"))
            .r#where("email", ComparisonOp::Eq, json!("a@x.pl"))
            .spec();
        let found = follower.query(&spec).unwrap();
        assert_eq!(found.keys, vec!["u1".to_string()]);
        assert_eq!(
            found.plan.index.as_deref(),
            Some("users_name_email"),
            "a rebuilt registry kept the declared index"
        );
    }

    #[test]
    fn refresh_opens_a_partition_file_another_handle_created_and_reads_it_whole() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("notes.jsonl");
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "notes",
            "primaryKey": "id",
            "partitionBy": "kind",
            "fields": { "id": { "type": "string" }, "kind": { "type": "string", "required": true } },
        }))
        .unwrap();
        let mut writer = Table::open(schema.clone(), &path).unwrap();
        let mut follower = Table::open(schema, &path).unwrap();
        writer.put(&json!({ "id": "a1", "kind": "adr" }), false).unwrap();
        writer.put(&json!({ "id": "n1", "kind": "note" }), false).unwrap();

        let outcome = follower.refresh().unwrap();

        assert_eq!(outcome, RefreshOutcome::CaughtUp { entries: 2 });
        let mut keys: Vec<&String> = follower.keys().collect();
        keys.sort();
        assert_eq!(keys, vec!["a1", "n1"]);
    }

    /// The gate a binding asks under a shared lock before deciding whether it has to take an
    /// exclusive one. It has to answer exactly what `follow_if_due` would then do, or the cheap
    /// check and the expensive one disagree and a follower silently stops following.
    #[test]
    fn follow_is_due_only_where_follow_if_due_would_reach_for_the_files() {
        let (_dir, mut writer, mut follower) = two_handles();
        assert!(!follower.follow_is_due(), "no policy is the writable handle's default");
        assert_eq!(follower.follow_if_due().unwrap(), None);

        follower.set_follow(Some(FollowPolicy {
            interval: Duration::ZERO,
        }));
        assert!(follower.follow_is_due(), "a zero interval is due on every read");

        follower.set_follow(Some(FollowPolicy {
            interval: Duration::from_secs(60),
        }));
        assert!(
            follower.follow_is_due(),
            "and a handle that never looked is due whatever it is"
        );
        writer.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        assert_eq!(
            follower.follow_if_due().unwrap(),
            Some(RefreshOutcome::CaughtUp { entries: 1 })
        );

        assert!(
            !follower.follow_is_due(),
            "the interval has not elapsed since that catch-up"
        );
        writer.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
        assert_eq!(follower.follow_if_due().unwrap(), None);
        assert!(!follower.has("u2"), "which is the whole point of the interval");
    }

    /// The other gate: `flush` is a no-op on an empty staging window, and this has to be true
    /// exactly when it would not be, so a read that flushes first can skip the flush.
    #[test]
    fn has_staged_writes_answers_whether_a_flush_would_do_anything() {
        let (_dir, mut table) = open_table();
        assert!(!table.has_staged_writes(), "a table nobody staged against");

        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        assert!(!table.has_staged_writes(), "put stages and flushes inside the one call");

        table.stage_put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
        assert!(table.has_staged_writes());
        assert_eq!(table.stats().queue_depth, 1);

        table.flush().unwrap();
        assert!(
            !table.has_staged_writes(),
            "and the window is empty again once it lands"
        );
    }

    #[test]
    fn keys_has_len_and_stats_report_the_live_records_not_the_bytes_behind_them() {
        let (_dir, mut table) = open_table();
        assert!(table.is_empty());
        table
            .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
            .unwrap();
        table.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
        table.patch("u1", &json!({ "age": 31 }), false).unwrap();
        table.delete("u2", false).unwrap();

        let mut keys: Vec<&String> = table.keys().collect();
        keys.sort();
        let stats = table.stats();

        assert_eq!(keys, vec!["u1"]);
        assert_eq!(table.len(), 1, "a tombstoned key is not a record");
        assert!(!table.is_empty());
        assert!(table.has("u1"));
        assert!(!table.has("u2"));
        assert!(!table.has("never-written"));
        assert_eq!(stats.records, 1);
        assert!(
            stats.dead_bytes > 0,
            "the superseded version and the tombstone are dead bytes"
        );
        assert_eq!(stats.log_bytes, stats.live_bytes + stats.dead_bytes);
        assert_eq!(stats.seq, 4, "one seq per appended line, tombstone included");
        assert_eq!(stats.gen, 0);
    }

    #[test]
    fn pending_is_empty_between_calls_because_every_write_commits_inside_its_own_call() {
        let (_dir, mut table) = open_table();
        table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();

        assert!(table.pending("u1").is_empty());
        assert!(table.pending("never-written").is_empty());
        assert_eq!(table.stats().queue_depth, 0);
    }

    /// The staging window `put`/`patch`/`delete` open and close inside their own call, driven a
    /// step at a time — the shape the N-API binding uses to answer a caller before its write is
    /// durable, and the semantics `packages/hayson/test/table.test.ts` measures.
    mod staging {
        use super::*;

        #[test]
        fn a_staged_put_is_read_back_before_it_is_a_record() {
            let (_dir, mut table) = open_table();
            let staged = table.stage_put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();

            assert_eq!(staged.get("name"), Some(&json!("Ania")));
            assert_eq!(table.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
            assert!(table.has("u1"));
            assert_eq!(table.len(), 0, "size counts the registry, never the window");
            assert!(table.keys().next().is_none(), "and so does keys()");
            assert_eq!(table.stats().queue_depth, 1);
            assert_eq!(
                table.pending("u1"),
                HashSet::from(["id".to_string(), "name".to_string()])
            );

            table.flush().unwrap();

            assert_eq!(table.len(), 1);
            assert_eq!(table.stats().queue_depth, 0);
            assert!(table.pending("u1").is_empty());
        }

        #[test]
        fn a_staged_delete_answers_gone_before_its_tombstone_is_written() {
            let (_dir, mut table) = open_table();
            table.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();

            assert!(
                table.stage_delete("u1", false).unwrap(),
                "it says what it removed right away"
            );
            assert_eq!(table.get("u1").unwrap(), None);
            assert!(!table.has("u1"));
            assert_eq!(table.len(), 1, "the record is a record until the tombstone lands");

            table.flush().unwrap();

            assert_eq!(table.len(), 0);
            assert_eq!(table.get("u1").unwrap(), None);
        }

        #[test]
        fn a_staged_write_outranks_a_peers_committed_one_on_the_same_key() {
            let (_dir, mut writer, mut follower) = two_handles();
            follower
                .stage_put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();
            writer.put(&json!({ "id": "u1", "name": "Zosia" }), false).unwrap();

            assert!(follower.refresh().unwrap().changed());
            assert_eq!(
                follower.get("u1").unwrap().unwrap().get("name"),
                Some(&json!("Ania")),
                "the foreign write landed in the registry, but the window is read first"
            );

            follower.flush().unwrap();

            assert_eq!(follower.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
            assert_eq!(follower.len(), 1);
        }

        #[test]
        fn a_staged_patch_claims_the_fields_it_changes_and_no_others() {
            let (_dir, mut table) = open_table();
            table
                .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
                .unwrap();

            table.stage_patch("u1", &json!({ "age": 31 }), false).unwrap();
            table.stage_patch("u1", &json!({ "name": "Anna" }), false).unwrap();
            assert_eq!(
                table.pending("u1"),
                HashSet::from(["age".to_string(), "name".to_string()])
            );

            let refused = table.stage_patch("u1", &json!({ "age": 32 }), false).unwrap_err();
            assert!(matches!(refused, TableError::Conflict(_)));

            table.stage_patch("u1", &json!({ "age": 44 }), true).unwrap();
            table.flush().unwrap();

            let found = table.get("u1").unwrap().unwrap();
            assert_eq!(
                found.get("age"),
                Some(&json!(44)),
                "the forced patch is the one that stuck"
            );
            assert_eq!(
                found.get("name"),
                Some(&json!("Anna")),
                "and it merged over the other one"
            );
        }

        #[test]
        fn a_read_of_committed_state_flushes_the_window_first() {
            let (_dir, mut table) = open_table();
            table
                .stage_put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
                .unwrap();

            let spec = QueryBuilder::new().r#where("age", ComparisonOp::Gt, json!(25)).spec();
            assert_eq!(table.query_following(&spec).unwrap().keys, vec!["u1".to_string()]);
            assert_eq!(table.stats().queue_depth, 0);

            table
                .stage_put(&json!({ "id": "u2", "name": "Piotr", "age": 24 }), false)
                .unwrap();
            let (counted, aggregate) = QueryBuilder::new().count().build_aggregate();
            assert_eq!(table.aggregate_following(&counted, &aggregate).unwrap().count, 2);
            assert_eq!(table.stats().queue_depth, 0);

            table
                .stage_put(&json!({ "id": "u3", "name": "Kasia", "age": 51 }), false)
                .unwrap();
            table.compact().unwrap();

            assert_eq!(
                table.stats().queue_depth,
                0,
                "compaction would have rewritten the log without it"
            );
            assert_eq!(table.len(), 3);
            assert_eq!(table.get("u3").unwrap().unwrap().get("name"), Some(&json!("Kasia")));
        }

        #[test]
        fn a_trigger_fires_on_the_flush_rather_than_on_the_staging() {
            let (_dir, mut table) = open_table();
            table.set_triggers(vec![self_audit_trigger()]);

            table.stage_put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
            assert_eq!(
                table.get("audit-u1").unwrap(),
                None,
                "an uncommitted write has no effect yet"
            );

            table.flush().unwrap();

            assert_eq!(
                table.get("audit-u1").unwrap().unwrap().get("name"),
                Some(&json!("Ania"))
            );
        }
    }

    #[test]
    fn stats_counts_a_rebuild_as_a_generation() {
        let (_dir, mut writer, mut follower) = two_handles();
        writer.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        follower.refresh().unwrap();
        assert_eq!(follower.stats().gen, 0);

        writer.compact().unwrap();
        follower.refresh().unwrap();

        assert_eq!(follower.stats().gen, 1);
    }

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

    fn open_docs_table() -> (tempfile::TempDir, Table) {
        let dir = tempdir().unwrap();
        let path = dir.path().join("docs.jsonl");
        let table = Table::open(docs_schema(), &path).unwrap();
        (dir, table)
    }

    #[test]
    fn a_put_with_an_image_field_writes_its_bytes_under_the_default_images_directory() {
        let (dir, mut table) = open_docs_table();
        let record = table
            .put(
                &json!({ "id": "d1", "cover": { "content": "bG9nby1ieXRlcw==", "virtualPath": "/images/logo.png" } }),
                false,
            )
            .unwrap();
        let physical_path = record
            .get("cover")
            .unwrap()
            .get("physicalPath")
            .unwrap()
            .as_str()
            .unwrap();

        assert_eq!(table.image_dir(), dir.path().join("docs.images"));
        let on_disk = fs::read(table.image_dir().join(physical_path)).unwrap();
        assert_eq!(on_disk, b"logo-bytes");
    }

    #[test]
    fn a_second_record_with_the_same_image_content_does_not_rewrite_the_existing_file() {
        let (_dir, mut table) = open_docs_table();
        table
            .put(
                &json!({ "id": "d1", "cover": { "content": "c2FtZS1ieXRlcw==", "virtualPath": "/images/a.png" } }),
                false,
            )
            .unwrap();
        let file_path = table.image_dir().join(physical_path_of(b"same-bytes"));
        let written_at = fs::metadata(&file_path).unwrap().modified().unwrap();

        table
            .put(
                &json!({ "id": "d2", "cover": { "content": "c2FtZS1ieXRlcw==", "virtualPath": "/images/b.png" } }),
                false,
            )
            .unwrap();

        assert_eq!(fs::metadata(&file_path).unwrap().modified().unwrap(), written_at);
        assert_eq!(fs::read(&file_path).unwrap(), b"same-bytes");
    }

    #[test]
    fn the_images_directory_is_created_automatically_on_first_write() {
        let (_dir, mut table) = open_docs_table();
        assert!(!table.image_dir().exists());

        table
            .put(
                &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
                false,
            )
            .unwrap();

        assert!(table.image_dir().is_dir());
    }

    #[test]
    fn a_patch_that_does_not_touch_the_image_field_still_validates() {
        let (_dir, mut table) = open_docs_table();
        table
            .put(
                &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
                false,
            )
            .unwrap();

        let patched = table.patch("d1", &json!({}), false).unwrap();
        assert_eq!(
            patched.get("cover").unwrap().get("virtualPath"),
            Some(&json!("/images/a.png"))
        );
    }

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

    fn open_pages_table() -> (tempfile::TempDir, Table) {
        let dir = tempdir().unwrap();
        let path = dir.path().join("pages.jsonl");
        let table = Table::open(pages_schema(), &path).unwrap();
        (dir, table)
    }

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

    fn open_pages_with_cover_table() -> (tempfile::TempDir, Table) {
        let dir = tempdir().unwrap();
        let path = dir.path().join("pages.jsonl");
        let table = Table::open(pages_with_cover_schema(), &path).unwrap();
        (dir, table)
    }

    /// `page` is a link target, `notes` deliberately is not — the pair every `addressable: false`
    /// test needs in one table.
    fn open_pages_with_notes_table() -> (tempfile::TempDir, Table) {
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "pages",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
                "notes": { "type": "markdown", "addressable": false },
            },
        }))
        .unwrap();
        let dir = tempdir().unwrap();
        let path = dir.path().join("pages.jsonl");
        let table = Table::open(schema, &path).unwrap();
        (dir, table)
    }

    #[test]
    fn a_non_addressable_field_links_to_an_addressable_one_and_the_link_resolves() {
        let (_dir, mut table) = open_pages_with_notes_table();

        let outcome = table
            .put_with_warnings(
                &json!({
                    "id": "a",
                    "page": { "content": "the target", "virtualPath": "/docs/a.md" },
                    "notes": { "content": "see [a](/docs/a.md)" },
                }),
                false,
            )
            .unwrap();

        assert!(
            outcome.warnings.is_empty(),
            "the content of a non-addressable field is still scanned, and this link resolves: {:?}",
            outcome.warnings
        );
        assert_eq!(
            table.broken_links_of("a").unwrap(),
            Vec::new(),
            "asking on demand has to agree"
        );
    }

    #[test]
    fn a_broken_link_in_a_non_addressable_field_is_still_reported_against_that_field() {
        let (_dir, mut table) = open_pages_with_notes_table();

        let outcome = table
            .put_with_warnings(
                &json!({
                    "id": "a",
                    "page": { "content": "", "virtualPath": "/docs/a.md" },
                    "notes": { "content": "see [gone](/docs/missing.md)" },
                }),
                false,
            )
            .unwrap();

        assert_eq!(
            outcome.warnings,
            vec![BrokenLink {
                field: "notes".to_string(),
                path: "/docs/missing.md".to_string()
            }]
        );
    }

    #[test]
    fn nothing_can_link_to_a_non_addressable_field_because_it_holds_no_virtual_path() {
        let (_dir, mut table) = open_pages_with_notes_table();
        table
            .put(
                &json!({
                    "id": "a",
                    "page": { "content": "", "virtualPath": "/docs/a.md" },
                    "notes": { "content": "unreachable", "virtualPath": "/notes/a.md" },
                }),
                false,
            )
            .unwrap();

        let outcome = table
            .put_with_warnings(
                &json!({
                    "id": "b",
                    "page": { "content": "[notes](/notes/a.md)", "virtualPath": "/docs/b.md" },
                }),
                false,
            )
            .unwrap();

        assert_eq!(
            table.registry.virtual_path_fields().cloned().collect::<Vec<String>>(),
            vec!["page".to_string()],
            "only the addressable field belongs in the virtual-path index"
        );
        assert_eq!(
            outcome.warnings,
            vec![BrokenLink {
                field: "page".to_string(),
                path: "/notes/a.md".to_string()
            }],
            "a path only a non-addressable field carries is not a link target"
        );
    }

    #[test]
    fn two_records_may_share_the_path_of_a_non_addressable_field() {
        let (_dir, mut table) = open_pages_with_notes_table();
        let record = |id: &str| {
            json!({
                "id": id,
                "page": { "content": "", "virtualPath": format!("/docs/{id}.md") },
                "notes": { "content": "x", "virtualPath": "/notes/shared.md" },
            })
        };

        table.put(&record("a"), false).unwrap();
        let second = table.put(&record("b"), false);

        assert!(
            second.is_ok(),
            "an unindexed path enforces no uniqueness: {:?}",
            second.err()
        );
    }

    #[test]
    fn a_non_addressable_field_needs_no_path_of_its_own_at_all() {
        let (_dir, mut table) = open_pages_with_notes_table();

        table
            .put(
                &json!({ "id": "a", "page": { "content": "", "virtualPath": "/docs/a.md" }, "notes": { "content": "x" } }),
                false,
            )
            .unwrap();

        let stored = table.get("a").unwrap().unwrap();
        assert_eq!(stored["notes"], json!({ "content": "x" }));
    }

    #[test]
    fn a_table_opened_with_no_link_parsing_option_defaults_to_eager() {
        let (_dir, table) = open_pages_table();
        assert_eq!(table.link_parsing(), LinkParsing::Eager);
    }

    #[test]
    fn eager_put_of_a_broken_link_still_succeeds_but_carries_a_warning() {
        let (_dir, mut table) = open_pages_table();
        let outcome = table
            .put_with_warnings(
                &json!({ "id": "a", "page": { "content": "[missing](/docs/missing.md)", "virtualPath": "/docs/a.md" } }),
                false,
            )
            .unwrap();

        assert!(table.get("a").unwrap().is_some());
        assert_eq!(
            outcome.warnings,
            vec![BrokenLink {
                field: "page".to_string(),
                path: "/docs/missing.md".to_string()
            }]
        );
    }

    #[test]
    fn eager_put_of_a_link_to_an_existing_page_carries_no_warnings() {
        let (_dir, mut table) = open_pages_table();
        table
            .put_with_warnings(
                &json!({ "id": "a", "page": { "content": "", "virtualPath": "/docs/a.md" } }),
                false,
            )
            .unwrap();

        let outcome = table
            .put_with_warnings(
                &json!({ "id": "b", "page": { "content": "[a](/docs/a.md)", "virtualPath": "/docs/b.md" } }),
                false,
            )
            .unwrap();

        assert!(outcome.warnings.is_empty());
    }

    #[test]
    fn eager_put_of_a_link_to_a_cover_images_virtual_path_in_the_same_record_carries_no_warnings() {
        let (_dir, mut table) = open_pages_with_cover_table();
        let outcome = table
            .put_with_warnings(
                &json!({
                    "id": "a",
                    "page": { "content": "[cover](/images/a.png)", "virtualPath": "/docs/a.md" },
                    "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" },
                }),
                false,
            )
            .unwrap();

        assert!(outcome.warnings.is_empty());
    }

    #[test]
    fn eager_put_of_a_link_to_a_cover_images_virtual_path_in_another_record_carries_no_warnings() {
        let (_dir, mut table) = open_pages_with_cover_table();
        table
            .put_with_warnings(
                &json!({ "id": "a", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
                false,
            )
            .unwrap();

        let outcome = table
            .put_with_warnings(
                &json!({ "id": "b", "page": { "content": "[cover](/images/a.png)", "virtualPath": "/docs/b.md" } }),
                false,
            )
            .unwrap();

        assert!(outcome.warnings.is_empty());
    }

    #[test]
    fn eager_put_of_a_link_to_a_path_no_field_holds_still_carries_a_warning() {
        let (_dir, mut table) = open_pages_with_cover_table();
        table
            .put_with_warnings(
                &json!({ "id": "a", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
                false,
            )
            .unwrap();

        let outcome = table
            .put_with_warnings(
                &json!({ "id": "b", "page": { "content": "[nope](/images/missing.png)", "virtualPath": "/docs/b.md" } }),
                false,
            )
            .unwrap();

        assert_eq!(
            outcome.warnings,
            vec![BrokenLink {
                field: "page".to_string(),
                path: "/images/missing.png".to_string()
            }]
        );
    }

    #[test]
    fn lazy_link_parsing_reports_no_automatic_warning_but_broken_links_of_finds_it_on_request() {
        let (_dir, mut table) = open_pages_table();
        table.set_link_parsing(LinkParsing::Lazy);

        let outcome = table
            .put_with_warnings(
                &json!({ "id": "a", "page": { "content": "[missing](/docs/missing.md)", "virtualPath": "/docs/a.md" } }),
                false,
            )
            .unwrap();
        assert!(outcome.warnings.is_empty());

        let broken = table.broken_links_of("a").unwrap();
        assert_eq!(
            broken,
            vec![BrokenLink {
                field: "page".to_string(),
                path: "/docs/missing.md".to_string()
            }]
        );
    }

    #[test]
    fn asking_after_the_write_gives_an_eager_table_the_same_warnings_put_with_warnings_carried() {
        let (_dir, mut table) = open_pages_table();
        let outcome = table
            .put_with_warnings(
                &json!({ "id": "a", "page": { "content": "[missing](/docs/missing.md)", "virtualPath": "/docs/a.md" } }),
                false,
            )
            .unwrap();

        assert_eq!(table.eager_link_warnings_of("a").unwrap(), outcome.warnings);
        assert_eq!(
            table.eager_link_warnings_of("a").unwrap(),
            vec![BrokenLink {
                field: "page".to_string(),
                path: "/docs/missing.md".to_string()
            }]
        );
    }

    #[test]
    fn asking_after_the_write_stays_silent_on_a_lazy_table_that_broken_links_of_would_answer_for() {
        let (_dir, mut table) = open_pages_table();
        table.set_link_parsing(LinkParsing::Lazy);
        table
            .put_with_warnings(
                &json!({ "id": "a", "page": { "content": "[missing](/docs/missing.md)", "virtualPath": "/docs/a.md" } }),
                false,
            )
            .unwrap();

        assert!(table.eager_link_warnings_of("a").unwrap().is_empty());
        assert_eq!(table.broken_links_of("a").unwrap().len(), 1);
    }

    #[test]
    fn plain_put_and_patch_still_return_a_bare_record_alongside_the_warning_variants() {
        let (_dir, mut table) = open_pages_table();
        let record = table
            .put(
                &json!({ "id": "a", "page": { "content": "no links here", "virtualPath": "/docs/a.md" } }),
                false,
            )
            .unwrap();
        assert_eq!(record.get("id"), Some(&json!("a")));

        let patched = table
            .patch(
                "a",
                &json!({ "page": { "content": "[missing](/docs/x.md)", "virtualPath": "/docs/a.md" } }),
                false,
            )
            .unwrap();
        assert_eq!(patched.get("id"), Some(&json!("a")));
    }

    #[test]
    fn compact_drops_dead_versions_but_leaves_the_table_fully_usable() {
        let (_dir, mut table) = open_table();
        table
            .put(
                &json!({ "id": "u1", "name": "Ania", "age": 30, "email": "u1@x.com" }),
                false,
            )
            .unwrap();
        table
            .put(
                &json!({ "id": "u2", "name": "Piotr", "age": 24, "email": "u2@x.com" }),
                false,
            )
            .unwrap();
        table
            .put(
                &json!({ "id": "u3", "name": "Kasia", "age": 40, "email": "u3@x.com" }),
                false,
            )
            .unwrap();
        table.patch("u2", &json!({ "age": 25 }), false).unwrap();
        table.delete("u3", false).unwrap();

        let result = table.compact().unwrap();
        assert_eq!(result.records, 2);
        assert!(result.reclaimed_bytes > 0);

        let u1 = table.get("u1").unwrap().unwrap();
        assert_eq!(u1.get("name"), Some(&json!("Ania")));
        let u2 = table.get("u2").unwrap().unwrap();
        assert_eq!(u2.get("age"), Some(&json!(25)));
        assert_eq!(table.get("u3").unwrap(), None);

        table
            .put(
                &json!({ "id": "u4", "name": "Zosia", "age": 22, "email": "u4@x.com" }),
                false,
            )
            .unwrap();
        assert_eq!(table.get("u4").unwrap().unwrap().get("name"), Some(&json!("Zosia")));
        table.snapshot().unwrap();
        assert!(!table.needs_snapshot());
    }

    mod half_written_tail {
        use std::io::Write as _;

        use super::*;

        /// The bytes a writer killed mid-append leaves behind: a line with no closing newline.
        const TORN: &[u8] = br#"{"_k":"u3","_op":"put","_seq":9,"_ts":1,"name":"Half"#;

        fn seeded(path: &Path) -> u64 {
            let mut table = Table::open(users_schema(), path).unwrap();
            table
                .put(
                    &json!({ "id": "u1", "name": "Ania", "age": 30, "email": "u1@x.com" }),
                    false,
                )
                .unwrap();
            table
                .put(
                    &json!({ "id": "u2", "name": "Piotr", "age": 40, "email": "u2@x.com" }),
                    false,
                )
                .unwrap();
            table.flush().unwrap();
            fs::metadata(path).unwrap().len()
        }

        fn tear(path: &Path) {
            let mut file = std::fs::OpenOptions::new().append(true).open(path).unwrap();
            file.write_all(TORN).unwrap();
        }

        fn tracked_size(table: &Table) -> u64 {
            table.logs.log_of(UNPARTITIONED).unwrap().size()
        }

        #[test]
        fn a_writable_open_cuts_the_torn_tail_off_the_log_so_the_next_append_stays_readable() {
            let dir = tempdir().unwrap();
            let path = dir.path().join("users.jsonl");
            let healthy = seeded(&path);
            tear(&path);
            assert!(fs::metadata(&path).unwrap().len() > healthy);

            let mut table = Table::open(users_schema(), &path).unwrap();
            assert_eq!(tracked_size(&table), healthy);
            assert_eq!(fs::metadata(&path).unwrap().len(), healthy);

            table
                .put(
                    &json!({ "id": "u3", "name": "Zosia", "age": 22, "email": "u3@x.com" }),
                    false,
                )
                .unwrap();
            table.flush().unwrap();

            // NOTE: no snapshot on this path, so this reopen is the full rescan the tear used to break.
            let reopened = Table::open(users_schema(), &path).unwrap();
            let mut keys: Vec<&str> = reopened.keys().map(String::as_str).collect();
            keys.sort_unstable();
            assert_eq!(keys, ["u1", "u2", "u3"]);
            assert_eq!(reopened.get("u3").unwrap().unwrap().get("name"), Some(&json!("Zosia")));

            let raw = fs::read_to_string(&path).unwrap();
            let lines: Vec<&str> = raw.trim().lines().collect();
            assert_eq!(lines.len(), 3);
            assert!(lines.iter().all(|line| serde_json::from_str::<Value>(line).is_ok()));
        }

        #[test]
        fn a_readonly_open_moves_its_own_end_marker_and_leaves_the_torn_file_alone() {
            let dir = tempdir().unwrap();
            let path = dir.path().join("users.jsonl");
            let healthy = seeded(&path);
            tear(&path);
            let torn_size = fs::metadata(&path).unwrap().len();

            let (table, outcome) = Table::open_with_snapshot_and_declared_indexes(
                users_schema(),
                &path,
                dir.path().join("users.idx"),
                "",
                0,
                true,
                &[],
            )
            .unwrap();

            assert_eq!(outcome, RestoreOutcome::Rebuilt);
            assert_eq!(tracked_size(&table), healthy);
            assert_eq!(fs::metadata(&path).unwrap().len(), torn_size);
            let mut keys: Vec<&str> = table.keys().map(String::as_str).collect();
            keys.sort_unstable();
            assert_eq!(keys, ["u1", "u2"]);
        }
    }

    mod cypher {
        use super::*;
        use crate::crypto::{CipherOptions, MissingKeyPolicy};
        use std::fs;

        fn key(byte: u8) -> Vec<u8> {
            vec![byte; 32]
        }

        fn secrets_schema() -> TableSchema {
            parse_schema(&json!({
                "schemaVersion": 1,
                "table": "secrets",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "label": { "type": "string", "index": "exact" },
                    "token": { "type": "string", "cypher": true },
                    "pin": { "type": "int", "cypher": true },
                },
            }))
            .unwrap()
        }

        fn cipher_with(keys: &[(&str, u8)], on_missing_key: Option<MissingKeyPolicy>) -> FieldCipher {
            FieldCipher::new(CipherOptions {
                keys: keys.iter().map(|(id, byte)| (id.to_string(), key(*byte))).collect(),
                key_id: None,
                on_missing_key,
            })
            .unwrap()
        }

        fn open_secrets(dir: &std::path::Path, cipher: Option<FieldCipher>) -> Table {
            let mut table = Table::open(secrets_schema(), dir.join("secrets.jsonl")).unwrap();
            table.set_cipher(cipher);
            table
        }

        #[test]
        fn never_writes_the_plaintext_into_the_log() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            table
                .put(
                    &json!({ "id": "s1", "label": "deploy", "token": "hunter2-super-secret", "pin": 4321 }),
                    false,
                )
                .unwrap();

            let log = fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
            assert!(!log.contains("hunter2-super-secret"));
            assert!(!log.contains("4321"));
            assert!(log.contains("$cypher$v1$k1$"));
            // NOTE: the non-encrypted field is the control — it proves the log really holds this record.
            assert!(log.contains("deploy"));
        }

        #[test]
        fn hands_the_plaintext_back_on_read_with_its_type_intact() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            table
                .put(&json!({ "id": "s1", "token": "hunter2", "pin": 4321 }), false)
                .unwrap();

            let found = table.get("s1").unwrap().unwrap();
            assert_eq!(found.get("token"), Some(&json!("hunter2")));
            assert_eq!(found.get("pin"), Some(&json!(4321)));
        }

        #[test]
        fn a_staged_value_reads_back_plain_and_still_lands_sealed() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            let path = dir.path().join("secrets.jsonl");
            table
                .stage_put(&json!({ "id": "s1", "label": "deploy", "token": "top-secret" }), false)
                .unwrap();

            assert_eq!(
                table.get("s1").unwrap().unwrap().get("token"),
                Some(&json!("top-secret"))
            );
            assert!(
                fs::read_to_string(&path).unwrap_or_default().is_empty(),
                "nothing is on disk yet"
            );

            table.flush().unwrap();

            let written = fs::read_to_string(&path).unwrap();
            assert!(!written.contains("top-secret"), "the plaintext never reached the log");
            assert!(written.contains("$cypher$v1$k1$"));
            assert_eq!(
                table.get("s1").unwrap().unwrap().get("token"),
                Some(&json!("top-secret"))
            );
        }

        #[test]
        fn two_equal_values_get_two_different_envelopes() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            table.put(&json!({ "id": "s1", "token": "same" }), false).unwrap();
            table.put(&json!({ "id": "s2", "token": "same" }), false).unwrap();

            let lines: Vec<String> = fs::read_to_string(dir.path().join("secrets.jsonl"))
                .unwrap()
                .trim()
                .lines()
                .map(str::to_string)
                .collect();
            let envelopes: Vec<String> = lines
                .iter()
                .map(|line| {
                    serde_json::from_str::<Value>(line)
                        .unwrap()
                        .get("token")
                        .unwrap()
                        .as_str()
                        .unwrap()
                        .to_string()
                })
                .collect();
            assert_ne!(envelopes[0], envelopes[1]);
        }

        #[test]
        fn leaves_null_alone() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            table.put(&json!({ "id": "s1", "token": Value::Null }), false).unwrap();
            let log = fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
            assert!(log.contains("\"token\":null"));
        }

        #[test]
        fn survives_a_patch_of_an_encrypted_field() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            table
                .put(
                    &json!({ "id": "s1", "label": "deploy", "token": "first", "pin": 1 }),
                    false,
                )
                .unwrap();
            table.patch("s1", &json!({ "token": "second" }), false).unwrap();

            assert_eq!(table.get("s1").unwrap().unwrap().get("token"), Some(&json!("second")));
            assert_eq!(table.get("s1").unwrap().unwrap().get("pin"), Some(&json!(1)));
            let log = fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
            assert!(!log.contains("\"second\""));
        }

        #[test]
        fn survives_compaction_without_a_key() {
            let dir = tempdir().unwrap();
            {
                let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
                table.put(&json!({ "id": "s1", "token": "keep-me" }), false).unwrap();
                table
                    .put(&json!({ "id": "s1", "token": "keep-me-instead" }), false)
                    .unwrap();
            }

            {
                let mut blind = open_secrets(dir.path(), None);
                blind.compact().unwrap();
            }

            let reopened = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            assert_eq!(
                reopened.get("s1").unwrap().unwrap().get("token"),
                Some(&json!("keep-me-instead"))
            );
        }

        #[test]
        fn matches_a_predicate_on_an_encrypted_field_by_scanning() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            table.put(&json!({ "id": "s1", "token": "alpha" }), false).unwrap();
            table.put(&json!({ "id": "s2", "token": "beta" }), false).unwrap();

            let spec = QueryBuilder::new()
                .r#where("token", ComparisonOp::Eq, json!("beta"))
                .spec();
            let found = table.query(&spec).unwrap();
            assert_eq!(found.keys, vec!["s2".to_string()]);
        }

        #[test]
        fn refuses_to_write_an_encrypted_field_with_no_key() {
            let dir = tempdir().unwrap();
            let mut blind = open_secrets(dir.path(), None);
            let error = blind.put(&json!({ "id": "s9", "token": "nope" }), false).unwrap_err();
            assert!(matches!(error, TableError::Cipher(_)));
        }

        #[test]
        fn refuses_to_read_without_the_key_that_wrote_the_value() {
            let dir = tempdir().unwrap();
            {
                let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
                table.put(&json!({ "id": "s1", "token": "secret" }), false).unwrap();
            }

            let wrong = open_secrets(dir.path(), Some(cipher_with(&[("other", 2)], None)));
            let error = wrong.get("s1").unwrap_err();
            // The whole typed refusal reaches the caller, key id included — the read path used to
            // flatten it into an io::Error whose text was all that was left of it.
            let TableError::Cipher(cipher) = &error else {
                panic!("a failed decrypt is a cipher refusal, not an io failure: {error:?}");
            };
            assert_eq!(cipher.table, "secrets");
            assert_eq!(cipher.field, "token");
            assert_eq!(cipher.key_id.as_deref(), Some("k1"));
            assert!(error.to_string().contains("was not given"));
        }

        #[test]
        fn refuses_to_read_with_no_key_at_all() {
            let dir = tempdir().unwrap();
            {
                let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
                table.put(&json!({ "id": "s1", "token": "secret" }), false).unwrap();
            }

            let blind = open_secrets(dir.path(), None);
            let error = blind.get("s1").unwrap_err();
            assert!(error.to_string().contains("cannot be read without a key"));
        }

        #[test]
        fn omits_the_field_when_the_policy_says_so() {
            let dir = tempdir().unwrap();
            {
                let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
                table
                    .put(&json!({ "id": "s1", "label": "deploy", "token": "secret" }), false)
                    .unwrap();
            }

            let partial = open_secrets(
                dir.path(),
                Some(cipher_with(&[("other", 2)], Some(MissingKeyPolicy::Omit))),
            );
            let found = partial.get("s1").unwrap().unwrap();
            assert_eq!(found.get("label"), Some(&json!("deploy")));
            assert_eq!(found.get("token"), None);
        }

        #[test]
        fn hands_back_the_envelope_when_the_policy_says_raw() {
            let dir = tempdir().unwrap();
            {
                let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
                table.put(&json!({ "id": "s1", "token": "secret" }), false).unwrap();
            }

            let raw = open_secrets(
                dir.path(),
                Some(cipher_with(&[("other", 2)], Some(MissingKeyPolicy::Raw))),
            );
            let found = raw.get("s1").unwrap().unwrap();
            assert!(found.get("token").unwrap().as_str().unwrap().contains("$cypher$v1$k1$"));
        }

        const MASQUERADE: &str = "$cypher$v1$k1$not$real";

        fn token_in(path: &std::path::Path, key: &str) -> String {
            fs::read_to_string(path)
                .unwrap()
                .lines()
                .filter_map(|line| serde_json::from_str::<Value>(line).ok())
                .rfind(|entry| entry["_k"] == json!(key))
                .unwrap()["token"]
                .as_str()
                .unwrap()
                .to_string()
        }

        #[test]
        fn a_value_that_only_looks_like_an_envelope_is_sealed_like_any_other() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));

            table
                .put(&json!({ "id": "s1", "label": "deploy", "token": MASQUERADE }), false)
                .unwrap();

            let path = dir.path().join("secrets.jsonl");
            let stored = token_in(&path, "s1");
            assert_ne!(stored, MASQUERADE, "the value reached the log in the clear");
            assert!(!fs::read_to_string(&path).unwrap().contains("$not$real"));
            assert_eq!(
                table.get("s1").unwrap().unwrap().get("token"),
                Some(&json!(MASQUERADE)),
                "and it still reads back as what was written"
            );
        }

        #[test]
        fn a_copy_between_databases_carries_the_envelope_byte_for_byte() {
            let source = tempdir().unwrap();
            {
                let mut table = open_secrets(source.path(), Some(cipher_with(&[("k1", 1)], None)));
                table
                    .put(&json!({ "id": "s1", "label": "deploy", "token": "hunter2" }), false)
                    .unwrap();
            }
            let envelope = {
                let raw = open_secrets(
                    source.path(),
                    Some(cipher_with(&[("other", 2)], Some(MissingKeyPolicy::Raw))),
                );
                raw.get("s1").unwrap().unwrap()["token"].as_str().unwrap().to_string()
            };

            // The restore side holds no key at all — it moves ciphertext it cannot read.
            let target = tempdir().unwrap();
            {
                let mut table = open_secrets(target.path(), None);
                table
                    .put_sealed(
                        &json!({ "id": "s1", "label": "deploy", "token": envelope }),
                        &["token".to_string()],
                        false,
                    )
                    .unwrap();
            }

            assert_eq!(
                token_in(&target.path().join("secrets.jsonl"), "s1"),
                envelope,
                "the copy re-sealed a value it was told was already sealed"
            );
            let restored = open_secrets(target.path(), Some(cipher_with(&[("k1", 1)], None)));
            assert_eq!(
                restored.get("s1").unwrap().unwrap().get("token"),
                Some(&json!("hunter2"))
            );
        }

        #[test]
        fn a_carried_field_is_refused_when_it_is_not_an_envelope() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));

            let error = table
                .put_sealed(
                    &json!({ "id": "s1", "token": MASQUERADE }),
                    &["token".to_string()],
                    false,
                )
                .unwrap_err();

            assert!(error.to_string().contains("is not a cypher envelope"), "{error}");
            assert!(table.is_empty(), "the refused write still landed");
        }

        #[test]
        fn a_carried_field_is_refused_when_the_envelope_belongs_to_another_record() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            table.put(&json!({ "id": "s2", "token": "yours" }), false).unwrap();
            let stolen = {
                let raw = open_secrets(
                    dir.path(),
                    Some(cipher_with(&[("other", 2)], Some(MissingKeyPolicy::Raw))),
                );
                raw.get("s2").unwrap().unwrap()["token"].as_str().unwrap().to_string()
            };

            let error = table
                .put_sealed(&json!({ "id": "s1", "token": stolen }), &["token".to_string()], false)
                .unwrap_err();

            assert!(
                error.to_string().contains("does not open for this record and field"),
                "{error}"
            );
        }

        #[test]
        fn a_field_that_is_not_declared_cypher_cannot_be_carried() {
            let dir = tempdir().unwrap();
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));

            let error = table
                .put_sealed(&json!({ "id": "s1", "label": "deploy" }), &["label".to_string()], false)
                .unwrap_err();

            assert!(error.to_string().contains("not declared \"cypher\""), "{error}");
        }

        #[test]
        fn a_carried_field_does_not_spare_another_write_in_the_same_window() {
            let source = tempdir().unwrap();
            {
                let mut table = open_secrets(source.path(), Some(cipher_with(&[("k1", 1)], None)));
                table.put(&json!({ "id": "s1", "token": "hunter2" }), false).unwrap();
            }
            let envelope = token_in(&source.path().join("secrets.jsonl"), "s1");

            let target = tempdir().unwrap();
            let mut table = open_secrets(target.path(), Some(cipher_with(&[("k1", 1)], None)));
            table
                .stage_put(&json!({ "id": "s2", "token": MASQUERADE }), false)
                .unwrap();
            table
                .put_sealed(
                    &json!({ "id": "s1", "token": envelope.clone() }),
                    &["token".to_string()],
                    false,
                )
                .unwrap();

            let path = target.path().join("secrets.jsonl");
            assert_eq!(token_in(&path, "s1"), envelope);
            assert_ne!(
                token_in(&path, "s2"),
                MASQUERADE,
                "the import's claim spared a record it was never about"
            );
        }

        #[test]
        fn refuses_a_value_moved_to_another_record() {
            let dir = tempdir().unwrap();
            {
                let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
                table.put(&json!({ "id": "s1", "token": "mine" }), false).unwrap();
                table.put(&json!({ "id": "s2", "token": "yours" }), false).unwrap();
            }

            let path = dir.path().join("secrets.jsonl");
            let lines: Vec<String> = fs::read_to_string(&path)
                .unwrap()
                .trim()
                .lines()
                .map(str::to_string)
                .collect();
            let mut first: serde_json::Map<String, Value> = serde_json::from_str(&lines[0]).unwrap();
            let mut second: serde_json::Map<String, Value> = serde_json::from_str(&lines[1]).unwrap();
            let stolen = first.remove("token").unwrap();
            second.insert("token".to_string(), stolen);
            fs::write(
                &path,
                format!(
                    "{}\n{}\n",
                    serde_json::to_string(&first).unwrap(),
                    serde_json::to_string(&second).unwrap()
                ),
            )
            .unwrap();

            let reopened = open_secrets(dir.path(), Some(cipher_with(&[("k1", 1)], None)));
            let error = reopened.get("s2").unwrap_err();
            assert!(error.to_string().contains("authentication tag"));
        }

        /// Several keys on the ring, `write_id` the one new values are sealed with — the shape a
        /// rotation is run under, which `cipher_with` cannot express (it leaves `keyId` unset).
        fn cipher_writing(
            keys: &[(&str, u8)],
            write_id: &str,
            on_missing_key: Option<MissingKeyPolicy>,
        ) -> FieldCipher {
            FieldCipher::new(CipherOptions {
                keys: keys.iter().map(|(id, byte)| ((*id).to_string(), key(*byte))).collect(),
                key_id: Some(write_id.to_string()),
                on_missing_key,
            })
            .unwrap()
        }

        /// One record sealed under `k1`, then a handle whose write key is `k2` and which writes a
        /// second record — the "one old, one new" table every rotation test starts from.
        fn seeded_across_two_keys(dir: &std::path::Path) -> Table {
            let mut old = open_secrets(dir, Some(cipher_with(&[("k1", 1)], None)));
            old.put(
                &json!({ "id": "s1", "label": "old", "token": "first", "pin": 1111 }),
                false,
            )
            .unwrap();
            drop(old);

            let mut table = open_secrets(dir, Some(cipher_writing(&[("k1", 1), ("k2", 2)], "k2", None)));
            table
                .put(
                    &json!({ "id": "s2", "label": "new", "token": "second", "pin": 2222 }),
                    false,
                )
                .unwrap();
            table
        }

        fn envelope_key_ids_in(path: &std::path::Path) -> Vec<String> {
            fs::read_to_string(path)
                .unwrap()
                .lines()
                .filter(|line| !line.is_empty())
                .map(|line| {
                    let record: serde_json::Map<String, Value> = serde_json::from_str(line).unwrap();
                    crate::crypto::envelope_key_id(&record["token"]).unwrap().to_string()
                })
                .collect()
        }

        #[test]
        fn rotation_re_encrypts_only_the_record_still_on_the_retired_key() {
            let dir = tempdir().unwrap();
            let mut table = seeded_across_two_keys(dir.path());

            let result = table.rotate_cypher_keys().unwrap();

            assert_eq!(
                (result.rotated, result.already_current, result.skipped),
                (1, 1, 0),
                "only \"s1\" was on k1"
            );
            let written = envelope_key_ids_in(&dir.path().join("secrets.jsonl"));
            assert_eq!(
                written,
                vec!["k1".to_string(), "k2".to_string(), "k2".to_string()],
                "the k1 line stays in the append-only log; the rotation appends the k2 version after it"
            );
        }

        #[test]
        fn a_second_rotation_reports_nothing_left_to_do() {
            let dir = tempdir().unwrap();
            let mut table = seeded_across_two_keys(dir.path());
            table.rotate_cypher_keys().unwrap();

            let again = table.rotate_cypher_keys().unwrap();

            assert_eq!(
                (again.rotated, again.skipped),
                (0, 0),
                "a rotation nothing needs must rewrite nothing"
            );
            assert_eq!(again.already_current, 2);
        }

        #[test]
        fn every_value_still_reads_back_as_itself_after_a_rotation() {
            let dir = tempdir().unwrap();
            let mut table = seeded_across_two_keys(dir.path());

            table.rotate_cypher_keys().unwrap();

            for (key, token, pin, label) in [("s1", "first", 1111, "old"), ("s2", "second", 2222, "new")] {
                let found = table.get(key).unwrap().unwrap();
                assert_eq!(found.get("token"), Some(&json!(token)), "{key}");
                assert_eq!(found.get("pin"), Some(&json!(pin)), "{key}");
                assert_eq!(found.get("label"), Some(&json!(label)), "{key}");
            }
        }

        #[test]
        fn a_rotated_record_reads_back_with_the_new_key_alone() {
            let dir = tempdir().unwrap();
            let mut table = seeded_across_two_keys(dir.path());
            table.rotate_cypher_keys().unwrap();
            drop(table);

            let without_the_old_key = open_secrets(dir.path(), Some(cipher_with(&[("k2", 2)], None)));

            assert_eq!(
                without_the_old_key.get("s1").unwrap().unwrap().get("token"),
                Some(&json!("first")),
                "the whole point of rotating: k1 can now be retired"
            );
        }

        #[test]
        fn a_table_with_no_cypher_field_answers_zero_without_reading_anything() {
            let dir = tempdir().unwrap();
            let mut table = Table::open(users_schema(), dir.path().join("users.jsonl")).unwrap();
            table
                .put(
                    &json!({ "id": "u1", "name": "Ania", "age": 30, "email": "u1@x.com" }),
                    false,
                )
                .unwrap();

            let result = table.rotate_cypher_keys().unwrap();

            assert_eq!(result, RotationResult::default());
        }

        #[test]
        fn a_missing_key_refuses_the_run_under_the_default_policy() {
            let dir = tempdir().unwrap();
            drop(seeded_across_two_keys(dir.path()));
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k2", 2)], None)));

            let error = table.rotate_cypher_keys().unwrap_err();

            assert!(
                error.to_string().contains("s1"),
                "the refusal has to name the record: {error}"
            );
            assert!(error.to_string().contains("k1"), "and the key it wanted: {error}");
        }

        #[test]
        fn a_missing_key_is_skipped_and_counted_when_the_policy_says_so() {
            let dir = tempdir().unwrap();
            drop(seeded_across_two_keys(dir.path()));
            let mut table = open_secrets(dir.path(), Some(cipher_with(&[("k2", 2)], Some(MissingKeyPolicy::Raw))));

            let result = table.rotate_cypher_keys().unwrap();

            assert_eq!(
                (result.rotated, result.already_current, result.skipped),
                (0, 1, 1),
                "the unreadable record is left alone rather than re-sealed from its own envelope"
            );
            let written = envelope_key_ids_in(&dir.path().join("secrets.jsonl"));
            assert_eq!(
                written,
                vec!["k1".to_string(), "k2".to_string()],
                "nothing was appended"
            );
        }

        #[test]
        fn rotation_reaches_every_partition_of_a_partitioned_table() {
            let schema = parse_schema(&json!({
                "schemaVersion": 1,
                "table": "secrets",
                "primaryKey": "id",
                "partitionBy": "tenant",
                "fields": {
                    "id": { "type": "string" },
                    "tenant": { "type": "string", "required": true },
                    "token": { "type": "string", "cypher": true },
                },
            }))
            .unwrap();
            let dir = tempdir().unwrap();
            let path = dir.path().join("secrets.jsonl");
            let open = |cipher| {
                let mut table = Table::open(schema.clone(), &path).unwrap();
                table.set_cipher(Some(cipher));
                table
            };
            let mut old = open(cipher_with(&[("k1", 1)], None));
            old.put(&json!({ "id": "s1", "tenant": "acme", "token": "a" }), false)
                .unwrap();
            old.put(&json!({ "id": "s2", "tenant": "globex", "token": "b" }), false)
                .unwrap();
            drop(old);
            let mut table = open(cipher_writing(&[("k1", 1), ("k2", 2)], "k2", None));

            let result = table.rotate_cypher_keys().unwrap();

            assert_eq!((result.rotated, result.already_current), (2, 0));
            for (key, tenant, token) in [("s1", "acme", "a"), ("s2", "globex", "b")] {
                let found = table.get(key).unwrap().unwrap();
                assert_eq!(found.get("token"), Some(&json!(token)));
                assert_eq!(
                    found.get("tenant"),
                    Some(&json!(tenant)),
                    "the record must not change partition"
                );
            }
            for tenant in ["acme", "globex"] {
                let log = fs::read_to_string(dir.path().join("secrets").join(format!("{tenant}.jsonl"))).unwrap();
                assert!(log.contains("$cypher$v1$k2$"), "{tenant} was rotated in its own file");
            }
        }

        #[test]
        fn rotation_is_refused_on_a_readonly_handle() {
            let dir = tempdir().unwrap();
            drop(seeded_across_two_keys(dir.path()));
            let mut table = Table::open_with_snapshot_and_declared_indexes(
                secrets_schema(),
                dir.path().join("secrets.jsonl"),
                dir.path().join("secrets.idx"),
                "",
                0,
                true,
                &[],
            )
            .unwrap()
            .0;
            table.set_cipher(Some(cipher_writing(&[("k1", 1), ("k2", 2)], "k2", None)));

            let error = table.rotate_cypher_keys().unwrap_err();

            assert!(error.to_string().contains("readonly"), "{error}");
        }
    }

    mod cypher_txn {
        use super::*;
        use crate::crypto::CipherOptions;
        use std::collections::HashMap;
        use std::fs;

        fn key(byte: u8) -> Vec<u8> {
            vec![byte; 32]
        }

        fn secrets_schema() -> TableSchema {
            parse_schema(&json!({
                "schemaVersion": 1,
                "table": "secrets",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "token": { "type": "string", "cypher": true },
                },
            }))
            .unwrap()
        }

        #[test]
        fn prepare_txn_ops_seals_the_line_but_keeps_the_prepared_record_plain() {
            let dir = tempdir().unwrap();
            let cipher = FieldCipher::new(CipherOptions {
                keys: HashMap::from([("k1".to_string(), key(1))]),
                key_id: None,
                on_missing_key: None,
            })
            .unwrap();
            let mut table = Table::open(secrets_schema(), dir.path().join("secrets.jsonl")).unwrap();
            table.set_cipher(Some(cipher));

            let prepared = table
                .prepare_txn_ops(&[TxnOp::Put(json!({ "id": "s1", "token": "top-secret" }))])
                .unwrap();
            assert_eq!(prepared.len(), 1);
            assert!(prepared[0].line.contains("$cypher$v1$k1$"));
            assert!(!prepared[0].line.contains("top-secret"));
            assert_eq!(
                prepared[0].record.as_ref().unwrap().get("token"),
                Some(&json!("top-secret"))
            );

            table.apply_txn_entries(prepared).unwrap();
            let log = fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
            assert!(!log.contains("top-secret"));
            assert_eq!(
                table.get("s1").unwrap().unwrap().get("token"),
                Some(&json!("top-secret"))
            );
        }
    }

    /// Ported one for one from `packages/hayson/test/partition.test.ts` — the TS engine's contract
    /// for `partitionBy` is the specification this port is measured against.
    mod partitions {
        use std::fs;

        use super::*;
        use crate::query::predicates::ComparisonOp;
        use crate::registry::fulltext_index::SearchMode;

        fn notes_schema(partitioned: bool) -> TableSchema {
            let mut spec = json!({
                "schemaVersion": 1,
                "table": "notes",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "kind": { "type": "string", "required": true, "index": "exact" },
                    "title": { "type": "string", "required": true, "fulltext": true },
                    "body": { "type": "string", "fulltext": true },
                    "slug": { "type": "string", "unique": true },
                },
            });
            if partitioned {
                spec.as_object_mut()
                    .unwrap()
                    .insert("partitionBy".into(), json!("kind"));
            }
            parse_schema(&spec).unwrap()
        }

        fn seed() -> Vec<Value> {
            vec![
                json!({ "id": "n1", "kind": "note", "title": "Backend developer", "body": "silnik zapytań", "slug": "n1" }),
                json!({ "id": "n2", "kind": "note", "title": "Frontend notes", "body": "react", "slug": "n2" }),
                json!({ "id": "a1", "kind": "adr", "title": "Backend decision", "body": "silnik składowania", "slug": "a1" }),
                json!({ "id": "r1", "kind": "rule", "title": "Naming rule", "body": "kebab case", "slug": "r1" }),
            ]
        }

        fn log_path(dir: &std::path::Path) -> std::path::PathBuf {
            dir.join("notes.jsonl")
        }

        fn open_notes(dir: &std::path::Path) -> Table {
            Table::open(notes_schema(true), log_path(dir)).unwrap()
        }

        fn partition_lines(dir: &std::path::Path, part: &str) -> Vec<Value> {
            fs::read_to_string(dir.join("notes").join(format!("{part}.jsonl")))
                .unwrap()
                .trim()
                .lines()
                .map(|line| serde_json::from_str(line).unwrap())
                .collect()
        }

        #[test]
        fn keeps_one_log_file_per_partition_value() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            for note in seed() {
                notes.put(&note, false).unwrap();
            }

            assert!(!log_path(dir.path()).exists());
            assert_eq!(partition_lines(dir.path(), "note").len(), 2);
            assert_eq!(partition_lines(dir.path(), "adr").len(), 1);
            assert_eq!(partition_lines(dir.path(), "rule").len(), 1);
            for line in partition_lines(dir.path(), "adr") {
                assert_eq!(line.get("kind"), Some(&json!("adr")));
            }
        }

        /// A live key cannot change partition (`queue::assert_partition_stable` refuses it), so the
        /// only way its history spans two files is delete-then-write-elsewhere — and then the
        /// tombstone sits in the old file while the new version sits in the new one. Sorting by
        /// `seq` is what puts the two halves back together.
        #[test]
        fn history_of_a_key_that_moved_partitions_reads_as_one_list_across_both_files() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            notes
                .put(
                    &json!({ "id": "n1", "kind": "note", "title": "First", "slug": "n1" }),
                    false,
                )
                .unwrap();
            notes.patch("n1", &json!({ "title": "Second" }), false).unwrap();
            let moved = notes.put(
                &json!({ "id": "n1", "kind": "adr", "title": "Third", "slug": "n1" }),
                false,
            );
            assert!(moved.is_err(), "a live key may not change partition");
            notes.delete("n1", false).unwrap();
            notes
                .put(
                    &json!({ "id": "n1", "kind": "adr", "title": "Third", "slug": "n1" }),
                    false,
                )
                .unwrap();

            let history = notes.history("n1").unwrap();

            assert_eq!(
                history.scanned,
                vec![
                    dir.path().join("notes").join("adr.jsonl"),
                    dir.path().join("notes").join("note.jsonl"),
                ]
            );
            let shape: Vec<(u64, LogOp, &str, Option<&Value>)> = history
                .entries
                .iter()
                .map(|entry| {
                    (
                        entry.seq,
                        entry.op,
                        entry.partition.as_str(),
                        entry.document.as_ref().and_then(|d| d.get("title")),
                    )
                })
                .collect();
            assert_eq!(
                shape,
                vec![
                    (1, LogOp::Put, "note", Some(&json!("First"))),
                    (2, LogOp::Put, "note", Some(&json!("Second"))),
                    (3, LogOp::Del, "note", None),
                    (4, LogOp::Put, "adr", Some(&json!("Third"))),
                ]
            );
            assert!(notes.history("n2").unwrap().entries.is_empty());
        }

        /// A table whose `partitionBy` was declared but whose split has not run yet still keeps
        /// everything in the flat log, and a readonly handle reads it there — `open_logs` says so,
        /// and history has to follow it rather than look in a directory that does not exist.
        #[test]
        fn history_of_a_declared_but_unsplit_table_reads_the_flat_log() {
            let dir = tempdir().unwrap();
            {
                let mut flat = Table::open(notes_schema(false), log_path(dir.path())).unwrap();
                flat.put(
                    &json!({ "id": "n1", "kind": "note", "title": "First", "slug": "n1" }),
                    false,
                )
                .unwrap();
                flat.patch("n1", &json!({ "title": "Second" }), false).unwrap();
            }

            let (notes, _) = Table::open_with_snapshot_and_declared_indexes(
                notes_schema(true),
                log_path(dir.path()),
                dir.path().join("notes.idx"),
                "hash",
                0,
                true,
                &[],
            )
            .unwrap();

            let history = notes.history("n1").unwrap();

            assert_eq!(history.scanned, vec![log_path(dir.path())]);
            assert_eq!(
                history
                    .entries
                    .iter()
                    .map(|entry| (entry.seq, entry.partition.clone()))
                    .collect::<Vec<_>>(),
                vec![(1, UNPARTITIONED.to_string()), (2, UNPARTITIONED.to_string())]
            );
        }

        #[test]
        fn reads_every_partition_back_after_a_restart() {
            let dir = tempdir().unwrap();
            {
                let mut notes = open_notes(dir.path());
                for note in seed() {
                    notes.put(&note, false).unwrap();
                }
            }

            let notes = open_notes(dir.path());
            assert_eq!(notes.registry.records(), 4);
            assert_eq!(
                notes.get("a1").unwrap().unwrap().get("title"),
                Some(&json!("Backend decision"))
            );
            assert_eq!(notes.get("r1").unwrap().unwrap().get("kind"), Some(&json!("rule")));

            let spec = QueryBuilder::new()
                .r#where("kind", ComparisonOp::Eq, json!("note"))
                .spec();
            let mut found = notes.query(&spec).unwrap().keys;
            found.sort();
            assert_eq!(found, vec!["n1".to_string(), "n2".to_string()]);
        }

        #[test]
        fn pairs_the_index_snapshot_with_every_partition_it_covers() {
            let dir = tempdir().unwrap();
            let snapshot_path = dir.path().join("notes.idx");
            let (mut notes, _) =
                Table::open_with_snapshot(notes_schema(true), log_path(dir.path()), &snapshot_path, "hash", 0).unwrap();
            for note in seed() {
                notes.put(&note, false).unwrap();
            }
            notes.snapshot().unwrap();

            let snapshot = crate::checkpoint::read_snapshot(&snapshot_path).unwrap().unwrap();
            let parts = snapshot.header.parts.unwrap();
            assert_eq!(
                parts.iter().map(|part| part.part.clone()).collect::<Vec<_>>(),
                vec!["adr".to_string(), "note".to_string(), "rule".to_string()]
            );
            for part in &parts {
                let on_disk = fs::metadata(dir.path().join("notes").join(format!("{}.jsonl", part.part))).unwrap();
                assert_eq!(part.watermark, on_disk.len());
                assert!(!part.log_hash.is_empty());
            }
            assert_eq!(snapshot.header.log_hash, None);
        }

        #[test]
        fn compaction_repairs_every_partition_watermark_in_the_snapshot() {
            let dir = tempdir().unwrap();
            let snapshot_path = dir.path().join("notes.idx");
            let (mut notes, _) =
                Table::open_with_snapshot(notes_schema(true), log_path(dir.path()), &snapshot_path, "hash", 0).unwrap();
            for note in seed() {
                notes.put(&note, false).unwrap();
            }
            notes
                .put(&json!({ "id": "n1", "kind": "note", "title": "Rewritten" }), false)
                .unwrap();
            notes.delete("r1", false).unwrap();
            notes.snapshot().unwrap();

            notes.compact().unwrap();

            let snapshot = crate::checkpoint::read_snapshot(&snapshot_path).unwrap().unwrap();
            for part in snapshot.header.parts.unwrap() {
                let on_disk = fs::metadata(dir.path().join("notes").join(format!("{}.jsonl", part.part))).unwrap();
                assert_eq!(part.watermark, on_disk.len(), "partition \"{}\" watermark", part.part);
                assert_eq!(part.dead_bytes, 0, "partition \"{}\" dead bytes", part.part);
            }
            assert_eq!(snapshot.header.watermark, notes.logs.total_size());

            let (reopened, outcome) =
                Table::open_with_snapshot(notes_schema(true), log_path(dir.path()), &snapshot_path, "hash", 0).unwrap();
            assert_eq!(outcome, RestoreOutcome::Restored);
            assert_eq!(reopened.registry.records(), 3);
            assert_eq!(
                reopened.get("n1").unwrap().unwrap().get("title"),
                Some(&json!("Rewritten"))
            );
            assert_eq!(reopened.get("r1").unwrap(), None);
        }

        #[test]
        fn restores_the_registry_from_a_snapshot_that_covers_every_partition() {
            let dir = tempdir().unwrap();
            let snapshot_path = dir.path().join("notes.idx");
            {
                let (mut notes, _) =
                    Table::open_with_snapshot(notes_schema(true), log_path(dir.path()), &snapshot_path, "hash", 0)
                        .unwrap();
                for note in seed() {
                    notes.put(&note, false).unwrap();
                }
                notes.snapshot().unwrap();
            }

            let (notes, outcome) =
                Table::open_with_snapshot(notes_schema(true), log_path(dir.path()), &snapshot_path, "hash", 0).unwrap();
            assert_eq!(outcome, RestoreOutcome::Restored);
            assert_eq!(notes.registry.records(), 4);
            assert_eq!(
                notes.get("a1").unwrap().unwrap().get("title"),
                Some(&json!("Backend decision"))
            );
        }

        #[test]
        fn rebuilds_when_one_partition_grew_after_the_snapshot_was_taken() {
            let dir = tempdir().unwrap();
            let snapshot_path = dir.path().join("notes.idx");
            {
                let (mut notes, _) =
                    Table::open_with_snapshot(notes_schema(true), log_path(dir.path()), &snapshot_path, "hash", 0)
                        .unwrap();
                for note in seed() {
                    notes.put(&note, false).unwrap();
                }
                notes.snapshot().unwrap();
                notes
                    .put(
                        &json!({ "id": "n3", "kind": "note", "title": "after the snapshot" }),
                        false,
                    )
                    .unwrap();
            }

            let (notes, outcome) =
                Table::open_with_snapshot(notes_schema(true), log_path(dir.path()), &snapshot_path, "hash", 0).unwrap();
            assert_eq!(outcome, RestoreOutcome::Rebuilt);
            assert_eq!(notes.registry.records(), 5);
            assert_eq!(
                notes.get("n3").unwrap().unwrap().get("title"),
                Some(&json!("after the snapshot"))
            );
        }

        #[test]
        fn holds_uniqueness_across_partitions_not_just_inside_one() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            notes
                .put(
                    &json!({ "id": "n1", "kind": "note", "title": "first", "slug": "shared" }),
                    false,
                )
                .unwrap();

            let error = notes
                .put(
                    &json!({ "id": "a1", "kind": "adr", "title": "second", "slug": "shared" }),
                    false,
                )
                .unwrap_err();
            assert!(matches!(error, TableError::Unique(_)));
        }

        #[test]
        fn refuses_a_partition_value_that_cannot_name_a_file() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            for kind in ["../escape", "Upper", "has space", ""] {
                let error = notes
                    .put(&json!({ "id": "x", "kind": kind, "title": "nope" }), false)
                    .unwrap_err();
                assert!(
                    matches!(error, TableError::Partition(_)),
                    "kind {kind:?} must be refused"
                );
            }
            assert!(!dir.path().join("notes").exists());
        }

        #[test]
        fn refuses_to_move_a_record_between_partitions() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            notes
                .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
                .unwrap();

            let put_error = notes
                .put(&json!({ "id": "n1", "kind": "adr", "title": "moved" }), false)
                .unwrap_err();
            assert!(matches!(put_error, TableError::Partition(_)));
            let patch_error = notes.patch("n1", &json!({ "kind": "adr" }), false).unwrap_err();
            assert!(matches!(patch_error, TableError::Partition(_)));

            assert_eq!(notes.get("n1").unwrap().unwrap().get("kind"), Some(&json!("note")));
            assert!(!dir.path().join("notes/adr.jsonl").exists());
        }

        #[test]
        fn refuses_a_staged_write_that_would_move_a_key_the_window_already_placed() {
            let dir = tempdir().unwrap();
            let mut table = open_notes(dir.path());
            table
                .stage_put(
                    &json!({ "id": "n1", "kind": "note", "title": "Backend", "slug": "n1" }),
                    false,
                )
                .unwrap();

            let refused = table
                .stage_put(
                    &json!({ "id": "n1", "kind": "adr", "title": "Backend", "slug": "n1" }),
                    true,
                )
                .unwrap_err();

            assert!(matches!(refused, TableError::Partition(_)));
            assert!(
                refused.to_string().contains("already lives in partition \"note\""),
                "the partition came from the window, which the registry knows nothing about: {refused}"
            );
        }

        #[test]
        fn rewrites_a_record_inside_its_own_partition() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            notes
                .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
                .unwrap();
            notes.patch("n1", &json!({ "title": "second" }), false).unwrap();

            assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("second")));
            assert_eq!(partition_lines(dir.path(), "note").len(), 2);
        }

        #[test]
        fn puts_a_tombstone_in_the_partition_the_record_lived_in() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            notes
                .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
                .unwrap();
            notes
                .put(&json!({ "id": "a1", "kind": "adr", "title": "kept" }), false)
                .unwrap();

            assert!(notes.delete("n1", false).unwrap());
            assert!(!notes.delete("nope", false).unwrap());

            let ops: Vec<Value> = partition_lines(dir.path(), "note")
                .iter()
                .map(|line| line.get("_op").unwrap().clone())
                .collect();
            assert_eq!(ops, vec![json!("put"), json!("del")]);
            assert_eq!(partition_lines(dir.path(), "adr").len(), 1);
        }

        #[test]
        fn lands_a_unit_of_work_that_spans_two_partitions_of_one_table() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            let prepared = notes
                .prepare_txn_ops(&[
                    TxnOp::Put(json!({ "id": "n1", "kind": "note", "title": "from unit" })),
                    TxnOp::Put(json!({ "id": "a1", "kind": "adr", "title": "from unit too" })),
                ])
                .unwrap();
            assert_eq!(prepared.len(), 2);
            notes.apply_txn_entries(prepared).unwrap();

            assert_eq!(
                notes.get("n1").unwrap().unwrap().get("title"),
                Some(&json!("from unit"))
            );
            assert_eq!(
                notes.get("a1").unwrap().unwrap().get("title"),
                Some(&json!("from unit too"))
            );
            assert_eq!(partition_lines(dir.path(), "note").len(), 1);
            assert_eq!(partition_lines(dir.path(), "adr").len(), 1);
        }

        #[test]
        fn refuses_a_unit_of_work_that_would_move_a_record_between_partitions() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            notes
                .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
                .unwrap();

            let outcome = notes.prepare_txn_ops(&[TxnOp::Put(json!({ "id": "n1", "kind": "adr", "title": "moved" }))]);
            assert!(matches!(outcome, Err(TableError::Partition(_))));
            assert_eq!(notes.get("n1").unwrap().unwrap().get("kind"), Some(&json!("note")));
        }

        #[test]
        fn compacts_one_partition_and_leaves_the_others_byte_for_byte() {
            let dir = tempdir().unwrap();
            let mut notes = open_notes(dir.path());
            notes
                .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
                .unwrap();
            notes
                .put(&json!({ "id": "a1", "kind": "adr", "title": "cold" }), false)
                .unwrap();
            for round in 0..8 {
                notes
                    .patch("n1", &json!({ "title": format!("round {round}") }), false)
                    .unwrap();
            }

            let cold_before = fs::read_to_string(dir.path().join("notes/adr.jsonl")).unwrap();
            assert_eq!(partition_lines(dir.path(), "note").len(), 9);

            let result = notes.compact().unwrap();
            assert_eq!(result.records, 2);
            assert_eq!(partition_lines(dir.path(), "note").len(), 1);
            assert_eq!(
                fs::read_to_string(dir.path().join("notes/adr.jsonl")).unwrap(),
                cold_before
            );
            assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("round 7")));
            assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("cold")));
        }

        fn seeded_flat(dir: &std::path::Path) -> Table {
            let mut notes = Table::open(notes_schema(false), log_path(dir)).unwrap();
            for note in seed() {
                notes.put(&note, false).unwrap();
            }
            notes.delete("n2", false).unwrap();
            notes
        }

        #[test]
        fn moves_the_live_records_into_one_file_per_partition_when_the_schema_starts_partitioning() {
            let dir = tempdir().unwrap();
            drop(seeded_flat(dir.path()));

            let notes = open_notes(dir.path());
            assert_eq!(notes.registry.records(), 3);
            assert_eq!(notes.get("n2").unwrap(), None);
            assert_eq!(
                notes.get("a1").unwrap().unwrap().get("title"),
                Some(&json!("Backend decision"))
            );

            assert!(!log_path(dir.path()).exists());
            assert_eq!(partition_lines(dir.path(), "note").len(), 1);
            assert_eq!(partition_lines(dir.path(), "adr").len(), 1);
            assert_eq!(partition_lines(dir.path(), "rule").len(), 1);
        }

        fn open_notes_readonly(dir: &std::path::Path) -> (Table, RestoreOutcome) {
            Table::open_with_snapshot_and_declared_indexes(
                notes_schema(true),
                log_path(dir),
                dir.join("notes.idx"),
                "hash",
                0,
                true,
                &[],
            )
            .unwrap()
        }

        #[test]
        fn a_readonly_open_reads_the_flat_log_the_move_has_not_touched_yet() {
            let dir = tempdir().unwrap();
            drop(seeded_flat(dir.path()));
            let before = fs::read_to_string(log_path(dir.path())).unwrap();

            let (notes, _) = open_notes_readonly(dir.path());

            assert_eq!(notes.registry.records(), 3);
            assert_eq!(notes.get("n2").unwrap(), None);
            assert_eq!(
                notes.get("a1").unwrap().unwrap().get("title"),
                Some(&json!("Backend decision"))
            );
            assert_eq!(
                fs::read_to_string(log_path(dir.path())).unwrap(),
                before,
                "a readonly open must leave the flat log byte for byte as it found it"
            );
            assert!(
                !dir.path().join("notes").exists(),
                "a readonly open must not create the partition directory"
            );
        }

        #[test]
        fn a_readonly_open_of_the_unmoved_flat_log_writes_no_index_snapshot() {
            let dir = tempdir().unwrap();
            drop(seeded_flat(dir.path()));

            let (notes, outcome) = open_notes_readonly(dir.path());

            assert_eq!(outcome, RestoreOutcome::Rebuilt);
            assert_eq!(notes.registry.records(), 3);
            assert!(
                !dir.path().join("notes.idx").exists(),
                "a readonly open must not write the index snapshot"
            );
        }

        #[test]
        fn a_writer_still_finishes_the_move_a_readonly_open_left_undone() {
            let dir = tempdir().unwrap();
            drop(seeded_flat(dir.path()));
            drop(open_notes_readonly(dir.path()));

            let notes = open_notes(dir.path());

            assert_eq!(notes.registry.records(), 3);
            assert_eq!(
                notes.get("a1").unwrap().unwrap().get("title"),
                Some(&json!("Backend decision"))
            );
            assert!(!log_path(dir.path()).exists());
            assert_eq!(partition_lines(dir.path(), "note").len(), 1);
            assert_eq!(partition_lines(dir.path(), "adr").len(), 1);
            assert_eq!(partition_lines(dir.path(), "rule").len(), 1);
        }

        #[test]
        fn a_readonly_open_reads_the_partitions_once_the_move_has_run() {
            let dir = tempdir().unwrap();
            drop(seeded_flat(dir.path()));
            drop(open_notes(dir.path()));

            let (notes, _) = open_notes_readonly(dir.path());

            assert!(notes.logs.is_partitioned());
            assert_eq!(notes.registry.records(), 3);
            assert_eq!(notes.get("r1").unwrap().unwrap().get("kind"), Some(&json!("rule")));
        }

        #[test]
        fn keeps_full_text_results_identical_to_the_unpartitioned_table() {
            let dir = tempdir().unwrap();
            let flat_hits = {
                let flat = seeded_flat(dir.path());
                let spec = QueryBuilder::new().search("silnik", SearchMode::All).spec();
                let mut hits = flat.query(&spec).unwrap().keys;
                hits.sort();
                hits
            };

            let notes = open_notes(dir.path());
            let spec = QueryBuilder::new().search("silnik", SearchMode::All).spec();
            let mut partitioned_hits = notes.query(&spec).unwrap().keys;
            partitioned_hits.sort();

            assert!(!flat_hits.is_empty());
            assert_eq!(partitioned_hits, flat_hits);
        }

        #[test]
        fn writes_and_reads_further_records_after_the_move() {
            let dir = tempdir().unwrap();
            drop(seeded_flat(dir.path()));

            let mut notes = open_notes(dir.path());
            notes
                .put(
                    &json!({ "id": "n3", "kind": "note", "title": "written after the move" }),
                    false,
                )
                .unwrap();
            assert!(notes.delete("a1", false).unwrap());

            assert_eq!(
                notes.get("n3").unwrap().unwrap().get("title"),
                Some(&json!("written after the move"))
            );
            assert_eq!(notes.get("a1").unwrap(), None);
            assert_eq!(partition_lines(dir.path(), "note").len(), 2);
            assert_eq!(partition_lines(dir.path(), "adr").len(), 2);
        }

        #[test]
        fn an_unpartitioned_table_keeps_writing_the_single_flat_log() {
            let dir = tempdir().unwrap();
            let mut notes = Table::open(notes_schema(false), log_path(dir.path())).unwrap();
            for note in seed() {
                notes.put(&note, false).unwrap();
            }

            assert!(log_path(dir.path()).is_file());
            assert!(!dir.path().join("notes").exists());
            assert_eq!(
                fs::read_to_string(log_path(dir.path())).unwrap().trim().lines().count(),
                4
            );
            assert_eq!(notes.get("n1").unwrap().unwrap().get("slug"), Some(&json!("n1")));
        }
    }

    /// The bounded LRU of parsed documents `Table::get` may answer from — see `table::cache`. Every
    /// test here is about what the cache may never do: outlive a write, a foreign write caught by
    /// `refresh`, a compaction, or its own size limit.
    mod read_cache {
        use std::io::Write as _;

        use super::*;

        /// Small enough that a record with a few-hundred-character field counts as large, and far
        /// above what the seeded records measure.
        const LARGE_BYTES: usize = 512;

        fn policy(entries: usize, large_entries: usize) -> CachePolicy {
            CachePolicy::Enabled {
                entries,
                large_entries,
                large_bytes: LARGE_BYTES,
            }
        }

        fn open_cached(cache: CachePolicy) -> (tempfile::TempDir, Table, PathBuf) {
            let (dir, mut table) = open_table();
            let path = dir.path().join("users.jsonl");
            table.set_cache(cache);
            (dir, table, path)
        }

        fn seed(table: &mut Table, key: &str, name: &str) {
            table
                .put(&json!({ "id": key, "name": name, "age": 30 }), false)
                .unwrap();
        }

        fn name_of(table: &Table, key: &str) -> Value {
            table.get(key).unwrap().unwrap().get("name").unwrap().clone()
        }

        /// Overwrites every byte of the log with something no reader can parse, leaving the file
        /// length alone so the registry's offsets still point inside it: a `get` that answers after
        /// this never went back to the file, and one that reaches for it fails loudly.
        fn scribble(path: &Path) {
            let length = fs::metadata(path).unwrap().len() as usize;
            let mut file = fs::OpenOptions::new().write(true).open(path).unwrap();
            file.write_all(&vec![b'#'; length]).unwrap();
        }

        #[test]
        fn a_second_get_of_a_cached_key_answers_without_touching_the_log_again() {
            let (_dir, mut table, path) = open_cached(policy(4, 1));
            seed(&mut table, "u1", "Ania");
            assert_eq!(name_of(&table, "u1"), json!("Ania"));

            scribble(&path);

            assert_eq!(name_of(&table, "u1"), json!("Ania"));
        }

        #[test]
        fn without_a_cache_the_same_second_get_goes_back_to_the_log() {
            let (dir, mut table) = open_table();
            let path = dir.path().join("users.jsonl");
            seed(&mut table, "u1", "Ania");
            assert_eq!(name_of(&table, "u1"), json!("Ania"));

            scribble(&path);

            assert!(
                matches!(table.get("u1"), Err(TableError::Io(_))),
                "an uncached read parses the log line every time"
            );
        }

        #[test]
        fn a_table_nobody_configured_caches_nothing_and_reads_exactly_as_before() {
            let (_dir, mut table) = open_table();
            assert_eq!(table.cache_policy(), CachePolicy::Disabled);

            seed(&mut table, "u1", "Ania");
            assert_eq!(name_of(&table, "u1"), json!("Ania"));
            table.patch("u1", &json!({ "name": "Zosia" }), false).unwrap();
            assert_eq!(name_of(&table, "u1"), json!("Zosia"));
            assert!(table.delete("u1", false).unwrap());
            assert_eq!(table.get("u1").unwrap(), None);
            assert_eq!(table.cache.len(), (0, 0));
        }

        #[test]
        fn a_put_over_a_cached_key_is_what_the_next_get_answers() {
            let (_dir, mut table, _path) = open_cached(policy(4, 1));
            seed(&mut table, "u1", "Ania");
            assert_eq!(name_of(&table, "u1"), json!("Ania"));

            seed(&mut table, "u1", "Zosia");

            assert_eq!(name_of(&table, "u1"), json!("Zosia"));
        }

        #[test]
        fn a_patch_over_a_cached_key_is_what_the_next_get_answers() {
            let (_dir, mut table, _path) = open_cached(policy(4, 1));
            seed(&mut table, "u1", "Ania");
            assert_eq!(name_of(&table, "u1"), json!("Ania"));

            table.patch("u1", &json!({ "age": 41 }), false).unwrap();

            assert_eq!(table.get("u1").unwrap().unwrap().get("age"), Some(&json!(41)));
        }

        #[test]
        fn a_delete_of_a_cached_key_leaves_nothing_for_the_next_get() {
            let (_dir, mut table, _path) = open_cached(policy(4, 1));
            seed(&mut table, "u1", "Ania");
            assert_eq!(name_of(&table, "u1"), json!("Ania"));

            assert!(table.delete("u1", false).unwrap());

            assert_eq!(table.get("u1").unwrap(), None);
            assert_eq!(table.cache.len(), (0, 0));
        }

        #[test]
        fn the_limit_evicts_the_least_recently_read_key_and_it_still_reads_off_the_log() {
            let (_dir, mut table, _path) = open_cached(policy(2, 1));
            seed(&mut table, "u1", "Ania");
            seed(&mut table, "u2", "Piotr");
            seed(&mut table, "u3", "Zosia");

            assert_eq!(name_of(&table, "u2"), json!("Piotr"));
            assert_eq!(name_of(&table, "u1"), json!("Ania"));
            // NOTE: this third read is the one that has to evict u2 — the least recently read of the two.
            assert_eq!(name_of(&table, "u3"), json!("Zosia"));

            assert_eq!(table.cache.len(), (2, 0));
            assert_eq!(
                name_of(&table, "u2"),
                json!("Piotr"),
                "the evicted key falls back to the log"
            );
        }

        #[test]
        fn an_evicted_key_reaches_for_the_log_while_the_two_kept_ones_do_not() {
            let (_dir, mut table, path) = open_cached(policy(2, 1));
            seed(&mut table, "u1", "Ania");
            seed(&mut table, "u2", "Piotr");
            seed(&mut table, "u3", "Zosia");
            table.get("u2").unwrap();
            table.get("u1").unwrap();
            table.get("u3").unwrap();

            scribble(&path);

            assert!(matches!(table.get("u2"), Err(TableError::Io(_))));
            assert_eq!(name_of(&table, "u1"), json!("Ania"));
            assert_eq!(name_of(&table, "u3"), json!("Zosia"));
        }

        #[test]
        fn large_documents_share_their_own_smaller_tier_and_leave_the_small_ones_alone() {
            let (_dir, mut table, path) = open_cached(policy(2, 1));
            let long = "x".repeat(LARGE_BYTES);
            seed(&mut table, "u1", "Ania");
            seed(&mut table, "u2", "Piotr");
            seed(&mut table, "d1", &long);
            seed(&mut table, "d2", &long);

            table.get("u1").unwrap();
            table.get("u2").unwrap();
            table.get("d1").unwrap();
            table.get("d2").unwrap();

            assert_eq!(
                table.cache.len(),
                (2, 1),
                "two large documents share one slot; the small ones keep both of theirs"
            );

            scribble(&path);

            assert_eq!(name_of(&table, "u1"), json!("Ania"));
            assert_eq!(name_of(&table, "u2"), json!("Piotr"));
            assert_eq!(name_of(&table, "d2"), json!(long));
            assert!(
                matches!(table.get("d1"), Err(TableError::Io(_))),
                "the older large document was the one evicted"
            );
        }

        #[test]
        fn a_refresh_that_catches_a_foreign_write_drops_the_document_it_cached() {
            let (_dir, mut writer, mut follower) = two_handles();
            follower.set_cache(policy(4, 1));
            writer
                .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
                .unwrap();
            follower.refresh().unwrap();
            assert_eq!(name_of(&follower, "u1"), json!("Ania"));

            writer
                .put(&json!({ "id": "u1", "name": "Zosia", "age": 30 }), false)
                .unwrap();

            assert_eq!(follower.refresh().unwrap(), RefreshOutcome::CaughtUp { entries: 1 });
            assert_eq!(name_of(&follower, "u1"), json!("Zosia"));
        }

        #[test]
        fn a_following_read_of_a_cached_key_sees_the_foreign_write_it_caught_up_with() {
            let (_dir, mut writer, mut follower) = two_handles();
            follower.set_cache(policy(4, 1));
            follower.set_follow(Some(FollowPolicy {
                interval: Duration::ZERO,
            }));
            writer
                .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
                .unwrap();
            assert_eq!(
                follower.get_following("u1").unwrap().unwrap().get("name"),
                Some(&json!("Ania"))
            );

            writer
                .put(&json!({ "id": "u1", "name": "Zosia", "age": 30 }), false)
                .unwrap();

            assert_eq!(
                follower.get_following("u1").unwrap().unwrap().get("name"),
                Some(&json!("Zosia"))
            );
        }

        #[test]
        fn a_reload_after_a_foreign_compaction_throws_the_whole_cache_away() {
            let (_dir, mut writer, mut follower) = two_handles();
            follower.set_cache(policy(4, 1));
            writer
                .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
                .unwrap();
            writer
                .put(&json!({ "id": "u2", "name": "Piotr", "age": 41 }), false)
                .unwrap();
            follower.refresh().unwrap();
            assert_eq!(name_of(&follower, "u1"), json!("Ania"));
            assert_eq!(name_of(&follower, "u2"), json!("Piotr"));

            writer
                .put(&json!({ "id": "u1", "name": "Zosia", "age": 30 }), false)
                .unwrap();
            writer.compact().unwrap();

            assert_eq!(follower.refresh().unwrap(), RefreshOutcome::Reloaded);
            assert_eq!(name_of(&follower, "u1"), json!("Zosia"));
            assert_eq!(name_of(&follower, "u2"), json!("Piotr"));
        }

        #[test]
        fn compacting_this_handles_own_log_leaves_nothing_cached_behind() {
            let (_dir, mut table, _path) = open_cached(policy(4, 1));
            seed(&mut table, "u1", "Ania");
            seed(&mut table, "u2", "Piotr");
            seed(&mut table, "u1", "Zosia");
            table.get("u1").unwrap();
            table.get("u2").unwrap();

            table.compact().unwrap();

            assert_eq!(
                table.cache.len(),
                (0, 0),
                "every offset the cache was filled under moved"
            );
            assert_eq!(name_of(&table, "u1"), json!("Zosia"));
            assert_eq!(name_of(&table, "u2"), json!("Piotr"));
        }

        #[test]
        fn a_unit_of_work_entry_applied_over_a_cached_key_invalidates_it() {
            let (_dir, mut table, _path) = open_cached(policy(4, 1));
            seed(&mut table, "u1", "Ania");
            assert_eq!(name_of(&table, "u1"), json!("Ania"));

            let prepared = table
                .prepare_txn_ops(&[TxnOp::Put(json!({ "id": "u1", "name": "Zosia", "age": 30 }))])
                .unwrap();
            table.apply_txn_entries(prepared).unwrap();

            assert_eq!(name_of(&table, "u1"), json!("Zosia"));
        }

        #[test]
        fn a_recovered_journal_line_over_a_cached_key_invalidates_it() {
            let (_dir, mut writer, mut follower) = two_handles();
            follower.set_cache(policy(4, 1));
            writer
                .put(&json!({ "id": "u1", "name": "Ania", "age": 30 }), false)
                .unwrap();
            follower.refresh().unwrap();
            assert_eq!(name_of(&follower, "u1"), json!("Ania"));

            let prepared = writer
                .prepare_txn_ops(&[TxnOp::Put(json!({ "id": "u1", "name": "Zosia", "age": 30 }))])
                .unwrap();
            let lines: Vec<(u64, String)> = prepared
                .iter()
                .map(|entry| (entry.seq(), entry.line().to_string()))
                .collect();
            assert_eq!(follower.recover_txn_lines(&lines).unwrap(), 1);

            assert_eq!(name_of(&follower, "u1"), json!("Zosia"));
        }

        #[test]
        fn re_arming_the_cache_starts_it_cold() {
            let (_dir, mut table, path) = open_cached(policy(4, 1));
            seed(&mut table, "u1", "Ania");
            table.get("u1").unwrap();

            table.set_cache(policy(4, 1));
            scribble(&path);

            assert!(matches!(table.get("u1"), Err(TableError::Io(_))));
        }
    }
}