hayson-engine 0.8.0

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

use serde_json::Value;

use crate::atomic::AtomicTemporaryFileError;
use crate::catalog::{
    check_integrity, empty_catalog, expand_indexes, index_identity, parse_catalog, relations_touching, write_catalog,
    Catalog, CatalogCompaction, CatalogLayout, ExpandedIndex, IntegrityReport, Relation, TableEntry, TableOptions,
    TriggerEntry,
};
use crate::checkpoint::{read_snapshot, RestoreOutcome};
use crate::crypto::{CipherOptions, FieldCipher};
use crate::errors::{HaysonCipherError, HaysonSchemaError};
use crate::lock::{DirLock, DirLockError, LockOptions};
use crate::query::builder::QueryBuilder;
use crate::schema::{is_partitioned, parse_schema, FieldSpec, TableSchema};
use crate::storage::{partition_names_in, LOG_SUFFIX};
use crate::table::{describe_table_document, CachePolicy, FollowPolicy, Table, TableError};
use crate::txn::{apply_prepared, recover_incomplete, PreparedUnit, TxnError, UnitOfWork, UnitOfWorkResult};

use super::layout::{resolve_kind, stored_kind_of, Layout, LayoutOptions, DEFAULT_METADATA_FILE};
/// Matches the TS `DEFAULT_SNAPSHOT_EVERY` (`packages/hayson/src/database/database.ts`) — ops
/// applied to a table between snapshots before the next commit writes one unprompted.
const DEFAULT_SNAPSHOT_EVERY: u64 = 1000;
/// Matches the TS `DEFAULT_COMPACTION` (`packages/hayson/src/database/database.ts`) — the garbage
/// a table's log has to hold before a write rewrites it on its own.
const DEFAULT_COMPACTION: CatalogCompaction = CatalogCompaction::Enabled {
    dead_bytes_ratio: 0.4,
    min_dead_bytes: 64 * 1024,
};
const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;

#[derive(Debug)]
pub enum DatabaseError {
    Lock(DirLockError),
    Schema(HaysonSchemaError),
    /// The handle was opened readonly and `operation` needs the writer. Its own variant rather than
    /// a `Schema` refusal so a binding can tell it apart without reading the message.
    ReadOnly {
        dir: PathBuf,
        operation: String,
    },
    /// `drop_field` refused because live records still carry the field. `records` is how many were
    /// found, which is what a caller weighing `force` needs.
    FieldInUse {
        table: String,
        field: String,
        records: usize,
    },
    Catalog(String),
    Io(io::Error),
    /// An IO failure a caller can only act on knowing which file it was about: `io::Error` carries
    /// no path, and a readonly open that cannot deliver one log has to name it.
    Log(PathBuf, io::Error),
    Table(String, TableError),
    Txn(TxnError),
    /// Boxed for the same reason as `TableError::Cipher` — the `open()`-time keyring is the only
    /// thing that ever builds one, so nothing else should pay for its size at rest.
    Cipher(Box<HaysonCipherError>),
}

impl fmt::Display for DatabaseError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            DatabaseError::Lock(error) => error.fmt(f),
            DatabaseError::Schema(error) => error.fmt(f),
            DatabaseError::ReadOnly { dir, operation } => {
                write!(f, "database \"{}\" is readonly — {operation} is not available", dir.display())
            }
            DatabaseError::FieldInUse { table, field, records } => write!(
                f,
                "field \"{field}\" of table \"{table}\" still holds a value in {records} live record(s) — pass force to drop it anyway"
            ),
            DatabaseError::Catalog(message) => write!(f, "{message}"),
            DatabaseError::Io(error) => error.fmt(f),
            DatabaseError::Log(path, error) => write!(f, "{error}, open \"{}\"", path.display()),
            DatabaseError::Table(name, error) => write!(f, "table \"{name}\": {error}"),
            DatabaseError::Txn(error) => error.fmt(f),
            DatabaseError::Cipher(error) => error.fmt(f),
        }
    }
}

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

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

impl From<DirLockError> for DatabaseError {
    fn from(value: DirLockError) -> Self {
        DatabaseError::Lock(value)
    }
}

impl From<HaysonSchemaError> for DatabaseError {
    fn from(value: HaysonSchemaError) -> Self {
        DatabaseError::Schema(value)
    }
}

impl From<TxnError> for DatabaseError {
    fn from(value: TxnError) -> Self {
        DatabaseError::Txn(value)
    }
}

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

/// Whether the tables of this database catch up with foreign writes on their own — the port of the
/// TS `OpenOptions.follow`, which takes `false`, `true` or `{ intervalMs }`.
///
/// [`FollowOption::Default`] resolves by `readonly`, exactly as the TS `resolveFollow` does: a
/// readonly handle follows (at [`DEFAULT_FOLLOW_INTERVAL`], which is zero — check on every read), a
/// writable one does not, since it is the only writer the directory allows and already sees its own
/// writes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum FollowOption {
    #[default]
    Default,
    /// Never catch up unprompted; only an explicit [`Table::refresh`]/[`Database::refresh`] does.
    Off,
    /// Catch up on a read, at most once per this interval. `Duration::ZERO` means every read.
    Every(Duration),
}

/// The interval a `follow` left at its default resolves to — the TS `DEFAULT_FOLLOW_INTERVAL_MS`,
/// which is `0`: check on every read, which is one `stat` per log file when nothing changed.
pub const DEFAULT_FOLLOW_INTERVAL: Duration = Duration::ZERO;

impl FollowOption {
    fn resolve(self, readonly: bool) -> Option<FollowPolicy> {
        match self {
            FollowOption::Off => None,
            FollowOption::Every(interval) => Some(FollowPolicy { interval }),
            FollowOption::Default if readonly => Some(FollowPolicy {
                interval: DEFAULT_FOLLOW_INTERVAL,
            }),
            FollowOption::Default => None,
        }
    }
}

/// Whether `Database::open` takes the writer lock at all, and on what terms when it does.
///
/// [`LockMode::Skip`] is the port of the TS `open(dir, { lock: false })` and is dangerous by
/// design: a handle that skipped the lock writes the catalog it holds in memory over whatever
/// another writer declared meanwhile, because the lock is the only guard there. See
/// `docs/format.md` and the TS `deklarowanie-indeksu-wspolbieznie.test.ts` for the exact window.
#[derive(Debug, Clone, Copy)]
pub enum LockMode {
    /// One writer per directory (ADR-hjxp/0006), on this lease tuning.
    Enabled(LockOptions),
    /// No lock at all — mutual exclusion between writers of this directory is not enforced.
    Skip,
}

impl Default for LockMode {
    fn default() -> Self {
        LockMode::Enabled(LockOptions::default())
    }
}

impl LockMode {
    /// The lease tuning of an enabled lock — `None` when the caller asked to skip locking, which
    /// leaves nothing to tune.
    pub fn options(&self) -> Option<&LockOptions> {
        match self {
            LockMode::Enabled(options) => Some(options),
            LockMode::Skip => None,
        }
    }

    /// The mutable half of [`LockMode::options`], for a host adjusting the defaults it opens with.
    pub fn options_mut(&mut self) -> Option<&mut LockOptions> {
        match self {
            LockMode::Enabled(options) => Some(options),
            LockMode::Skip => None,
        }
    }
}

/// How `Database::open` acquires the writer lock and whether it opens for writing at all.
#[derive(Debug, Clone)]
pub struct OpenOptions {
    pub readonly: bool,
    pub lock: LockMode,
    /// Ops applied to a table between periodic `<table>.idx` snapshots — see [`Table::snapshot`].
    /// `0` disables periodic snapshotting; defaults to [`DEFAULT_SNAPSHOT_EVERY`].
    pub snapshot_every: u64,
    /// Overrides the automatic-compaction policy of every table for this handle — the TS
    /// `compaction` option, where `Disabled` is its `false`. `None` leaves the choice to the
    /// catalog: the table's own `options.compaction`, then `defaults.compaction`, then
    /// [`DEFAULT_COMPACTION`].
    pub compaction: Option<CatalogCompaction>,
    /// Keys for the fields a schema marks `cypher`. Without them such fields cannot be written.
    pub cypher: Option<CipherOptions>,
    /// Whether reads catch up with what other processes appended, without an explicit refresh.
    pub follow: FollowOption,
    /// How many parsed documents each table of this handle keeps in memory in front of its log —
    /// see [`CachePolicy`]. `Disabled` by default, and named here rather than in the catalog on
    /// purpose: it costs memory in this process only, so it is the opener's call and not a property
    /// of the database on disk.
    pub cache: CachePolicy,
    /// Where in the directory the metadata document and the table files live. Left at its default,
    /// this is the flat arrangement every database used before layouts existed, under
    /// `hayson.json` — and an existing database is opened in the arrangement it is stored in,
    /// whether or not the caller names one.
    pub layout: LayoutOptions,
}

impl Default for OpenOptions {
    fn default() -> Self {
        OpenOptions {
            readonly: false,
            lock: LockMode::default(),
            snapshot_every: DEFAULT_SNAPSHOT_EVERY,
            compaction: None,
            cypher: None,
            follow: FollowOption::default(),
            cache: CachePolicy::default(),
            layout: LayoutOptions::default(),
        }
    }
}

fn now_ms() -> u64 {
    chrono::Utc::now().timestamp_millis().max(0) as u64
}

fn fnv1a(bytes: &[u8]) -> String {
    let mut hash = FNV_OFFSET;
    for byte in bytes {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(FNV_PRIME);
    }
    format!("{hash:016x}")
}

/// Identifies a `TableSchema` and the index set it is opened with for
/// `checkpoint::restore_or_rebuild`'s pairing check — cheap and deterministic, not cryptographic:
/// it only has to change whenever either does. The index set belongs in it because a snapshot
/// entry carries the projection the registry held when it was written: a snapshot taken before an
/// index was declared projects none of that index's fields, so restoring from it would rebuild the
/// new index empty and answer queries against it with silence. An empty `declared` appends
/// nothing, mirroring the TS `schemaFingerprint`.
fn schema_fingerprint(schema: &TableSchema, declared: &[ExpandedIndex]) -> String {
    if declared.is_empty() {
        return fnv1a(format!("{schema:?}").as_bytes());
    }
    let mut described: Vec<String> = declared.iter().map(|index| format!("{index:?}")).collect();
    described.sort();
    fnv1a(format!("{schema:?}|{}", described.join("|")).as_bytes())
}

/// The inverse of `catalog::table_entry_of`: rebuilds the `TableSchema` a `TableEntry` describes,
/// so a table can be opened straight from the catalog without a `<table>.schema.json` file. No
/// such conversion exists in `catalog/` yet (format 1's `describeTableEntry` was never ported), so
/// it lives here — the only place format 2 without format 1 needs it.
fn table_schema_of(name: &str, entry: &TableEntry) -> TableSchema {
    let fields: BTreeMap<String, FieldSpec> = entry
        .fields
        .iter()
        .map(|(field, spec)| (field.clone(), spec.spec.clone()))
        .collect();
    TableSchema {
        table: name.to_string(),
        revision: entry.revision as u32,
        primary_key: entry.primary_key.clone(),
        partition_by: entry.partition_by.clone(),
        fields,
    }
}

/// The trigger declarations of `catalog` naming `name` as their source — what `Table::set_triggers`
/// takes.
fn triggers_of(catalog: &Catalog, name: &str) -> Vec<TriggerEntry> {
    let Some(declared) = catalog.triggers.as_ref() else {
        return Vec::new();
    };
    declared
        .iter()
        .filter(|trigger| trigger.table == name)
        .cloned()
        .collect()
}

/// Hands every open table the trigger declarations naming it as their source (`Table::
/// set_triggers`'s contract) — called after `open()` rebuilds `tables` from the catalog, and again
/// after `declare_trigger` changes `catalog.triggers`, so a live table is never out of step with
/// what is on disk. A table this handle reopens on its own gets its set from
/// [`open_catalog_table`] instead, which covers only the one table it brings up.
fn sync_table_triggers(catalog: &Catalog, tables: &mut HashMap<String, Table>) {
    for (name, table) in tables.iter_mut() {
        table.set_triggers(triggers_of(catalog, name));
    }
}

/// One table option as the catalog resolves it: what the table entry declares beats what
/// `catalog.defaults` declares for the whole database, the same order [`compaction_of`] spells out
/// for the two sources it shares with this.
fn option_of<T>(catalog: &Catalog, entry: &TableEntry, pick: impl Fn(&TableOptions) -> Option<T>) -> Option<T> {
    entry
        .options
        .as_ref()
        .and_then(&pick)
        .or_else(|| catalog.defaults.as_ref().and_then(&pick))
}

/// Applies the `imageDir` [`option_of`] resolves, against the database directory, to a just-opened
/// table — left at its own computed default (a `<table>.images/` sibling of the log file) when
/// neither the entry nor `catalog.defaults` names one.
fn apply_image_dir(dir: &Path, catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
    if let Some(image_dir) = option_of(catalog, entry, |options| options.image_dir.clone()) {
        table.set_image_dir(dir.join(image_dir));
    }
}

/// Applies the `sync` [`option_of`] resolves to a just-opened table — left at `Table`'s own
/// `Commit` default when neither the entry nor `catalog.defaults` names one, which is the
/// behaviour every table had while the field was parsed but never read (`docs/format.md`).
fn apply_sync(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
    if let Some(sync) = option_of(catalog, entry, |options| options.sync) {
        table.set_sync(sync);
    }
}

/// Applies the table entry's `validation` to a just-opened table — left at `Table`'s own `Strict`
/// default when the entry declares none, which is the behaviour every table had while the field was
/// parsed but never read (`docs/format.md`).
///
/// [`option_of`] does not apply here: `validation` is a key of the table section itself, and
/// neither `tables.*.options` nor `catalog.defaults` carries it, so there is no database-wide
/// declaration to fall back to.
fn apply_validation(entry: &TableEntry, table: &mut Table) {
    if let Some(validation) = entry.validation {
        table.set_validation(validation);
    }
}

/// Hands a just-opened table the fields its entry marks `deprecated`, each against its note (never
/// empty — `parse_catalog` refuses `deprecated: ""`) — left empty when the entry marks none.
///
/// The mark lives on the catalog field spec, not on `schema::FieldSpec`, so the schema
/// `table_schema_of` rebuilds cannot carry it and the table has to be told separately — the same
/// way [`apply_validation`] tells it about a key of the table section.
fn apply_deprecated(entry: &TableEntry, table: &mut Table) {
    let marked: BTreeMap<String, String> = entry
        .fields
        .iter()
        .filter_map(|(name, field)| field.deprecated.as_ref().map(|note| (name.clone(), note.clone())))
        .collect();
    if !marked.is_empty() {
        table.set_deprecated_fields(marked);
    }
}

/// Applies the `linkParsing` [`option_of`] resolves to a just-opened table — left at `Table`'s own
/// `Eager` default when neither the entry nor `catalog.defaults` names one, per the "default eager,
/// configurable to lazy" call.
fn apply_link_parsing(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
    if let Some(link_parsing) = option_of(catalog, entry, |options| options.link_parsing) {
        table.set_link_parsing(link_parsing);
    }
}

/// Arms a just-(re)opened table's automatic compaction, resolving the four places a policy can
/// come from in the order the catalog documents: what `open()` named beats what the table entry
/// declares, which beats `catalog.defaults`, which beats [`DEFAULT_COMPACTION`].
fn compaction_of(named: Option<CatalogCompaction>, catalog: &Catalog, entry: Option<&TableEntry>) -> CatalogCompaction {
    named
        .or_else(|| {
            entry
                .and_then(|entry| entry.options.as_ref())
                .and_then(|options| options.compaction)
        })
        .or_else(|| catalog.defaults.as_ref().and_then(|defaults| defaults.compaction))
        .unwrap_or(DEFAULT_COMPACTION)
}

/// Arms a just-(re)opened table with the policy [`compaction_of`] resolved — the thresholds
/// `Table::compact_if_due` measures each partition against.
fn apply_compaction(compaction: CatalogCompaction, table: &mut Table) {
    table.set_compaction(compaction);
}

/// Hands a just-(re)opened table the keyring `options.cypher` resolved into a `FieldCipher` — left
/// at `None` when `open()` named no keys, 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.
fn apply_cipher(cipher: Option<&FieldCipher>, table: &mut Table) {
    table.set_cipher(cipher.cloned());
}

/// Counts live records of `open_table` still carrying `field` and refuses when any is left — the
/// half of `Database::drop_field` that `force` waives, kept apart from the catalog half so the two
/// can be sequenced independently.
fn assert_no_residual_values(name: &str, open_table: &Table, field: &str) -> Result<(), DatabaseError> {
    let all = open_table
        .query(&QueryBuilder::new().spec())
        .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
    let mut residual = 0usize;
    for key in &all.keys {
        let record = open_table
            .get(key)
            .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
        if record.is_some_and(|record| record.contains_key(field)) {
            residual += 1;
        }
    }
    if residual > 0 {
        return Err(DatabaseError::FieldInUse {
            table: name.to_string(),
            field: field.to_string(),
            records: residual,
        });
    }
    Ok(())
}

/// Sorts a failed catalog write into the variant that keeps the most of it. `write_catalog` erases
/// its refusals into `Box<dyn Error>`, and one of them — the `.tmp` path that could not be claimed,
/// raised as an [`AtomicTemporaryFileError`] inside an `io::Error` — names a file the caller has to
/// remove before retrying. Flattening it into `Catalog(String)` would leave that path recoverable
/// only from the message text; everything else has nothing typed to keep and stays a catalog error.
fn catalog_write_error(error: Box<dyn std::error::Error>) -> DatabaseError {
    match error.downcast::<io::Error>() {
        Ok(io) if AtomicTemporaryFileError::downcast(&io).is_some() => DatabaseError::Io(*io),
        Ok(io) => DatabaseError::Catalog(io.to_string()),
        Err(other) => DatabaseError::Catalog(other.to_string()),
    }
}

/// Directory a partitioned table keeps its `<value>.jsonl` files in — the same one `Table::open`
/// derives from the log path it is handed, spelled out here because this check runs before it.
fn partition_dir_of(layout: &Layout, table: &str) -> PathBuf {
    layout.tables().join(table)
}

/// Whether anything at all sits at `path`, without following a link: a dangling symlink is present
/// here, so that it is the open — not this check — that reports what is wrong with it.
fn on_disk(path: &Path) -> bool {
    fs::symlink_metadata(path).is_ok()
}

/// The log files the catalog says this table has and the disk does not — port of the TS
/// `missingLogsOf` (`packages/hayson/src/database/database.ts`).
///
/// An unpartitioned table has exactly one log, so its absence is the whole answer. A partitioned
/// one has as many as its index snapshot claims: a table nothing was ever written to legitimately
/// holds no file at all, and only a partition the snapshot recorded but the directory no longer
/// lists counts as gone.
fn missing_logs_of(layout: &Layout, index_path: &Path, schema: &TableSchema) -> io::Result<Vec<PathBuf>> {
    let flat = layout.log_path(&schema.table);
    if !is_partitioned(schema) {
        return Ok(if on_disk(&flat) { Vec::new() } else { vec![flat] });
    }
    let dir = partition_dir_of(layout, &schema.table);
    let present = partition_names_in(&dir);
    // NOTE: a flat log under a partitioned schema is the move `open_logs` finishes, not a gap.
    if present.is_empty() && on_disk(&flat) {
        return Ok(Vec::new());
    }
    let claimed = read_snapshot(index_path)?
        .and_then(|snapshot| snapshot.header.parts)
        .unwrap_or_default();
    Ok(claimed
        .into_iter()
        .filter(|part| !present.contains(&part.part))
        .map(|part| dir.join(format!("{}{LOG_SUFFIX}", part.part)))
        .collect())
}

/// The log files a readonly open of this table is about to take on, as the disk lists them.
fn present_logs_of(layout: &Layout, schema: &TableSchema) -> Vec<PathBuf> {
    let flat = layout.log_path(&schema.table);
    if !is_partitioned(schema) {
        return if on_disk(&flat) { vec![flat] } else { Vec::new() };
    }
    let dir = partition_dir_of(layout, &schema.table);
    let present = partition_names_in(&dir);
    if present.is_empty() {
        return if on_disk(&flat) { vec![flat] } else { Vec::new() };
    }
    present
        .iter()
        .map(|part| dir.join(format!("{part}{LOG_SUFFIX}")))
        .collect()
}

/// How one open table differs from the section the catalog on disk now holds for it — the port of
/// the TS `TableDrift` (`packages/hayson/src/database/database.ts`, before the binding cutover).
enum TableDrift {
    Absent,
    Schema { revision: u64 },
    Indexes { unserved: Vec<String>, stale: Vec<String> },
}

struct DriftedTable {
    name: String,
    drift: TableDrift,
}

/// Names in `expected` that `present` holds no counterpart for, matched by [`index_identity`]
/// rather than by name — a renamed index is the same index, and must not read as drift.
fn index_names_missing_from(expected: &[ExpandedIndex], present: &[ExpandedIndex]) -> Vec<String> {
    let identities: HashSet<String> = present.iter().map(index_identity).collect();
    let mut names: Vec<String> = expected
        .iter()
        .filter(|index| !identities.contains(&index_identity(index)))
        .map(|index| index.name.clone())
        .collect();
    names.sort();
    names
}

/// The schema without the three field modifiers the index half of the drift check already covers
/// in full: `unique` implies an exact index (`schema::parse`), and `index`/`fulltext` are what
/// `sugar_indexes_of` expands, so nothing is lost by dropping them here.
///
/// They have to go, because a schema rebuilt from the catalog (`table_schema_of`) carries only the
/// field modifiers the entry itself spells out, while one parsed from a schema document carries
/// whatever `schema::parse` derived — comparing them would report a handle as drifted from its own
/// write.
fn shape_of(schema: &TableSchema) -> TableSchema {
    let mut bare = schema.clone();
    for spec in bare.fields.values_mut() {
        spec.index = None;
        spec.unique = false;
        spec.fulltext = false;
    }
    bare
}

fn quoted(names: &[String]) -> String {
    names
        .iter()
        .map(|name| format!("\"{name}\""))
        .collect::<Vec<_>>()
        .join(", ")
}

fn stored_side_of(drift: &TableDrift) -> String {
    match drift {
        TableDrift::Absent => "declares no table at all".to_string(),
        TableDrift::Schema { revision } => format!("declares revision {revision}"),
        TableDrift::Indexes { unserved, stale } => {
            let mut halves = Vec::new();
            if !unserved.is_empty() {
                halves.push(format!(
                    "declares indexes this handle does not serve: {}",
                    quoted(unserved)
                ));
            }
            if !stale.is_empty() {
                halves.push(format!(
                    "no longer declares indexes this handle still serves: {}",
                    quoted(stale)
                ));
            }
            halves.join(", and ")
        }
    }
}

fn missing_logs_error(dir: &Path, name: &str, missing: &[PathBuf]) -> HaysonSchemaError {
    let listed = missing
        .iter()
        .map(|path| format!("\"{}\"", path.display()))
        .collect::<Vec<_>>()
        .join(", ");
    let one = missing.len() == 1;
    let (its, is, it, file, log) = if one {
        ("its log".to_string(), "is", "it", "file", "log")
    } else {
        (format!("{} of its logs", missing.len()), "are", "them", "files", "logs")
    };
    HaysonSchemaError(format!(
        "the catalog of \"{}\" declares table \"{name}\", but {its} {listed} {is} not on disk — this handle is readonly, so nothing here will write {it}; restore the {file}, or open the database for writing to have the {log} created",
        dir.display()
    ))
}

/// The guard a readonly open owes every table the catalog declares: a log the catalog names but the
/// disk does not hold is a refusal, not a table that opens empty, because nothing on this handle
/// will ever write it (F8, TS `#openIfOnDisk`).
///
/// A writable open needs none of this — it creates the logs it misses on the way in.
///
/// The two halves answer two different questions. The first is what the catalog promised and the
/// listing denies; the second is whether what the listing does show can actually be opened, since
/// `LogSet` opens its files with `create(true)` and would otherwise silently make a fresh log out
/// of a path that is broken — a dangling link being the case that shows it.
fn refuse_unreadable_logs(layout: &Layout, name: &str, entry: &TableEntry) -> Result<(), DatabaseError> {
    let schema = table_schema_of(name, entry);
    let missing = missing_logs_of(layout, &layout.index_path(name), &schema)?;
    if !missing.is_empty() {
        return Err(DatabaseError::Schema(missing_logs_error(layout.root(), name, &missing)));
    }
    for path in present_logs_of(layout, &schema) {
        fs::File::open(&path).map_err(|error| DatabaseError::Log(path, error))?;
    }
    Ok(())
}

/// Brings up one table exactly as the catalog describes it: the schema rebuilt from its entry, its
/// snapshot paired against its log, the catalog's full expanded index list, and the options
/// (`imageDir`, `linkParsing`, `sync`, `validation`, `compaction`, the keyring, the follow policy,
/// the triggers naming it as their source, the fields its entry marks `deprecated`) applied on
/// top. Every caller that swaps a live handle out — `open`, [`Database::refresh`],
/// [`Database::create_table`], [`Database::drop_field`], [`Database::declare_index`] — goes through
/// here, so the set of things a reopened table carries has one place to grow.
#[allow(clippy::too_many_arguments)]
fn open_catalog_table(
    layout: &Layout,
    name: &str,
    catalog: &Catalog,
    entry: &TableEntry,
    snapshot_every: u64,
    cipher: Option<&FieldCipher>,
    follow: Option<FollowPolicy>,
    readonly: bool,
    compaction: CatalogCompaction,
    cache: CachePolicy,
) -> Result<(Table, RestoreOutcome), DatabaseError> {
    let schema = table_schema_of(name, entry);
    let declared = expand_indexes(name, entry);
    let hash = schema_fingerprint(&schema, &declared);
    let (mut table, outcome) = Table::open_with_snapshot_and_declared_indexes(
        schema,
        layout.log_path(name),
        layout.index_path(name),
        &hash,
        snapshot_every,
        readonly,
        &declared,
    )
    .map_err(|error| DatabaseError::Table(name.to_string(), TableError::Io(error)))?;
    apply_image_dir(layout.root(), catalog, entry, &mut table);
    apply_link_parsing(catalog, entry, &mut table);
    apply_sync(catalog, entry, &mut table);
    apply_validation(entry, &mut table);
    apply_deprecated(entry, &mut table);
    apply_compaction(compaction, &mut table);
    table.set_cache(cache);
    apply_cipher(cipher, &mut table);
    table.set_follow(follow);
    table.set_triggers(triggers_of(catalog, name));
    Ok((table, outcome))
}

/// The first table log in `layout`, looked for where the layout puts them: directly in the tables
/// directory, and one level down inside it, which is where a partitioned table keeps its per-value
/// logs. Sorted, so the same directory always names the same log in a refusal.
fn first_table_log(layout: &Layout) -> Option<PathBuf> {
    let has_log_suffix = |path: &Path| {
        path.extension()
            .is_some_and(|ext| format!(".{}", ext.to_string_lossy()) == LOG_SUFFIX)
    };
    let entries_of = |dir: &Path| -> Vec<PathBuf> {
        let Ok(read) = fs::read_dir(dir) else { return Vec::new() };
        let mut found: Vec<PathBuf> = read.filter_map(Result::ok).map(|entry| entry.path()).collect();
        found.sort();
        found
    };

    let top = entries_of(layout.tables());
    if let Some(log) = top.iter().find(|path| path.is_file() && has_log_suffix(path)) {
        return Some(log.clone());
    }
    top.iter().filter(|path| path.is_dir()).find_map(|dir| {
        entries_of(dir)
            .into_iter()
            .find(|path| path.is_file() && has_log_suffix(path))
    })
}

/// Refuses to treat a directory that already holds table logs as a brand new database.
///
/// Without `hayson.json` there is nothing that says which schema those logs were written under, so
/// the only safe answer is to stop: opening would otherwise write a fresh empty catalog next to real
/// data, and the next write would append records of a foreign schema onto a log nobody can read back.
/// A `.gitignore` that swallowed the catalog, or a restore that copied only the data, both look
/// exactly like this. See `packages/hayson/test/katalog-bez-metadanych.test.ts`.
fn refuse_orphan_logs(layout: &Layout) -> Result<(), HaysonSchemaError> {
    let Some(log) = first_table_log(layout) else {
        return Ok(());
    };
    Err(HaysonSchemaError(format!(
        "\"{}\" holds table data but no \"{}\" — refusing to open it as an empty database, which would write a fresh catalog next to \"{}\" and append records of a foreign schema onto it; restore \"{}\" from a backup, or move the logs aside to start over",
        layout.root().display(),
        layout.metadata_file(),
        log.display(),
        layout.metadata_file(),
    )))
}

fn read_catalog(path: &Path) -> Result<Option<Catalog>, DatabaseError> {
    let raw = match fs::read_to_string(path) {
        Ok(raw) => raw,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error.into()),
    };
    let value: Value = serde_json::from_str(&raw)
        .map_err(|error| DatabaseError::Catalog(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
    Ok(Some(parse_catalog(&value)?))
}

/// Owns the writer lock for the length of one operation and no longer: every refusal after
/// `DirLock::acquire` unwinds through this `Drop` and takes the lock file with it, while a
/// successful open moves the lock out with [`HeldLock::into_held`] and keeps it for the handle.
///
/// `DirLock` itself does not clean up on drop (its `release` is the only place that removes files),
/// so without this a refused open would leave a lock file naming a process that is still alive —
/// which no staleness rule ever reclaims, and which locks the directory out for good.
///
/// Shared with `migrate_layout`, which holds the same lock for the length of a move and wants the
/// same guarantee about a refusal halfway through it.
pub(super) struct HeldLock(pub(super) Option<DirLock>);

impl HeldLock {
    fn into_held(mut self) -> Option<DirLock> {
        self.0.take()
    }
}

impl Drop for HeldLock {
    fn drop(&mut self) {
        if let Some(lock) = self.0.as_mut() {
            lock.release();
        }
    }
}

/// A directory of tables: the JSONL logs, their index snapshots, the metadata catalog
/// (`hayson.json`) describing them, and one writer lock (`lock::DirLock`, ADR-hjxp/0006 — one
/// writer per database directory).
pub struct Database {
    layout: Layout,
    lock: Option<DirLock>,
    catalog: Catalog,
    tables: HashMap<String, Table>,
    restore_outcomes: HashMap<String, RestoreOutcome>,
    readonly: bool,
    snapshot_every: u64,
    /// What `open()` named, not what any table resolved to — kept so a table opened later by
    /// `refresh`/`createTable` is armed on the same terms as the ones opened at startup.
    compaction: Option<CatalogCompaction>,
    cipher: Option<FieldCipher>,
    follow: Option<FollowPolicy>,
    /// What `open()` named, for the same reason as `compaction`: a table opened later by
    /// `refresh`/`createTable` caches on the same terms as the ones opened at startup.
    cache: CachePolicy,
}

impl Database {
    pub fn open(dir: &Path) -> Result<Database, DatabaseError> {
        Database::open_with(dir, OpenOptions::default())
    }

    /// Opens (and creates when missing) a Hayson database directory:
    ///
    /// 1. locks the directory for writing, unless `options.readonly` or `options.lock` is
    ///    [`LockMode::Skip`] (ADR-hjxp/0006);
    /// 2. reads `hayson.json`, or starts an empty catalog when the directory is new;
    /// 3. for each declared table, opens its log and restores its registry from `<table>.idx`
    ///    via `checkpoint::restore_or_rebuild` rather than always scanning the whole log;
    /// 4. finishes an unfinished `.hayson.txn` journal, if one is on disk, before returning —
    ///    so no write reaches a table until whatever a prior crash left half-committed is done.
    ///
    /// A refusal at any step after the lock was taken releases it again before returning, so a
    /// directory this call rejected is left no more locked than it was found.
    pub fn open_with(dir: &Path, options: OpenOptions) -> Result<Database, DatabaseError> {
        let layout = Database::resolve_layout(dir, &options.layout)?;

        if !options.readonly {
            for directory in layout.directories() {
                fs::create_dir_all(directory)?;
            }
        }

        let lock = HeldLock(match options.lock {
            LockMode::Enabled(lock) if !options.readonly => Some(DirLock::acquire(layout.lock_path(), lock)?),
            LockMode::Enabled(_) | LockMode::Skip => None,
        });

        Database::open_locked(layout, options, lock)
    }

    /// Works out where this database's files are before anything is opened or locked: the metadata
    /// document name the options ask for (refusing `metadataFile` and `manifest` naming two
    /// different files), then the arrangement — whatever the document on disk records, or the
    /// requested one when there is no document yet.
    ///
    /// Runs before `create_dir_all`, so a refusal here leaves the directory exactly as it found it.
    /// Public for that same reason: a caller that only needs to know where the writer lock or the
    /// metadata document of a directory sits gets the paths without opening the database, taking
    /// the lock, or creating anything.
    ///
    /// @param dir the database directory
    /// @param options the `layout`/`metadataFile`/`manifest` triple of `OpenOptions`, unresolved
    /// @returns where this database's files are, or a refusal to name the metadata document twice
    pub fn resolve_layout(dir: &Path, options: &LayoutOptions) -> Result<Layout, DatabaseError> {
        let metadata_file = options.metadata_file()?.unwrap_or(DEFAULT_METADATA_FILE);
        let stored = stored_kind_of(&dir.join(metadata_file));
        let kind = resolve_kind(dir, stored, options.kind)?;
        Ok(Layout::new(dir, kind, metadata_file))
    }

    fn open_locked(layout: Layout, options: OpenOptions, lock: HeldLock) -> Result<Database, DatabaseError> {
        let found = read_catalog(layout.metadata_path())?;
        let catalog = match found {
            Some(catalog) => catalog,
            None => {
                refuse_orphan_logs(&layout)?;
                let catalog = empty_catalog(CatalogLayout::from(layout.kind()), now_ms());
                if !options.readonly {
                    write_catalog(layout.metadata_path(), &catalog)
                        .map_err(|error| DatabaseError::Catalog(error.to_string()))?;
                }
                catalog
            }
        };

        let cipher = options.cypher.clone().map(FieldCipher::new).transpose()?;
        let follow = options.follow.resolve(options.readonly);

        let mut tables = HashMap::new();
        let mut restore_outcomes = HashMap::new();
        for (name, entry) in &catalog.tables {
            if options.readonly {
                refuse_unreadable_logs(&layout, name, entry)?;
            }
            let (table, outcome) = open_catalog_table(
                &layout,
                name,
                &catalog,
                entry,
                options.snapshot_every,
                cipher.as_ref(),
                follow,
                options.readonly,
                compaction_of(options.compaction, &catalog, Some(entry)),
                options.cache,
            )?;
            tables.insert(name.clone(), table);
            restore_outcomes.insert(name.clone(), outcome);
        }

        if !options.readonly {
            recover_incomplete(&layout.journal_path(), &mut tables)?;
        }

        sync_table_triggers(&catalog, &mut tables);

        Ok(Database {
            layout,
            lock: lock.into_held(),
            catalog,
            tables,
            restore_outcomes,
            readonly: options.readonly,
            snapshot_every: options.snapshot_every,
            compaction: options.compaction,
            cipher,
            follow,
            cache: options.cache,
        })
    }

    pub fn dir(&self) -> &Path {
        self.layout.root()
    }

    /// Where this handle's files are on disk — the arrangement and the metadata document name it
    /// resolved at open, the read side of the TS `Database.layout` property.
    pub fn layout(&self) -> &Layout {
        &self.layout
    }

    /// The follow policy every table of this database was opened with — `None` when reads never
    /// catch up unprompted, which is what a writable open resolves to.
    pub fn follow(&self) -> Option<FollowPolicy> {
        self.follow
    }

    pub fn table_names(&self) -> Vec<&str> {
        let mut names: Vec<&str> = self.tables.keys().map(String::as_str).collect();
        names.sort_unstable();
        names
    }

    pub fn table(&self, name: &str) -> Option<&Table> {
        self.tables.get(name)
    }

    /// Handing out a writable table is the write path this type cannot see the end of, so the
    /// lease is renewed here as well — a host that never calls [`DirLock::tick`] still keeps its
    /// lock as long as it keeps writing. A lock already lost cannot be reported through an
    /// `Option`; the next call that returns a `Result` — `commit` and every schema change — raises
    /// [`DatabaseError::Lock`] instead.
    pub fn table_mut(&mut self, name: &str) -> Option<&mut Table> {
        let _ = self.renew_lock();
        self.tables.get_mut(name)
    }

    /// Renews the writer lease from outside — [`DirLock::tick`] reached through the lock this
    /// handle owns privately. A host without background threads (`wasm32-wasip1`) has no heartbeat
    /// of its own, so this is what a timer on its side drives; call it at least twice per
    /// `LockOptions::heartbeat_ms`. Does nothing on a readonly handle, which holds no lock.
    pub fn tick(&self) -> Result<(), DatabaseError> {
        self.renew_lock()
    }

    fn renew_lock(&self) -> Result<(), DatabaseError> {
        match &self.lock {
            Some(lock) => Ok(lock.tick()?),
            None => Ok(()),
        }
    }

    /// How the table named `name` was brought up at open — `Restored` from `<table>.idx` or
    /// `Rebuilt` by a full scan of its log. `None` when the database holds no such table.
    pub fn restore_outcome(&self, name: &str) -> Option<RestoreOutcome> {
        self.restore_outcomes.get(name).copied()
    }

    /// The metadata catalog as this handle last read it, without the drift gate — every table it
    /// describes may be one this handle does not serve. `catalog` is what a caller outside the
    /// engine should ask for.
    pub fn catalog_unchecked(&self) -> &Catalog {
        &self.catalog
    }

    /// The metadata catalog as this handle last read it — the parsed form of `hayson.json`, which
    /// `catalog::catalog_document` turns back into the document a caller can serialize.
    ///
    /// Refuses when any table open here has drifted from what the catalog now declares, because
    /// handing the document over would describe tables this handle does not serve. Refusing is all
    /// that happens: the handle is never reloaded behind the caller's back, so the schema and the
    /// index set every open table was opened with stay exactly as they were.
    pub fn catalog(&self) -> Result<&Catalog, DatabaseError> {
        let drifted = self.drifted_tables();
        if drifted.is_empty() {
            return Ok(&self.catalog);
        }
        Err(DatabaseError::Schema(self.drift_error(&drifted)))
    }

    /// The schema of one table as a standalone document, the shape `create_table` accepts — what
    /// the TS `describeTable` answers with.
    ///
    /// Read off the live handle (`table::describe_table_document`), not off the catalog entry, so
    /// it describes the schema and the index set the table is actually being served with — every
    /// table the catalog declares is open here, both after `open` and after
    /// [`Database::refresh`], so there is no catalog-only table to fall back to.
    ///
    /// The drift gate here is the one table named, unlike [`Database::catalog`], whose document
    /// covers every table and so has to answer for every table: a table of its own that moved on
    /// says nothing about this one.
    pub fn describe_table(&self, name: &str) -> Result<Value, DatabaseError> {
        if let Some(drift) = self.drift_of(name) {
            let drifted = [DriftedTable {
                name: name.to_string(),
                drift,
            }];
            return Err(DatabaseError::Schema(self.drift_error(&drifted)));
        }
        let table = self.tables.get(name).ok_or_else(|| {
            DatabaseError::Schema(HaysonSchemaError(format!("the catalog declares no table \"{name}\"")))
        })?;
        Ok(describe_table_document(table))
    }

    fn drift_of(&self, name: &str) -> Option<TableDrift> {
        let table = self.tables.get(name)?;
        let Some(entry) = self.catalog.tables.get(name) else {
            return Some(TableDrift::Absent);
        };
        if shape_of(&table_schema_of(name, entry)) != shape_of(table.schema()) {
            return Some(TableDrift::Schema {
                revision: entry.revision,
            });
        }
        let declared = expand_indexes(name, entry);
        let served = table.declared_indexes();
        let unserved = index_names_missing_from(&declared, served);
        let stale = index_names_missing_from(served, &declared);
        if unserved.is_empty() && stale.is_empty() {
            return None;
        }
        Some(TableDrift::Indexes { unserved, stale })
    }

    fn drifted_tables(&self) -> Vec<DriftedTable> {
        let mut names: Vec<&String> = self.tables.keys().collect();
        names.sort();
        names
            .into_iter()
            .filter_map(|name| {
                self.drift_of(name).map(|drift| DriftedTable {
                    name: name.clone(),
                    drift,
                })
            })
            .collect()
    }

    fn drift_error(&self, drifted: &[DriftedTable]) -> HaysonSchemaError {
        let detail = drifted
            .iter()
            .map(|DriftedTable { name, drift }| {
                let revision = self.tables.get(name).map(|table| table.schema().revision);
                let open = revision.map_or_else(|| "not open".to_string(), |at| format!("revision {at}"));
                format!(
                    "\"{name}\" is open here at {open} while the catalog {}",
                    stored_side_of(drift)
                )
            })
            .collect::<Vec<_>>()
            .join("; ");
        HaysonSchemaError(format!(
            "the catalog of \"{}\" moved on under this handle: {detail} — an open table keeps the schema and the index set it was opened with, so describing it from the catalog would describe a table this handle does not serve; reopen the database to work with the catalog on disk",
            self.layout.root().display()
        ))
    }

    /// The relations the catalog declares.
    ///
    /// @param table When given, keeps only the relations touching this table, on either end.
    pub fn relations(&self, table: Option<&str>) -> Vec<Relation> {
        relations_touching(&self.catalog, table)
    }

    /// Walks every declared relation and reports the records whose reference has no counterpart —
    /// a diagnosis, not a gate; nothing is repaired, removed or written.
    pub fn check_integrity(&self) -> io::Result<IntegrityReport> {
        let refs: HashMap<String, &Table> = self.tables.iter().map(|(name, table)| (name.clone(), table)).collect();
        check_integrity(&self.catalog, &refs)
    }

    /// Catches up with everything other processes did to this directory since this handle last
    /// looked — the port of the TS `Database.refresh()`, in the same two steps and the same order:
    ///
    /// 1. `hayson.json` is read again, and every table it now declares that this handle does not
    ///    hold is opened, through the very code path `open` uses per table;
    /// 2. every table already open here is caught up through [`Table::refresh`].
    ///
    /// Answers whether anything at all changed. A catalog that lost a table on disk does not close
    /// the one already open here: a live table is not dropped out from under a caller holding it.
    pub fn refresh(&mut self) -> Result<bool, DatabaseError> {
        let mut changed = false;
        if let Some(catalog) = read_catalog(self.layout.metadata_path())? {
            if catalog != self.catalog {
                self.catalog = catalog;
                changed = true;
            }
        }

        let missing: Vec<String> = self
            .catalog
            .tables
            .keys()
            .filter(|name| !self.tables.contains_key(*name))
            .cloned()
            .collect();
        for name in missing {
            let Some(entry) = self.catalog.tables.get(&name) else {
                continue;
            };
            let (table, outcome) = open_catalog_table(
                &self.layout,
                &name,
                &self.catalog,
                entry,
                self.snapshot_every,
                self.cipher.as_ref(),
                self.follow,
                self.readonly,
                compaction_of(self.compaction, &self.catalog, Some(entry)),
                self.cache,
            )?;
            self.tables.insert(name.clone(), table);
            self.restore_outcomes.insert(name, outcome);
            changed = true;
        }

        for (name, table) in self.tables.iter_mut() {
            let outcome = table
                .refresh()
                .map_err(|error| DatabaseError::Table(name.clone(), TableError::Io(error)))?;
            changed |= outcome.changed();
        }

        // NOTE: triggers are resynced after the catalog re-read — a table opened above has none yet.
        sync_table_triggers(&self.catalog, &mut self.tables);
        Ok(changed)
    }

    /// Starts a unit of work: stage `put`/`patch`/`delete` across several tables, then pass it to
    /// [`Database::commit`], which lands it whole or not at all.
    pub fn unit_of_work(&mut self) -> UnitOfWork {
        UnitOfWork::new()
    }

    /// Runs every operation staged in `unit`, across the tables it names, through the six-step
    /// commit `txn::UnitOfWork::commit` documents (resolve, barrier, prepare, journal, write,
    /// cleanup) — refused when the database is readonly.
    ///
    /// The unit is borrowed, not consumed: a commit refused before anything was journalled leaves
    /// it staged and retriable, exactly as the TS engine leaves it.
    pub fn commit(&mut self, unit: &mut UnitOfWork) -> Result<UnitOfWorkResult, DatabaseError> {
        let prepared = self.prepare_commit(unit)?;
        unit.journal(&prepared, &self.layout.journal_path())?;
        self.apply_commit(prepared)
    }

    /// The first phase of [`Database::commit`] — the readonly refusal, the lock renewal and
    /// [`UnitOfWork::prepare`]. Nothing is appended and no file is written, so a refusal from here
    /// leaves `unit` staged and retriable exactly as a refused `commit` does.
    ///
    /// Split out for a caller that keeps the database behind a lock and wants to release it across
    /// the journal write, which touches no table: prepare here, [`UnitOfWork::journal`] with the
    /// database let go, then [`Database::apply_commit`] once it is held again. The caller must stay
    /// the only writer across all three, or the burned `_seq` numbers would land out of order.
    pub fn prepare_commit(&mut self, unit: &mut UnitOfWork) -> Result<PreparedUnit, DatabaseError> {
        if self.readonly {
            return Err(DatabaseError::ReadOnly {
                dir: self.layout.root().to_path_buf(),
                operation: "commit".to_string(),
            });
        }
        self.renew_lock()?;
        Ok(unit.prepare(&mut self.tables)?)
    }

    /// The last phase of [`Database::commit`]: every prepared entry appended and the journal
    /// removed. See [`Database::prepare_commit`] for the contract a caller running the phases apart
    /// has to keep.
    pub fn apply_commit(&mut self, prepared: PreparedUnit) -> Result<UnitOfWorkResult, DatabaseError> {
        let journal_path = self.layout.journal_path();
        Ok(apply_prepared(prepared, &mut self.tables, &journal_path)?)
    }

    /// Refuses the operation when this handle is not the writer, and renews the lease while it is
    /// — so a lock stolen out from under a live handle stops the next write instead of letting it
    /// through into a directory somebody else now owns.
    fn assert_writable(&self, operation: &str) -> Result<(), DatabaseError> {
        if self.readonly {
            return Err(DatabaseError::ReadOnly {
                dir: self.layout.root().to_path_buf(),
                operation: operation.to_string(),
            });
        }
        self.renew_lock()
    }

    fn write_catalog_to_disk(&self, catalog: &Catalog) -> Result<(), DatabaseError> {
        write_catalog(self.layout.metadata_path(), catalog).map_err(catalog_write_error)
    }

    /// Creates a table, or brings an existing one up to a newer (additive) schema — `catalog::build`
    /// does not check revision compatibility yet (see its own doc comment), so a genuinely breaking
    /// change is not refused here either; that check is still on the TODO list this method inherits.
    ///
    /// Disk first: the catalog is written before `self.catalog`/`self.tables` change, so a failure
    /// past that point is the only way this call can leave the two out of step with each other.
    ///
    /// The handle comes from `open_catalog_table` — the same helper `open()`/`refresh()` use — so a
    /// table migrated here carries exactly what it would carry after a restart, triggers included.
    ///
    /// `name` is the key the table lands under in the catalog, and the schema's own `table` is
    /// carried into every log line and index snapshot written for it. The two disagreeing is
    /// refused rather than resolved: the catalog key wins wherever they are read back, so accepting
    /// the spec would silently drop the name its author wrote.
    pub fn create_table(&mut self, name: &str, schema_spec: Value) -> Result<(), DatabaseError> {
        self.assert_writable("createTable")?;
        let schema = parse_schema(&schema_spec)?;
        if schema.table != name {
            return Err(DatabaseError::Schema(HaysonSchemaError(format!(
                "schema declares table \"{}\" but it is being created as \"{name}\" — name them the same",
                schema.table
            ))));
        }
        let now = now_ms();
        let next_catalog = crate::catalog::build::create_table(&self.catalog, name, &schema, now);
        self.write_catalog_to_disk(&next_catalog)?;

        let next_entry = &next_catalog.tables[name];
        let (table, outcome) = open_catalog_table(
            &self.layout,
            name,
            &next_catalog,
            next_entry,
            self.snapshot_every,
            self.cipher.as_ref(),
            self.follow,
            self.readonly,
            compaction_of(self.compaction, &next_catalog, Some(next_entry)),
            self.cache,
        )?;

        self.catalog = next_catalog;
        self.tables.insert(name.to_string(), table);
        self.restore_outcomes.insert(name.to_string(), outcome);
        Ok(())
    }

    /// Takes a field off the schema of a table without touching its log — records written before
    /// the drop keep carrying the field on disk, reads just stop handing it back.
    ///
    /// Refuses (leaving disk and `self` untouched) when the field is the primary key, is what the
    /// table partitions by, or an index/check/relation of the catalog still depends on it
    /// (`catalog::build::drop_field`, unconditional on `force`) — and, unless `force` is `true`,
    /// when a live record still holds a value for it (checked here, against the open table, per the
    /// TODO `catalog::build::drop_field` leaves for this layer).
    ///
    /// The declaration check runs first on purpose: with both refusals due the caller is told about
    /// the one `force` cannot get them past, and the error class matches what the field's
    /// dependents are — see `packages/hayson/test/drop-pola-kolejnosc-odmowy.test.ts`.
    pub fn drop_field(&mut self, table: &str, field: &str, force: bool) -> Result<(), DatabaseError> {
        self.assert_writable("dropField")?;
        self.flush_table(table)?;
        let existing = self.tables.get(table).ok_or_else(|| {
            DatabaseError::Catalog(format!(
                "table \"{table}\" does not exist in \"{}\"",
                self.layout.root().display()
            ))
        })?;

        let now = now_ms();
        let (next_catalog, next_entry) = crate::catalog::build::drop_field(&self.catalog, table, field, now)?;

        if !force {
            assert_no_residual_values(table, existing, field)?;
        }

        self.write_catalog_to_disk(&next_catalog)?;

        let (mut table_handle, _) = open_catalog_table(
            &self.layout,
            table,
            &next_catalog,
            &next_entry,
            self.snapshot_every,
            self.cipher.as_ref(),
            self.follow,
            self.readonly,
            compaction_of(self.compaction, &next_catalog, Some(&next_entry)),
            self.cache,
        )?;
        // NOTE: the .idx on disk still projects the dropped field — rewrite it before anyone restores from it.
        table_handle
            .snapshot()
            .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;

        self.catalog = next_catalog;
        self.tables.insert(table.to_string(), table_handle);
        Ok(())
    }

    /// Adds a named index to a table of the catalog — a unique index is refused whole if a live
    /// record already shares a value on it (`catalog::declare_index`, checked against the open
    /// table). The reopened table's registry is built from the catalog's full expanded index list
    /// (`catalog::expand_indexes`), so a composite or `where`-scoped index is enforced/served right
    /// away, same as the single-field unconditional case `catalog::expand_indexes` derives as sugar.
    ///
    /// Reopened through `open_catalog_table` — the same helper `open()`/`refresh()` use — so the
    /// handle keeps its snapshot pairing, and the stale `.idx` is rewritten before this returns.
    pub fn declare_index(&mut self, table: &str, index_spec: Value) -> Result<(), DatabaseError> {
        self.assert_writable("declareIndex")?;
        self.flush_table(table)?;
        let existing = self.tables.get(table).ok_or_else(|| {
            DatabaseError::Catalog(format!(
                "table \"{table}\" does not exist in \"{}\"",
                self.layout.root().display()
            ))
        })?;

        let now = now_ms();
        let next_catalog = crate::catalog::declare_index(&self.catalog, table, &index_spec, existing, now)?;
        self.write_catalog_to_disk(&next_catalog)?;

        let next_entry = &next_catalog.tables[table];
        let (mut table_handle, _) = open_catalog_table(
            &self.layout,
            table,
            &next_catalog,
            next_entry,
            self.snapshot_every,
            self.cipher.as_ref(),
            self.follow,
            self.readonly,
            compaction_of(self.compaction, &next_catalog, Some(next_entry)),
            self.cache,
        )?;
        // NOTE: the .idx on disk projects nothing of the new index — rewrite it before anyone restores from it.
        table_handle
            .snapshot()
            .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;

        self.catalog = next_catalog;
        self.tables.insert(table.to_string(), table_handle);
        Ok(())
    }

    /// Declares an AFTER trigger (`catalog::build::declare_trigger`, which refuses a spec naming a
    /// table the catalog does not declare) and refreshes every open table's live trigger set from
    /// the result — both the source table `trigger_spec.table` names and the effect table
    /// `trigger_spec.effect.table` names, since either could be a live table this call did not
    /// itself just open. Disk first, same discipline as `create_table`/`declare_index`.
    pub fn declare_trigger(&mut self, trigger_spec: Value) -> Result<(), DatabaseError> {
        self.assert_writable("declareTrigger")?;

        let now = now_ms();
        let next_catalog = crate::catalog::declare_trigger(&self.catalog, &trigger_spec, now)?;
        self.write_catalog_to_disk(&next_catalog)?;

        self.catalog = next_catalog;
        sync_table_triggers(&self.catalog, &mut self.tables);
        Ok(())
    }

    /// Appends every write staged against any table of this database and not yet committed — the
    /// TS `Database.flush()`. A no-op for a table with nothing in flight, which is every table of a
    /// database whose writes all went through `put`/`patch`/`delete`.
    pub fn flush(&mut self) -> Result<(), DatabaseError> {
        for (name, table) in &mut self.tables {
            table
                .flush()
                .map_err(|error| DatabaseError::Table(name.clone(), error))?;
        }
        Ok(())
    }

    /// Commits what is staged against one table. Every path that reopens a table over its log, or
    /// rewrites that log, has to run this first: the handle it replaces takes its staging window
    /// with it.
    fn flush_table(&mut self, table: &str) -> Result<(), DatabaseError> {
        let Some(handle) = self.tables.get_mut(table) else {
            return Ok(());
        };
        handle
            .flush()
            .map_err(|error| DatabaseError::Table(table.to_string(), error))
    }

    /// Appends whatever every table still holds staged, snapshots the ones with writes since their
    /// last snapshot (see [`Table::needs_snapshot`]), then releases the writer lock, if this handle
    /// holds one — mirrors the TS `Database.close()`. The catalog needs no flush of its own: it is
    /// already durable (`fsync`) by the time the call that wrote it returns.
    pub fn close(mut self) -> io::Result<()> {
        if !self.readonly {
            for table in self.tables.values_mut() {
                // NOTE: flush before the snapshot — a staged write is in neither the log nor it.
                table.flush()?;
                if table.needs_snapshot() {
                    table.snapshot()?;
                }
            }
        }
        if let Some(mut lock) = self.lock.take() {
            lock.release();
        }
        Ok(())
    }
}

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

    use super::*;
    use crate::catalog::create_table;
    use crate::catalog::types::{CatalogIndexKind, IndexEntry};
    use crate::checkpoint::write_snapshot;
    use crate::registry::table_registry::TableRegistry;
    use crate::schema::parse_schema;
    use crate::storage::{serialize_entry, LogMeta, LogOp};
    use crate::txn::journal::{write_journal, Journal, JournalEntry, JournalHeader, JOURNAL_VERSION};
    use crate::txn::JOURNAL_FILE;

    const MANIFEST_FILE: &str = DEFAULT_METADATA_FILE;

    fn flat(dir: &Path) -> Layout {
        Layout::new(dir, crate::database::LayoutKind::Flat, MANIFEST_FILE)
    }

    fn log_path(dir: &Path, table: &str) -> std::path::PathBuf {
        flat(dir).log_path(table)
    }

    /// The binding above this crate keeps the database behind an `RwLock` shared across the libuv
    /// pool, which needs `Sync` as well as `Send`. Nothing here runs: the call fails to compile the
    /// moment an interior-mutability cell that is not thread-safe reappears anywhere below.
    #[test]
    fn a_database_and_its_tables_can_be_shared_across_threads() {
        fn shareable<T: Send + Sync>() {}
        shareable::<Database>();
        shareable::<Table>();
        shareable::<crate::registry::fulltext_index::FulltextIndex>();
    }

    fn index_path(dir: &Path, table: &str) -> std::path::PathBuf {
        flat(dir).index_path(table)
    }

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

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

    fn catalog_of(schemas: &[TableSchema]) -> Catalog {
        let mut catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        for schema in schemas {
            catalog = create_table(&catalog, &schema.table, schema, 1_000);
        }
        catalog
    }

    fn write_catalog_to(dir: &Path, catalog: &Catalog) {
        write_catalog(&dir.join(MANIFEST_FILE), catalog).unwrap();
    }

    #[test]
    fn open_builds_a_database_with_every_declared_table_available() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));

        let mut db = Database::open(dir.path()).unwrap();
        assert_eq!(db.table_names(), vec!["orders", "users"]);

        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        assert_eq!(
            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        db.close().unwrap();
    }

    #[test]
    fn open_restores_the_registry_from_a_paired_snapshot_instead_of_scanning_the_log() {
        let dir = tempdir().unwrap();
        let catalog = catalog_of(&[users_schema()]);
        write_catalog_to(dir.path(), &catalog);

        let schema = table_schema_of("users", &catalog.tables["users"]);
        let hash = schema_fingerprint(&schema, &expand_indexes("users", &catalog.tables["users"]));
        let log_path = log_path(dir.path(), "users");
        let mut log = crate::storage::LogFile::open(&log_path, false).unwrap();
        log.append(&[serialize_entry(
            &LogMeta {
                key: "u1".into(),
                op: LogOp::Put,
                seq: 1,
                ts: 1,
            },
            Some(&{
                let mut record = crate::schema::HaysonRecord::new();
                record.insert("id".into(), json!("u1"));
                record.insert("name".into(), json!("Ania"));
                record
            }),
        )])
        .unwrap();
        let registry = TableRegistry::build_from_log(schema.clone(), &log).unwrap();
        write_snapshot(&index_path(dir.path(), "users"), &registry, "users", &hash, 0, &log).unwrap();

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(db.restore_outcome("users"), Some(RestoreOutcome::Restored));
        assert_eq!(
            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        db.close().unwrap();
    }

    #[test]
    fn open_finishes_an_unfinished_journal_before_returning() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));

        {
            let mut users = Table::open(users_schema(), log_path(dir.path(), "users")).unwrap();
            let mut orders = Table::open(orders_schema(), log_path(dir.path(), "orders")).unwrap();
            let user_entries = users
                .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
                .unwrap();
            let order_entries = orders
                .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "o1", "userId": "u1" }))])
                .unwrap();

            let mut entries = Vec::new();
            for entry in &user_entries {
                entries.push(JournalEntry {
                    t: "users".into(),
                    s: entry.seq(),
                    l: entry.line().to_string(),
                });
            }
            for entry in &order_entries {
                entries.push(JournalEntry {
                    t: "orders".into(),
                    s: entry.seq(),
                    l: entry.line().to_string(),
                });
            }
            write_journal(
                &dir.path().join(JOURNAL_FILE),
                &Journal {
                    header: JournalHeader {
                        journal_version: JOURNAL_VERSION,
                        kind: "unit".into(),
                        id: "crash".into(),
                        started_at: 0,
                        tables: vec!["users".into(), "orders".into()],
                    },
                    entries,
                },
            )
            .unwrap();
            // process dies here: prepared and journalled, never applied to either log.
        }

        assert!(dir.path().join(JOURNAL_FILE).exists());
        let db = Database::open(dir.path()).unwrap();
        assert!(!dir.path().join(JOURNAL_FILE).exists());
        assert_eq!(
            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        assert_eq!(
            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
            Some(&json!("u1"))
        );
        db.close().unwrap();
    }

    #[test]
    fn a_second_open_of_the_same_directory_is_denied_while_the_first_still_holds_the_lock() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));

        let _first = Database::open(dir.path()).unwrap();
        let second = Database::open_with(
            dir.path(),
            OpenOptions {
                readonly: false,
                lock: LockMode::Enabled(LockOptions {
                    timeout_ms: Some(0),
                    ..Default::default()
                }),
                ..Default::default()
            },
        );
        assert!(matches!(second, Err(DatabaseError::Lock(_))));
    }

    fn lock_path_of(dir: &Path) -> std::path::PathBuf {
        dir.join(crate::lock::LOCK_FILE)
    }

    fn open_refusing(dir: &Path) -> DatabaseError {
        Database::open_with(
            dir,
            OpenOptions {
                lock: LockMode::Enabled(LockOptions {
                    timeout_ms: Some(0),
                    ..Default::default()
                }),
                ..Default::default()
            },
        )
        .map(|_| ())
        .expect_err("this directory must not open")
    }

    #[test]
    fn a_directory_with_logs_but_no_catalog_is_refused_instead_of_opening_empty() {
        let dir = tempdir().unwrap();
        let mut db = Database::open(dir.path()).unwrap();
        db.create_table(
            "users",
            json!({
                "schemaVersion": 1,
                "table": "users",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "name": { "type": "string", "required": true },
                },
            }),
        )
        .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        db.close().unwrap();

        let log = fs::read(dir.path().join("users.jsonl")).unwrap();
        fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();

        let refusal = open_refusing(dir.path()).to_string();
        assert!(
            refusal.contains("users.jsonl"),
            "the refusal has to name the log it found: {refusal}"
        );
        assert!(
            refusal.contains(MANIFEST_FILE),
            "the refusal has to name the catalog: {refusal}"
        );

        // The refusal is worth nothing if it already wrote a fresh catalog over the data it refused.
        assert!(
            !dir.path().join(MANIFEST_FILE).exists(),
            "a refused open must not leave a catalog"
        );
        assert_eq!(
            fs::read(dir.path().join("users.jsonl")).unwrap(),
            log,
            "the log must be untouched"
        );
    }

    #[test]
    fn an_empty_directory_and_one_holding_no_logs_still_open_as_a_new_database() {
        let empty = tempdir().unwrap();
        assert!(
            Database::open(empty.path()).is_ok(),
            "an empty directory is still a new database"
        );

        let littered = tempdir().unwrap();
        fs::write(littered.path().join("README.md"), "# not a table\n").unwrap();
        fs::create_dir(littered.path().join("users-backup")).unwrap();
        assert!(Database::open(littered.path()).is_ok(), "nothing here is a table log");
    }

    #[test]
    fn the_logs_of_a_partitioned_table_are_found_one_level_down_too() {
        let dir = tempdir().unwrap();
        fs::create_dir_all(dir.path().join("notes")).unwrap();
        fs::write(dir.path().join("notes").join("note.jsonl"), "").unwrap();

        let refusal = open_refusing(dir.path()).to_string();
        assert!(
            refusal.contains("note.jsonl"),
            "a partitioned log still counts as data: {refusal}"
        );
    }

    /// Every way the catalog read under the lock can refuse. The lock is taken before any of them
    /// runs (`stored_kind_of` is deliberately lenient), so each one is a chance to strand the lock
    /// file on a process that is still alive — which no staleness rule would ever reclaim.
    #[test]
    fn a_directory_refused_at_open_is_left_unlocked_whichever_way_the_catalog_was_broken() {
        for (case, document) in [
            ("not JSON at all", "{ this is not json".to_string()),
            (
                "no formatVersion",
                json!({ "catalogVersion": 2, "tables": {} }).to_string(),
            ),
            (
                "the format-1 marker",
                json!({ "hayson": 1, "formatVersion": 2, "tables": {} }).to_string(),
            ),
            (
                "a formatVersion from the future",
                json!({ "catalogVersion": 2, "formatVersion": 3, "tables": {} }).to_string(),
            ),
        ] {
            let dir = tempdir().unwrap();
            std::fs::write(dir.path().join(MANIFEST_FILE), &document).unwrap();

            open_refusing(dir.path());

            assert!(
                !lock_path_of(dir.path()).exists(),
                "{case}: the refused open left its lock file behind"
            );
        }
    }

    /// The point of releasing on refusal: the directory has to stay openable. Nothing else reclaims
    /// such a lock — its owner is the very process that got the refusal, and it is still running.
    #[test]
    fn a_directory_opens_normally_after_an_open_that_was_refused_a_broken_catalog() {
        let dir = tempdir().unwrap();
        std::fs::write(dir.path().join(MANIFEST_FILE), "{ this is not json").unwrap();

        open_refusing(dir.path());

        std::fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));

        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        db.close().unwrap();
    }

    /// A refusal raised past the catalog — the cipher is built after it parsed — has to unwind the
    /// same way, which is what putting the lock behind a guard instead of one `release` call buys.
    #[test]
    fn a_directory_refused_for_a_bad_cipher_key_is_left_unlocked_too() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));

        let error = Database::open_with(
            dir.path(),
            OpenOptions {
                cypher: Some(CipherOptions {
                    keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
                    key_id: None,
                    on_missing_key: None,
                }),
                ..OpenOptions::default()
            },
        )
        .map(|_| ())
        .unwrap_err();

        assert!(matches!(error, DatabaseError::Cipher(_)));
        assert!(!lock_path_of(dir.path()).exists());
        Database::open(dir.path()).unwrap().close().unwrap();
    }

    /// The other half of the guard: a successful open must still be holding the lock, not have
    /// released it on the way out of `open_locked`.
    #[test]
    fn an_open_that_succeeded_still_holds_the_lock_until_the_handle_closes() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));

        let held = Database::open(dir.path()).unwrap();
        assert!(lock_path_of(dir.path()).exists());
        assert!(matches!(open_refusing(dir.path()), DatabaseError::Lock(_)));

        held.close().unwrap();
        assert!(!lock_path_of(dir.path()).exists());
    }

    /// A second writer of a directory somebody else already holds — the port of the TS
    /// `open(dir, { lock: false })`, deliberately dangerous and only ever reached on request.
    fn open_skipping_the_lock(dir: &Path) -> Database {
        Database::open_with(
            dir,
            OpenOptions {
                lock: LockMode::Skip,
                ..Default::default()
            },
        )
        .unwrap()
    }

    fn declared_on_disk(dir: &Path, table: &str) -> Vec<String> {
        let catalog = read_catalog(&dir.join(MANIFEST_FILE))
            .unwrap()
            .expect("a catalog on disk");
        let mut names: Vec<String> = catalog.tables[table]
            .indexes
            .iter()
            .flatten()
            .map(|index| index.name.clone())
            .collect();
        names.sort();
        names
    }

    /// The fixture the three lock-skipping tests share: a directory with two tables, two records,
    /// and no live handle — the `usersWithTwoRecords` of the TS
    /// `deklarowanie-indeksu-wspolbieznie.test.ts`.
    fn two_tables_with_records(dir: &Path) {
        write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
        let mut db = Database::open(dir).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
            .unwrap();
        db.close().unwrap();
    }

    #[test]
    fn a_second_writer_that_skips_the_lock_opens_and_writes_beside_the_one_holding_it() {
        let dir = tempdir().unwrap();
        two_tables_with_records(dir.path());

        let mut locked = Database::open(dir.path()).unwrap();
        let mut unlocked = open_skipping_the_lock(dir.path());

        // Both are writers, on their own table each — the lock never came into it.
        locked
            .table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u3", "name": "Kasia" }), false)
            .unwrap();
        unlocked
            .table_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o1", "userId": "u1" }), false)
            .unwrap();

        // While a third handle asking for the lock is still refused, so nothing was released.
        let third = Database::open_with(
            dir.path(),
            OpenOptions {
                lock: LockMode::Enabled(LockOptions {
                    timeout_ms: Some(0),
                    ..Default::default()
                }),
                ..Default::default()
            },
        );
        assert!(matches!(third, Err(DatabaseError::Lock(_))));

        unlocked.close().unwrap();
        locked.close().unwrap();
    }

    /// The window `declare_index` documents: it writes the catalog it holds in memory, with no
    /// check against the disk, and the writer lock is the only guard. A handle that skipped the
    /// lock therefore overwrites what the other one declared. This is the TS behaviour, asserted
    /// so a change to it is a decision and not an accident.
    #[test]
    fn a_lock_skipping_handle_that_never_refreshed_loses_the_declaration_of_the_locked_one() {
        let dir = tempdir().unwrap();
        two_tables_with_records(dir.path());

        let mut locked = Database::open(dir.path()).unwrap();
        let mut unlocked = open_skipping_the_lock(dir.path());

        locked
            .declare_index(
                "users",
                json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
            )
            .unwrap();
        unlocked
            .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
            .unwrap();

        unlocked.close().unwrap();
        locked.close().unwrap();

        assert_eq!(declared_on_disk(dir.path(), "users"), vec!["by_id".to_string()]);
    }

    #[test]
    fn a_lock_skipping_handle_that_refreshes_first_keeps_both_declarations() {
        let dir = tempdir().unwrap();
        two_tables_with_records(dir.path());

        let mut locked = Database::open(dir.path()).unwrap();
        let mut unlocked = open_skipping_the_lock(dir.path());

        locked
            .declare_index(
                "users",
                json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
            )
            .unwrap();
        assert!(unlocked.refresh().unwrap(), "the catalog changed under it");
        unlocked
            .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
            .unwrap();

        unlocked.close().unwrap();
        locked.close().unwrap();

        assert_eq!(
            declared_on_disk(dir.path(), "users"),
            vec!["by_id".to_string(), "by_name".to_string()]
        );
    }

    /// A reader of the same directory as the live writer: readonly, so it takes no lock and the
    /// writer keeps its own — the shape a follower process has.
    fn open_follower(dir: &Path) -> Database {
        Database::open_with(
            dir,
            OpenOptions {
                readonly: true,
                lock: LockMode::default(),
                ..Default::default()
            },
        )
        .unwrap()
    }

    #[test]
    fn refresh_with_nothing_new_on_disk_reports_no_change() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut writer = Database::open(dir.path()).unwrap();
        writer
            .table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        let mut follower = open_follower(dir.path());
        follower.refresh().unwrap();

        assert!(!follower.refresh().unwrap());
        writer.close().unwrap();
    }

    #[test]
    fn refresh_picks_up_both_a_table_and_the_records_another_process_wrote() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut writer = Database::open(dir.path()).unwrap();
        let mut follower = open_follower(dir.path());
        assert_eq!(follower.table_names(), vec!["users"]);

        writer
            .table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        writer
            .create_table(
                "orders",
                json!({
                    "schemaVersion": 1,
                    "table": "orders",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "userId": { "type": "string", "required": true } },
                }),
            )
            .unwrap();
        writer
            .table_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o1", "userId": "u1" }), false)
            .unwrap();

        assert!(follower.refresh().unwrap());

        assert_eq!(follower.table_names(), vec!["orders", "users"]);
        assert_eq!(
            follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        assert_eq!(
            follower
                .table("orders")
                .unwrap()
                .get("o1")
                .unwrap()
                .unwrap()
                .get("userId"),
            Some(&json!("u1"))
        );
        writer.close().unwrap();
    }

    #[test]
    fn refresh_survives_a_compaction_run_by_the_writer_process() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut writer = Database::open(dir.path()).unwrap();
        let users = writer.table_mut("users").unwrap();
        users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
        users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
        users.patch("u1", &json!({ "name": "Anna" }), false).unwrap();
        let mut follower = open_follower(dir.path());
        assert_eq!(follower.table("users").unwrap().len(), 2);

        writer.table_mut("users").unwrap().compact().unwrap();
        assert!(follower.refresh().unwrap());

        assert_eq!(follower.table("users").unwrap().len(), 2);
        assert_eq!(
            follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Anna"))
        );
        assert_eq!(follower.table("users").unwrap().stats().gen, 1);
        writer.close().unwrap();
    }

    #[test]
    fn unit_of_work_commits_across_two_tables_and_the_database_sees_both() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
        let mut db = Database::open(dir.path()).unwrap();

        let mut unit = db.unit_of_work();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        unit.put("orders", json!({ "id": "o1", "userId": "u1" }));
        let result = db.commit(&mut unit).unwrap();
        assert_eq!(result.operations, 2);

        assert_eq!(
            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        assert_eq!(
            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
            Some(&json!("u1"))
        );
        db.close().unwrap();
    }

    #[test]
    fn create_table_on_an_open_database_adds_a_new_working_table_immediately() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();

        db.create_table(
            "orders",
            json!({
                "schemaVersion": 1,
                "table": "orders",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "userId": { "type": "string", "required": true },
                },
            }),
        )
        .unwrap();

        assert_eq!(db.table_names(), vec!["orders", "users"]);
        assert!(db.catalog.tables.contains_key("orders"));

        db.table_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o1", "userId": "u1" }), false)
            .unwrap();
        assert_eq!(
            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
            Some(&json!("u1"))
        );

        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
        assert!(reread.tables.contains_key("orders"));
        db.close().unwrap();
    }

    #[test]
    fn create_table_refuses_a_schema_naming_a_table_other_than_the_one_being_created() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();

        let error = db
            .create_table(
                "orders",
                json!({
                    "schemaVersion": 1,
                    "table": "invoices",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" } },
                }),
            )
            .expect_err("a schema naming another table is refused");

        assert!(matches!(error, DatabaseError::Schema(_)), "got {error:?}");
        let message = error.to_string();
        assert!(
            message.contains("invoices"),
            "the refusal names the schema's table: {message}"
        );
        assert!(
            message.contains("orders"),
            "the refusal names the key it was created under: {message}"
        );

        assert_eq!(db.table_names(), vec!["users"]);
        assert!(!db.catalog.tables.contains_key("orders"));
        assert!(!db.catalog.tables.contains_key("invoices"));
        assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
        db.close().unwrap();
    }

    #[test]
    fn create_table_still_takes_a_schema_naming_exactly_the_table_it_creates() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();

        db.create_table(
            "orders",
            json!({
                "schemaVersion": 1,
                "table": "orders",
                "primaryKey": "id",
                "fields": { "id": { "type": "string" } },
            }),
        )
        .unwrap();

        assert_eq!(db.table("orders").unwrap().schema().table, "orders");
        db.close().unwrap();
    }

    #[test]
    fn create_table_refuses_on_a_readonly_database_and_changes_nothing() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        // NOTE: a readonly open refuses a catalog whose logs are not on disk yet (F8) — make them.
        Database::open(dir.path()).unwrap().close().unwrap();
        let mut db = Database::open_with(
            dir.path(),
            OpenOptions {
                readonly: true,
                lock: LockMode::default(),
                ..Default::default()
            },
        )
        .unwrap();

        let error = db
            .create_table(
                "orders",
                json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
            )
            .unwrap_err();
        assert!(
            matches!(&error, DatabaseError::ReadOnly { dir: refused, operation } if refused == dir.path() && operation == "createTable"),
            "expected a typed readonly refusal, got {error:?}"
        );
        assert_eq!(db.table_names(), vec!["users"]);
    }

    #[test]
    fn create_table_refused_by_a_directory_on_the_catalog_temporary_path_keeps_the_typed_refusal() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
        // NOTE: a directory on the .tmp path fails the write with EISDIR whatever the uid is.
        let temporary = dir.path().join(format!("{MANIFEST_FILE}.tmp"));
        fs::create_dir(&temporary).unwrap();

        let error = db
            .create_table(
                "orders",
                json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
            )
            .unwrap_err();

        let refusal = match &error {
            DatabaseError::Io(inner) => AtomicTemporaryFileError::downcast(inner)
                .expect("a .tmp path that could not be claimed stays a temporary-file refusal"),
            other => panic!("expected an io error keeping the typed refusal, got {other:?}"),
        };
        assert_eq!(refusal.path, temporary);
        assert!(refusal.reason.contains("could not be cleared"), "{}", refusal.reason);
        assert_eq!(db.table_names(), vec!["users"]);
        assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
    }

    #[test]
    fn a_catalog_write_that_fails_for_any_other_reason_is_still_a_catalog_refusal() {
        let unreadable = catalog_write_error(Box::new(HaysonSchemaError("nope".into())));

        assert!(matches!(unreadable, DatabaseError::Catalog(_)), "got {unreadable:?}");
        assert!(matches!(
            catalog_write_error(Box::new(io::Error::from(io::ErrorKind::PermissionDenied))),
            DatabaseError::Catalog(_)
        ));
    }

    #[test]
    fn drop_field_without_dependents_removes_it_from_disk_and_from_memory() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        db.table_mut("users").unwrap().delete("u1", false).unwrap();

        db.drop_field("users", "name", false).unwrap();

        assert!(!db.table("users").unwrap().schema().fields.contains_key("name"));
        assert!(!db.catalog.tables["users"].fields.contains_key("name"));

        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
        assert!(!reread.tables["users"].fields.contains_key("name"));
        db.close().unwrap();
    }

    #[test]
    fn drop_field_refuses_a_live_residual_field_without_force() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();

        let error = db.drop_field("users", "name", false);
        assert!(error.is_err());
        assert!(db.catalog.tables["users"].fields.contains_key("name"));
        assert!(db.table("users").unwrap().schema().fields.contains_key("name"));

        db.drop_field("users", "name", true).unwrap();
        assert!(!db.catalog.tables["users"].fields.contains_key("name"));
        db.close().unwrap();
    }

    fn users_catalog_indexing_name() -> Catalog {
        let mut catalog = catalog_of(&[users_schema()]);
        catalog.tables.get_mut("users").unwrap().indexes = Some(vec![IndexEntry {
            name: "by_name".to_string(),
            fields: vec!["name".to_string()],
            kind: CatalogIndexKind::Exact,
            unique: false,
            where_: None,
            expire_after_seconds: None,
        }]);
        catalog
    }

    #[test]
    fn drop_field_names_the_declaration_when_residual_records_also_block_it() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &users_catalog_indexing_name());
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();

        let error = db.drop_field("users", "name", false).unwrap_err();

        assert!(
            matches!(error, DatabaseError::Schema(_)),
            "expected a schema refusal, got {error:?}"
        );
        let message = error.to_string();
        assert!(message.contains("by_name"), "{message}");
        assert!(!message.contains("live record(s)"), "{message}");
        assert!(db.catalog.tables["users"].fields.contains_key("name"));
        db.close().unwrap();
    }

    #[test]
    fn drop_field_names_the_declaration_when_it_is_the_only_thing_blocking_it() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &users_catalog_indexing_name());
        let mut db = Database::open(dir.path()).unwrap();

        let error = db.drop_field("users", "name", false).unwrap_err();

        assert!(
            matches!(error, DatabaseError::Schema(_)),
            "expected a schema refusal, got {error:?}"
        );
        assert!(error.to_string().contains("by_name"));
        assert!(db.catalog.tables["users"].fields.contains_key("name"));
        db.close().unwrap();
    }

    #[test]
    fn drop_field_names_the_residual_records_when_no_declaration_blocks_it() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();

        let error = db.drop_field("users", "name", false).unwrap_err();

        assert!(
            matches!(
                &error,
                DatabaseError::FieldInUse { table, field, records }
                    if table == "users" && field == "name" && *records == 1
            ),
            "expected a typed field-in-use refusal, got {error:?}"
        );
        assert!(db.catalog.tables["users"].fields.contains_key("name"));
        db.close().unwrap();
    }

    #[test]
    fn drop_field_keeps_refusing_a_declared_dependency_even_with_force() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &users_catalog_indexing_name());
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();

        let error = db.drop_field("users", "name", true).unwrap_err();

        assert!(
            matches!(error, DatabaseError::Schema(_)),
            "expected a schema refusal, got {error:?}"
        );
        assert!(error.to_string().contains("by_name"));
        db.close().unwrap();
    }

    #[test]
    fn declare_index_with_a_conflicting_unique_value_refuses_and_changes_nothing() {
        let dir = tempdir().unwrap();
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "email": { "type": "string" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[schema]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "email": "same@example.com" }), false)
            .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u2", "email": "same@example.com" }), false)
            .unwrap();

        let error = db.declare_index(
            "users",
            json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
        );
        assert!(error.is_err());
        assert!(db.catalog.tables["users"].indexes.is_none());

        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
        assert!(reread.tables["users"].indexes.is_none());
        db.close().unwrap();
    }

    #[test]
    fn declare_index_without_a_conflict_covers_records_written_before_the_declaration() {
        let dir = tempdir().unwrap();
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "email": { "type": "string" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[schema]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
            .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
            .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u3", "email": "kasia@example.com" }), false)
            .unwrap();

        db.declare_index(
            "users",
            json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
        )
        .unwrap();

        assert_eq!(db.catalog.tables["users"].indexes.as_ref().unwrap().len(), 1);

        let spec = QueryBuilder::new()
            .r#where(
                "email",
                crate::query::predicates::ComparisonOp::Eq,
                json!("piotr@example.com"),
            )
            .spec();
        let found = db.table("users").unwrap().query(&spec).unwrap();
        assert_eq!(found.keys, vec!["u2".to_string()]);

        let conflicting = db
            .table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u4", "email": "piotr@example.com" }), false);
        assert!(conflicting.is_err());
        db.close().unwrap();
    }

    #[test]
    fn declare_index_snapshots_the_reopened_table_and_keeps_snapshotting_it_afterwards() {
        let dir = tempdir().unwrap();
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "email": { "type": "string" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[schema]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
            .unwrap();

        db.declare_index(
            "users",
            json!({ "name": "users_email_exact", "fields": ["email"], "kind": "exact" }),
        )
        .unwrap();

        assert!(
            index_path(dir.path(), "users").exists(),
            "declare_index owes the table a snapshot carrying the new index's projection"
        );

        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
            .unwrap();
        assert!(
            db.table("users").unwrap().needs_snapshot(),
            "the handle declare_index left behind must still be able to snapshot itself"
        );
        db.close().unwrap();

        let reopened = Database::open(dir.path()).unwrap();
        assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
        let spec = QueryBuilder::new()
            .r#where(
                "email",
                crate::query::predicates::ComparisonOp::Eq,
                json!("piotr@example.com"),
            )
            .spec();
        assert_eq!(
            reopened.table("users").unwrap().query(&spec).unwrap().keys,
            vec!["u2".to_string()]
        );
        assert_eq!(
            declared_on_disk(dir.path(), "users"),
            vec!["users_email_exact".to_string()]
        );
        reopened.close().unwrap();
    }

    #[test]
    fn a_database_wide_sync_default_reaches_a_table_declaring_none_of_its_own() {
        let dir = tempdir().unwrap();
        let mut catalog = catalog_of(&[users_schema()]);
        catalog.defaults = Some(crate::catalog::TableOptions {
            sync: Some(crate::catalog::CatalogSyncMode::Never),
            ..Default::default()
        });
        write_catalog_to(dir.path(), &catalog);

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("users").unwrap().sync_mode(),
            crate::catalog::CatalogSyncMode::Never
        );
        db.close().unwrap();
    }

    #[test]
    fn a_table_declaring_its_own_sync_mode_beats_the_database_wide_default() {
        let dir = tempdir().unwrap();
        let mut catalog = catalog_of(&[users_schema(), orders_schema()]);
        catalog.defaults = Some(crate::catalog::TableOptions {
            sync: Some(crate::catalog::CatalogSyncMode::Never),
            ..Default::default()
        });
        catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
            sync: Some(crate::catalog::CatalogSyncMode::Commit),
            ..Default::default()
        });
        write_catalog_to(dir.path(), &catalog);

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("users").unwrap().sync_mode(),
            crate::catalog::CatalogSyncMode::Commit
        );
        assert_eq!(
            db.table("orders").unwrap().sync_mode(),
            crate::catalog::CatalogSyncMode::Never
        );
        db.close().unwrap();
    }

    #[test]
    fn a_table_declaring_validation_warn_takes_a_write_with_a_stray_field_and_stores_it_without_that_field() {
        let dir = tempdir().unwrap();
        let mut catalog = catalog_of(&[users_schema()]);
        catalog.tables.get_mut("users").unwrap().validation = Some(crate::catalog::CatalogValidation::Warn);
        write_catalog_to(dir.path(), &catalog);

        let mut db = Database::open(dir.path()).unwrap();
        let document = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });
        let table = db.table_mut("users").unwrap();
        let record = table.put(&document, false).unwrap();

        assert_eq!(record.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()]
        );
        db.close().unwrap();

        let reopened = Database::open(dir.path()).unwrap();
        let stored = reopened.table("users").unwrap().get("u1").unwrap().unwrap();
        assert_eq!(stored.get("name"), Some(&json!("Ania")));
        assert_eq!(stored.get("nickname"), None);
        reopened.close().unwrap();
    }

    #[test]
    fn a_table_declaring_no_validation_refuses_a_write_with_a_stray_field() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));

        let mut db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("users").unwrap().validation(),
            crate::catalog::CatalogValidation::Strict
        );
        let error = db
            .table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }), false)
            .unwrap_err();

        assert!(error.to_string().contains("is not declared in the schema"), "{error}");
        db.close().unwrap();
    }

    fn deprecating_age(dir: &Path, validation: Option<crate::catalog::CatalogValidation>, note: &str) {
        let mut catalog = catalog_of(&[users_with_age_schema()]);
        let entry = catalog.tables.get_mut("users").unwrap();
        entry.validation = validation;
        entry.fields.get_mut("age").unwrap().deprecated = Some(note.to_string());
        write_catalog_to(dir, &catalog);
    }

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

    #[test]
    fn a_write_touching_a_field_the_catalog_marks_deprecated_warns_with_the_field_and_the_note() {
        let dir = tempdir().unwrap();
        deprecating_age(dir.path(), None, "use \"bornAt\" instead");

        let mut db = Database::open(dir.path()).unwrap();
        let table = db.table_mut("users").unwrap();
        let touching = json!({ "id": "u1", "name": "Ania", "age": 30 });
        let record = table.put(&touching, false).unwrap();

        assert_eq!(record.get("age"), Some(&json!(30)));
        assert_eq!(
            table.deprecated_field_warnings(&touching),
            vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()]
        );
        assert!(table
            .deprecated_field_warnings(&json!({ "id": "u2", "name": "Ola" }))
            .is_empty());
        db.close().unwrap();
    }

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

        for validation in [None, Some(crate::catalog::CatalogValidation::Warn)] {
            let dir = tempdir().unwrap();
            deprecating_age(dir.path(), validation, "use \"bornAt\" instead");

            let mut db = Database::open(dir.path()).unwrap();
            let table = db.table_mut("users").unwrap();
            table.put(&document, false).unwrap();
            assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
            table.put(&document, false).unwrap();
            assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
            db.close().unwrap();

            let reopened = Database::open(dir.path()).unwrap();
            assert_eq!(
                reopened.table("users").unwrap().deprecated_field_warnings(&document),
                expected
            );
            reopened.close().unwrap();
        }
    }

    #[test]
    fn a_catalog_marking_no_field_deprecated_leaves_every_table_without_the_warning() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_with_age_schema()]));

        let db = Database::open(dir.path()).unwrap();
        let table = db.table("users").unwrap();

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

    #[test]
    fn a_catalog_naming_no_sync_anywhere_leaves_every_table_committing() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("users").unwrap().sync_mode(),
            crate::catalog::CatalogSyncMode::Commit
        );
        db.close().unwrap();
    }

    #[test]
    fn a_sync_never_table_survives_a_close_and_reopen_with_its_writes_and_its_mode() {
        let dir = tempdir().unwrap();
        let mut catalog = catalog_of(&[users_schema()]);
        catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
            sync: Some(crate::catalog::CatalogSyncMode::Never),
            ..Default::default()
        });
        write_catalog_to(dir.path(), &catalog);

        {
            let mut db = Database::open(dir.path()).unwrap();
            db.table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();
            db.close().unwrap();
        }

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("users").unwrap().sync_mode(),
            crate::catalog::CatalogSyncMode::Never
        );
        assert_eq!(
            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        db.close().unwrap();
    }

    #[test]
    fn a_database_wide_default_names_the_link_parsing_mode_a_table_declaring_none_opens_with() {
        let dir = tempdir().unwrap();
        let pages_schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "pages",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
            },
        }))
        .unwrap();
        let mut catalog = catalog_of(&[pages_schema]);
        catalog.defaults = Some(crate::catalog::TableOptions {
            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
            ..Default::default()
        });
        write_catalog_to(dir.path(), &catalog);

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("pages").unwrap().link_parsing(),
            crate::catalog::LinkParsing::Lazy
        );
        db.close().unwrap();
    }

    #[test]
    fn a_table_declaring_its_own_link_parsing_mode_beats_the_database_wide_default() {
        let dir = tempdir().unwrap();
        let pages_schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "pages",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
            },
        }))
        .unwrap();
        let mut catalog = catalog_of(&[pages_schema]);
        catalog.defaults = Some(crate::catalog::TableOptions {
            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
            ..Default::default()
        });
        catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
            link_parsing: Some(crate::catalog::LinkParsing::Eager),
            ..Default::default()
        });
        write_catalog_to(dir.path(), &catalog);

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("pages").unwrap().link_parsing(),
            crate::catalog::LinkParsing::Eager
        );
        db.close().unwrap();
    }

    #[test]
    fn a_database_wide_default_names_the_image_dir_a_table_declaring_none_writes_under() {
        let dir = tempdir().unwrap();
        let docs_schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "cover": { "type": "image" },
            },
        }))
        .unwrap();
        let mut catalog = catalog_of(&[docs_schema]);
        catalog.defaults = Some(crate::catalog::TableOptions {
            image_dir: Some("shared-images".to_string()),
            ..Default::default()
        });
        write_catalog_to(dir.path(), &catalog);

        let mut db = Database::open(dir.path()).unwrap();
        let record = db
            .table_mut("docs")
            .unwrap()
            .put(
                &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
                false,
            )
            .unwrap();
        let physical_path = record
            .get("cover")
            .unwrap()
            .get("physicalPath")
            .unwrap()
            .as_str()
            .unwrap();

        assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
        assert!(dir.path().join("shared-images").join(physical_path).exists());
        db.close().unwrap();
    }

    #[test]
    fn declare_index_of_a_composite_field_pair_serves_a_query_matching_both_fields() {
        let dir = tempdir().unwrap();
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "orders",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "userId": { "type": "string" },
                "status": { "type": "string" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[schema]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
            .unwrap();
        db.table_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
            .unwrap();
        db.table_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o3", "userId": "u2", "status": "open" }), false)
            .unwrap();

        db.declare_index(
            "orders",
            json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
        )
        .unwrap();

        let spec = QueryBuilder::new()
            .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
            .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("open"))
            .spec();
        let found = db.table("orders").unwrap().query(&spec).unwrap();
        assert_eq!(found.keys, vec!["o1".to_string()]);
        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
        assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
        assert_eq!(found.plan.candidates, 1);
        db.close().unwrap();
    }

    #[test]
    fn a_where_scoped_index_answers_only_a_query_that_repeats_its_predicate() {
        let dir = tempdir().unwrap();
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "role": { "type": "string" },
                "status": { "type": "string" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[schema]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "role": "admin", "status": "active" }), false)
            .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u2", "role": "admin", "status": "inactive" }), false)
            .unwrap();

        db.declare_index(
            "users",
            json!({
                "name": "users_role_active",
                "fields": ["role"],
                "kind": "exact",
                "where": [{ "field": "status", "op": "=", "value": "active" }],
            }),
        )
        .unwrap();

        let eq = crate::query::predicates::ComparisonOp::Eq;
        let omitted = QueryBuilder::new().r#where("role", eq, json!("admin")).spec();
        let found = db.table("users").unwrap().query(&omitted).unwrap();
        assert_eq!(
            found.keys,
            vec!["u1".to_string(), "u2".to_string()],
            "the index holds u1 alone, but the query never asked for active records only"
        );
        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Scan);

        let repeated = QueryBuilder::new()
            .r#where("role", eq, json!("admin"))
            .r#where("status", eq, json!("active"))
            .spec();
        let narrowed = db.table("users").unwrap().query(&repeated).unwrap();
        assert_eq!(narrowed.keys, vec!["u1".to_string()]);
        assert_eq!(narrowed.plan.strategy, crate::query::planner::QueryStrategy::Exact);
        assert_eq!(narrowed.plan.index, Some("users_role_active".to_string()));
        assert_eq!(narrowed.plan.candidates, 1);
        db.close().unwrap();
    }

    #[test]
    fn a_declared_composite_index_still_serves_queries_after_closing_and_reopening_the_database() {
        let dir = tempdir().unwrap();
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "orders",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "userId": { "type": "string" },
                "status": { "type": "string" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[schema]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
            .unwrap();
        db.table_mut("orders")
            .unwrap()
            .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
            .unwrap();
        db.declare_index(
            "orders",
            json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
        )
        .unwrap();
        db.close().unwrap();

        let reopened = Database::open(dir.path()).unwrap();
        let spec = QueryBuilder::new()
            .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
            .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("closed"))
            .spec();
        let found = reopened.table("orders").unwrap().query(&spec).unwrap();
        assert_eq!(found.keys, vec!["o2".to_string()]);
        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
        assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
        reopened.close().unwrap();
    }

    fn add_index_to_catalog_on_disk(dir: &Path, table: &str, index: IndexEntry) {
        let mut catalog = read_catalog(&dir.join(MANIFEST_FILE)).unwrap().unwrap();
        let entry = catalog.tables.get_mut(table).unwrap();
        entry.indexes = Some(vec![index]);
        write_catalog_to(dir, &catalog);
    }

    fn sorted_index_entry(name: &str, fields: &[&str]) -> IndexEntry {
        IndexEntry {
            name: name.to_string(),
            fields: fields.iter().map(|field| field.to_string()).collect(),
            kind: CatalogIndexKind::Sorted,
            unique: false,
            where_: None,
            expire_after_seconds: None,
        }
    }

    /// A snapshot entry carries the projection the registry held when it was written, so a snapshot
    /// taken before an index was declared knows none of that index's fields. Restoring from it
    /// would leave the new index empty and answer every query against it with silence — the
    /// fingerprint has to count the index set, not the schema alone, so this pairing is stale.
    #[test]
    fn a_snapshot_written_before_an_index_was_declared_is_rebuilt_rather_than_restored_into_it() {
        let dir = tempdir().unwrap();
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "orders",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "userId": { "type": "string" },
                "total": { "type": "int" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[schema]));
        let mut db = Database::open(dir.path()).unwrap();
        let orders = db.table_mut("orders").unwrap();
        orders
            .put(&json!({ "id": "a", "userId": "u1", "total": 10 }), false)
            .unwrap();
        orders.put(&json!({ "id": "b", "userId": "u1" }), false).unwrap();
        orders
            .put(&json!({ "id": "c", "userId": "u0", "total": 5 }), false)
            .unwrap();
        db.close().unwrap();
        assert!(
            index_path(dir.path(), "orders").exists(),
            "the fixture needs a snapshot to restore from"
        );

        add_index_to_catalog_on_disk(
            dir.path(),
            "orders",
            sorted_index_entry("orders_user_total", &["userId", "total"]),
        );

        let reopened = Database::open(dir.path()).unwrap();
        assert_eq!(reopened.restore_outcome("orders"), Some(RestoreOutcome::Rebuilt));

        let spec = QueryBuilder::new()
            .r#where("userId", crate::query::predicates::ComparisonOp::Gte, json!("u1"))
            .spec();
        let found = reopened.table("orders").unwrap().query(&spec).unwrap();
        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Sorted);
        assert_eq!(found.plan.index, Some("orders_user_total".to_string()));
        let mut keys = found.keys.clone();
        keys.sort();
        assert_eq!(
            keys,
            vec!["a".to_string(), "b".to_string()],
            "b has no total, but it still has a userId"
        );
        reopened.close().unwrap();
    }

    /// The registry spans every partition of a table, so a composite unique index does too — but
    /// only if it was filled from the log rather than from a snapshot that predates it.
    #[test]
    fn a_composite_unique_index_declared_after_a_snapshot_still_holds_across_partitions() {
        let dir = tempdir().unwrap();
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "tickets",
            "primaryKey": "id",
            "partitionBy": "kind",
            "fields": {
                "id": { "type": "string" },
                "kind": { "type": "string", "required": true },
                "userId": { "type": "string", "required": true },
                "orderId": { "type": "int" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[schema]));
        let mut db = Database::open(dir.path()).unwrap();
        let tickets = db.table_mut("tickets").unwrap();
        tickets
            .put(
                &json!({ "id": "t1", "kind": "sale", "userId": "u1", "orderId": 1 }),
                false,
            )
            .unwrap();
        tickets
            .put(
                &json!({ "id": "t2", "kind": "sale", "userId": "u1", "orderId": 2 }),
                false,
            )
            .unwrap();
        tickets
            .put(
                &json!({ "id": "t3", "kind": "refund", "userId": "u1", "orderId": 3 }),
                false,
            )
            .unwrap();
        db.close().unwrap();

        add_index_to_catalog_on_disk(
            dir.path(),
            "tickets",
            IndexEntry {
                name: "tickets_user_order".to_string(),
                fields: vec!["userId".to_string(), "orderId".to_string()],
                kind: CatalogIndexKind::Exact,
                unique: true,
                where_: None,
                expire_after_seconds: None,
            },
        );

        let mut reopened = Database::open(dir.path()).unwrap();
        let tickets = reopened.table_mut("tickets").unwrap();
        // NOTE: the clash sits in the "sale" partition, the write lands in "refund" — one key, two files.
        let clash = tickets.put(
            &json!({ "id": "t9", "kind": "refund", "userId": "u1", "orderId": 2 }),
            false,
        );
        match clash {
            Err(TableError::Unique(violation)) => {
                assert_eq!(violation.field, "userId, orderId");
                assert_eq!(violation.holder, "t2");
            }
            other => panic!("a repeated composite key must be refused, got {other:?}"),
        }
        tickets
            .put(
                &json!({ "id": "t10", "kind": "refund", "userId": "u1", "orderId": 9 }),
                false,
            )
            .unwrap();
        reopened.close().unwrap();
    }

    #[test]
    fn close_snapshots_dirty_tables_so_a_reopen_restores_instead_of_rebuilding() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();

        // NOTE: well under DEFAULT_SNAPSHOT_EVERY — only close(), not the threshold, should snapshot.
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
            .unwrap();
        assert!(!index_path(dir.path(), "users").exists());

        db.close().unwrap();
        assert!(index_path(dir.path(), "users").exists());

        let reopened = Database::open(dir.path()).unwrap();
        assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
        assert_eq!(
            reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        assert_eq!(
            reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
            Some(&json!("Piotr"))
        );
        reopened.close().unwrap();
    }

    #[test]
    fn close_commits_a_staged_write_so_a_reopen_still_finds_it() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.table_mut("users")
            .unwrap()
            .stage_put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        assert_eq!(
            db.table("users").unwrap().len(),
            0,
            "nothing is committed before the close"
        );

        db.close().unwrap();

        let reopened = Database::open(dir.path()).unwrap();
        assert_eq!(
            reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
            Some(&json!("Ania"))
        );
        reopened.close().unwrap();
    }

    #[test]
    fn a_low_snapshot_every_writes_the_index_without_waiting_for_close() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open_with(
            dir.path(),
            OpenOptions {
                snapshot_every: 2,
                ..Default::default()
            },
        )
        .unwrap();

        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        assert!(!index_path(dir.path(), "users").exists());
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
            .unwrap();
        assert!(index_path(dir.path(), "users").exists());

        db.close().unwrap();
    }

    #[test]
    fn a_snapshot_taken_mid_stream_does_not_disturb_writes_that_follow_it() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open_with(
            dir.path(),
            OpenOptions {
                snapshot_every: 2,
                ..Default::default()
            },
        )
        .unwrap();

        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u1", "name": "Ania" }), false)
            .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
            .unwrap();
        db.table_mut("users")
            .unwrap()
            .put(&json!({ "id": "u3", "name": "Kasia" }), false)
            .unwrap();
        assert_eq!(
            db.table("users").unwrap().get("u3").unwrap().unwrap().get("name"),
            Some(&json!("Kasia"))
        );
        db.close().unwrap();
    }

    fn users_insert_audits_into_orders() -> Value {
        json!({
            "name": "users_insert_audit",
            "table": "users",
            "event": "insert",
            "effect": {
                "table": "orders",
                "fields": [
                    { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
                    { "target": "userId", "source": { "kind": "new", "field": "id" } },
                ],
            },
        })
    }

    #[test]
    fn a_table_declaring_an_image_dir_option_writes_bytes_under_it_instead_of_the_default() {
        let dir = tempdir().unwrap();
        let docs_schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "cover": { "type": "image" },
            },
        }))
        .unwrap();
        let mut catalog = catalog_of(&[docs_schema]);
        catalog.tables.get_mut("docs").unwrap().options = Some(crate::catalog::TableOptions {
            image_dir: Some("shared-images".to_string()),
            ..Default::default()
        });
        write_catalog_to(dir.path(), &catalog);

        let mut db = Database::open(dir.path()).unwrap();
        let record = db
            .table_mut("docs")
            .unwrap()
            .put(
                &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
                false,
            )
            .unwrap();
        let physical_path = record
            .get("cover")
            .unwrap()
            .get("physicalPath")
            .unwrap()
            .as_str()
            .unwrap();

        assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
        assert!(dir.path().join("shared-images").join(physical_path).exists());
        db.close().unwrap();
    }

    #[test]
    fn a_table_declaring_a_lazy_link_parsing_option_opens_with_it_instead_of_the_eager_default() {
        let dir = tempdir().unwrap();
        let pages_schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "pages",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
            },
        }))
        .unwrap();
        let mut catalog = catalog_of(&[pages_schema]);
        catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
            ..Default::default()
        });
        write_catalog_to(dir.path(), &catalog);

        let mut db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("pages").unwrap().link_parsing(),
            crate::catalog::LinkParsing::Lazy
        );

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

    #[test]
    fn a_table_declaring_no_link_parsing_option_defaults_to_eager() {
        let dir = tempdir().unwrap();
        let pages_schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "pages",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
            },
        }))
        .unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[pages_schema]));

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(
            db.table("pages").unwrap().link_parsing(),
            crate::catalog::LinkParsing::Eager
        );
    }

    #[test]
    fn declare_trigger_writes_the_catalog_and_refreshes_both_tables_live_trigger_set() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
        let mut db = Database::open(dir.path()).unwrap();

        db.declare_trigger(users_insert_audits_into_orders()).unwrap();

        assert_eq!(db.catalog.triggers.as_ref().unwrap().len(), 1);
        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
        assert!(db.table("orders").unwrap().triggers().is_empty());

        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
        assert_eq!(reread.triggers.as_ref().unwrap().len(), 1);
        assert_eq!(reread.triggers.as_ref().unwrap()[0].name, "users_insert_audit");

        db.close().unwrap();
    }

    #[test]
    fn a_declared_trigger_fires_its_effect_immediately_on_the_next_unit_of_work() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.declare_trigger(users_insert_audits_into_orders()).unwrap();

        let mut unit = db.unit_of_work();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        db.commit(&mut unit).unwrap();

        assert_eq!(
            db.table("orders")
                .unwrap()
                .get("o-audit")
                .unwrap()
                .unwrap()
                .get("userId"),
            Some(&json!("u1"))
        );
        db.close().unwrap();
    }

    #[test]
    fn a_declared_trigger_still_fires_after_closing_and_reopening_the_database() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
        db.close().unwrap();

        let mut reopened = Database::open(dir.path()).unwrap();
        let mut unit = reopened.unit_of_work();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        reopened.commit(&mut unit).unwrap();

        assert_eq!(
            reopened
                .table("orders")
                .unwrap()
                .get("o-audit")
                .unwrap()
                .unwrap()
                .get("userId"),
            Some(&json!("u1"))
        );
        reopened.close().unwrap();
    }

    #[test]
    fn a_declared_trigger_still_fires_after_declare_index_reopened_its_source_table() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.declare_trigger(users_insert_audits_into_orders()).unwrap();

        db.declare_index(
            "users",
            json!({ "name": "users_by_name", "fields": ["name"], "kind": "exact" }),
        )
        .unwrap();

        assert_eq!(db.table("users").unwrap().triggers().len(), 1);

        let mut unit = db.unit_of_work();
        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
        db.commit(&mut unit).unwrap();

        assert_eq!(
            db.table("orders")
                .unwrap()
                .get("o-audit")
                .unwrap()
                .unwrap()
                .get("userId"),
            Some(&json!("u1"))
        );
        db.close().unwrap();
    }

    #[test]
    fn a_declared_trigger_still_fires_after_drop_field_reopened_its_source_table() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.declare_trigger(users_insert_audits_into_orders()).unwrap();

        db.drop_field("users", "name", false).unwrap();

        assert_eq!(db.table("users").unwrap().triggers().len(), 1);

        let mut unit = db.unit_of_work();
        unit.put("users", json!({ "id": "u1" }));
        db.commit(&mut unit).unwrap();

        assert_eq!(
            db.table("orders")
                .unwrap()
                .get("o-audit")
                .unwrap()
                .unwrap()
                .get("userId"),
            Some(&json!("u1"))
        );
        db.close().unwrap();
    }

    #[test]
    fn a_declared_trigger_still_fires_after_create_table_migrated_its_source_table() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
        let mut db = Database::open(dir.path()).unwrap();
        db.declare_trigger(users_insert_audits_into_orders()).unwrap();

        db.create_table(
            "users",
            json!({
                "schemaVersion": 1,
                "table": "users",
                "revision": 2,
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "name": { "type": "string", "required": true },
                    "city": { "type": "string" },
                },
            }),
        )
        .unwrap();

        assert_eq!(db.table("users").unwrap().triggers().len(), 1);

        let mut unit = db.unit_of_work();
        unit.put("users", json!({ "id": "u1", "name": "Ania", "city": "Kraków" }));
        db.commit(&mut unit).unwrap();

        assert_eq!(
            db.table("orders")
                .unwrap()
                .get("o-audit")
                .unwrap()
                .unwrap()
                .get("userId"),
            Some(&json!("u1"))
        );
        db.close().unwrap();
    }

    #[test]
    fn declare_trigger_refuses_an_unknown_effect_table_and_changes_nothing_on_disk() {
        let dir = tempdir().unwrap();
        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
        let mut db = Database::open(dir.path()).unwrap();

        let spec = json!({
            "name": "users_insert_audit",
            "table": "users",
            "event": "insert",
            "effect": {
                "table": "ghost",
                "fields": [
                    { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
                ],
            },
        });
        let error = db.declare_trigger(spec);
        assert!(error.is_err());
        assert!(db.catalog.triggers.is_none());

        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
        assert!(reread.triggers.is_none());
        db.close().unwrap();
    }

    mod automatic_compaction {
        use super::*;
        use crate::catalog::TableOptions;

        fn enabled(dead_bytes_ratio: f64, min_dead_bytes: u64) -> CatalogCompaction {
            CatalogCompaction::Enabled {
                dead_bytes_ratio,
                min_dead_bytes,
            }
        }

        fn opened(dir: &Path, compaction: Option<CatalogCompaction>) -> Database {
            Database::open_with(
                dir,
                OpenOptions {
                    compaction,
                    ..Default::default()
                },
            )
            .unwrap()
        }

        fn lines_of(path: &Path) -> Vec<String> {
            fs::read_to_string(path).unwrap().lines().map(str::to_string).collect()
        }

        /// One put and `overwrites` patches on the same key, so every patch but the last turns a
        /// whole line into garbage without changing how many records the table holds.
        fn churn(db: &mut Database, table: &str, key: &str, overwrites: usize) {
            let users = db.table_mut(table).unwrap();
            users.put(&json!({ "id": key, "name": "Ania" }), false).unwrap();
            for round in 0..overwrites {
                users
                    .patch(key, &json!({ "name": format!("Ania {round}") }), false)
                    .unwrap();
            }
        }

        #[test]
        fn garbage_under_the_absolute_floor_leaves_the_log_alone_however_large_its_share() {
            let dir = tempdir().unwrap();
            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
            let mut db = opened(dir.path(), Some(enabled(0.1, 1_000_000)));

            churn(&mut db, "users", "u1", 12);

            let stats = db.table("users").unwrap().stats();
            assert_eq!(stats.gen, 0, "minDeadBytes was never reached");
            assert!(stats.dead_bytes > 0, "the overwrites did make garbage");
            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
            db.close().unwrap();
        }

        #[test]
        fn garbage_below_the_ratio_leaves_the_log_alone_however_many_bytes_it_is() {
            let dir = tempdir().unwrap();
            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
            let mut db = opened(dir.path(), Some(enabled(0.99, 1)));

            churn(&mut db, "users", "u1", 4);

            let stats = db.table("users").unwrap().stats();
            assert_eq!(stats.gen, 0, "four of five lines is garbage, but not 99% of the bytes");
            assert!(stats.dead_bytes > 1);
            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 5);
            db.close().unwrap();
        }

        #[test]
        fn crossing_both_thresholds_rewrites_the_log_and_moves_the_generation_on() {
            let dir = tempdir().unwrap();
            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));

            churn(&mut db, "users", "u1", 12);

            let stats = db.table("users").unwrap().stats();
            assert!(stats.gen > 0, "the write that crossed the thresholds compacted");
            assert_eq!(stats.dead_bytes, 0);
            assert_eq!(stats.records, 1);
            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
            assert_eq!(
                db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
                Some(&json!("Ania 11"))
            );
            db.close().unwrap();
        }

        #[test]
        fn a_log_the_thresholds_rewrote_reopens_at_the_generation_and_the_records_it_left() {
            let dir = tempdir().unwrap();
            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
            churn(&mut db, "users", "u1", 12);
            db.table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u2", "name": "Piotr" }), false)
                .unwrap();
            let before = db.table("users").unwrap().stats();
            db.close().unwrap();

            let reopened = opened(dir.path(), Some(enabled(0.5, 200)));

            let after = reopened.table("users").unwrap().stats();
            assert_eq!(
                after.gen, before.gen,
                "the generation is carried by the snapshot, not reset"
            );
            assert_eq!(after.records, 2);
            assert_eq!(after.dead_bytes, 0);
            let mut keys: Vec<&String> = reopened.table("users").unwrap().keys().collect();
            keys.sort();
            assert_eq!(keys, vec!["u1", "u2"]);
            assert_eq!(
                reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
                Some(&json!("Ania 11"))
            );
            reopened.close().unwrap();
        }

        #[test]
        fn a_partitioned_table_is_measured_and_rewritten_one_partition_at_a_time() {
            let dir = tempdir().unwrap();
            let schema = parse_schema(&json!({
                "schemaVersion": 1,
                "table": "tickets",
                "primaryKey": "id",
                "partitionBy": "kind",
                "fields": {
                    "id": { "type": "string" },
                    "kind": { "type": "string", "required": true },
                    "note": { "type": "string" },
                },
            }))
            .unwrap();
            write_catalog_to(dir.path(), &catalog_of(&[schema]));
            // NOTE: a floor this far above what 12 rewrites of one line make can only be crossed once.
            let mut db = opened(dir.path(), Some(enabled(0.5, 700)));

            let tickets = db.table_mut("tickets").unwrap();
            tickets
                .put(&json!({ "id": "r1", "kind": "refund", "note": "n" }), false)
                .unwrap();
            let refund = dir.path().join("tickets").join("refund.jsonl");
            let quiet_size = fs::metadata(&refund).unwrap().len();

            tickets
                .put(&json!({ "id": "s1", "kind": "sale", "note": "n" }), false)
                .unwrap();
            for round in 0..12 {
                tickets
                    .patch("s1", &json!({ "note": format!("n{round}") }), false)
                    .unwrap();
            }

            let sale = dir.path().join("tickets").join("sale.jsonl");
            assert!(
                lines_of(&sale).len() < 13,
                "the partition over its thresholds was rewritten, so not all 13 lines are still there"
            );
            assert_eq!(
                fs::metadata(&refund).unwrap().len(),
                quiet_size,
                "the partition holding no garbage was left byte for byte"
            );
            assert_eq!(lines_of(&refund).len(), 1);
            let stats = db.table("tickets").unwrap().stats();
            assert_eq!(
                stats.gen, 1,
                "one pass moves the generation once, not once per partition"
            );
            assert_eq!(stats.records, 2);
            db.close().unwrap();
        }

        #[test]
        fn a_catalog_declaring_compaction_false_keeps_the_garbage_an_open_call_named_nothing_about() {
            let dir = tempdir().unwrap();
            let mut catalog = catalog_of(&[users_schema()]);
            catalog.defaults = Some(TableOptions {
                compaction: Some(CatalogCompaction::Disabled),
                ..Default::default()
            });
            write_catalog_to(dir.path(), &catalog);
            let mut db = opened(dir.path(), None);

            churn(&mut db, "users", "u1", 12);

            assert_eq!(db.table("users").unwrap().stats().gen, 0);
            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
            db.close().unwrap();
        }

        #[test]
        fn a_tables_own_entry_beats_the_catalog_defaults_and_an_open_call_beats_them_both() {
            let dir = tempdir().unwrap();
            let mut catalog = catalog_of(&[users_schema()]);
            catalog.defaults = Some(TableOptions {
                compaction: Some(CatalogCompaction::Disabled),
                ..Default::default()
            });
            catalog.tables.get_mut("users").unwrap().options = Some(TableOptions {
                compaction: Some(enabled(0.5, 200)),
                ..Default::default()
            });
            write_catalog_to(dir.path(), &catalog);

            let db = opened(dir.path(), None);
            assert_eq!(db.table("users").unwrap().compaction(), enabled(0.5, 200));
            db.close().unwrap();

            let db = opened(dir.path(), Some(CatalogCompaction::Disabled));
            assert_eq!(db.table("users").unwrap().compaction(), CatalogCompaction::Disabled);
            db.close().unwrap();
        }

        #[test]
        fn a_table_created_under_a_handle_is_armed_on_the_same_terms_as_one_opened_with_it() {
            let dir = tempdir().unwrap();
            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
            db.create_table(
                "users",
                json!({
                    "schemaVersion": 1,
                    "table": "users",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
                }),
            )
            .unwrap();

            churn(&mut db, "users", "u1", 12);

            assert!(db.table("users").unwrap().stats().gen > 0);
            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
            db.close().unwrap();
        }

        #[test]
        fn an_open_call_naming_nothing_falls_back_to_the_engine_default_thresholds() {
            let dir = tempdir().unwrap();
            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
            let db = opened(dir.path(), None);

            assert_eq!(db.table("users").unwrap().compaction(), DEFAULT_COMPACTION);
            db.close().unwrap();
        }
    }

    /// `OpenOptions::cache` reaching the tables — the read cache itself is pinned in
    /// `table::cache` and `table::handle`.
    mod read_cache {
        use super::*;

        fn schema() -> Value {
            json!({
                "schemaVersion": 1,
                "table": "users",
                "primaryKey": "id",
                "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
            })
        }

        fn opened(dir: &Path, cache: CachePolicy) -> Database {
            Database::open_with(
                dir,
                OpenOptions {
                    cache,
                    ..Default::default()
                },
            )
            .unwrap()
        }

        #[test]
        fn an_open_that_names_no_cache_leaves_every_table_caching_nothing() {
            let dir = tempdir().unwrap();
            let mut db = Database::open(dir.path()).unwrap();
            db.create_table("users", schema()).unwrap();
            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
            db.close().unwrap();

            let db = Database::open(dir.path()).unwrap();
            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
            db.close().unwrap();
        }

        #[test]
        fn the_policy_an_open_call_names_arms_the_tables_it_opens_and_the_ones_it_creates() {
            let dir = tempdir().unwrap();
            let mut db = opened(dir.path(), CachePolicy::with_entries(64));
            db.create_table("users", schema()).unwrap();
            assert_eq!(
                db.table("users").unwrap().cache_policy(),
                CachePolicy::with_entries(64),
                "a table created under the handle is armed on the same terms"
            );
            db.close().unwrap();

            let db = opened(dir.path(), CachePolicy::with_entries(64));
            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::with_entries(64));
            db.close().unwrap();
        }

        #[test]
        fn a_cached_table_read_through_the_database_answers_the_same_document_twice() {
            let dir = tempdir().unwrap();
            let mut db = opened(dir.path(), CachePolicy::with_entries(8));
            db.create_table("users", schema()).unwrap();
            let users = db.table_mut("users").unwrap();
            users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();

            let first = users.get("u1").unwrap();
            let second = users.get("u1").unwrap();

            assert_eq!(first, second);
            assert_eq!(first.unwrap().get("name"), Some(&json!("Ania")));
            db.close().unwrap();
        }
    }

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

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

        fn secrets_schema_json() -> Value {
            json!({
                "schemaVersion": 1,
                "table": "secrets",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "token": { "type": "string", "cypher": true },
                },
            })
        }

        fn secrets_schema_json_with_note() -> Value {
            json!({
                "schemaVersion": 1,
                "table": "secrets",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "token": { "type": "string", "cypher": true },
                    "note": { "type": "string" },
                },
            })
        }

        fn open_with_keys(dir: &Path, keys: &[(&str, u8)], key_id: Option<&str>) -> Database {
            Database::open_with(
                dir,
                OpenOptions {
                    cypher: Some(CipherOptions {
                        keys: keys.iter().map(|(id, byte)| (id.to_string(), key(*byte))).collect(),
                        key_id: key_id.map(str::to_string),
                        on_missing_key: None,
                    }),
                    ..OpenOptions::default()
                },
            )
            .unwrap()
        }

        #[test]
        fn a_cypher_field_round_trips_through_open_created_table_and_reopen() {
            let dir = tempdir().unwrap();
            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
            db.create_table("secrets", secrets_schema_json()).unwrap();
            db.table_mut("secrets")
                .unwrap()
                .put(&json!({ "id": "s1", "token": "hunter2" }), false)
                .unwrap();
            db.close().unwrap();

            let reopened = open_with_keys(dir.path(), &[("k1", 1)], None);
            assert_eq!(
                reopened
                    .table("secrets")
                    .unwrap()
                    .get("s1")
                    .unwrap()
                    .unwrap()
                    .get("token"),
                Some(&json!("hunter2"))
            );
            reopened.close().unwrap();
        }

        #[test]
        fn reads_with_an_old_key_and_writes_with_the_new_one_after_rotation() {
            let dir = tempdir().unwrap();
            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
            db.create_table("secrets", secrets_schema_json()).unwrap();
            db.table_mut("secrets")
                .unwrap()
                .put(&json!({ "id": "s1", "token": "written-with-k1" }), false)
                .unwrap();
            db.close().unwrap();

            let mut rotated = open_with_keys(dir.path(), &[("k1", 1), ("k2", 2)], Some("k2"));
            assert_eq!(
                rotated
                    .table("secrets")
                    .unwrap()
                    .get("s1")
                    .unwrap()
                    .unwrap()
                    .get("token"),
                Some(&json!("written-with-k1"))
            );
            rotated
                .table_mut("secrets")
                .unwrap()
                .put(&json!({ "id": "s2", "token": "written-with-k2" }), false)
                .unwrap();
            rotated.close().unwrap();

            let log = std::fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
            assert!(log.contains("$cypher$v1$k1$"));
            assert!(log.contains("$cypher$v1$k2$"));
        }

        #[test]
        fn keeps_the_keyring_across_drop_field_reopening_the_table() {
            let dir = tempdir().unwrap();
            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
            db.create_table("secrets", secrets_schema_json_with_note()).unwrap();
            db.table_mut("secrets")
                .unwrap()
                .put(&json!({ "id": "s1", "token": "hunter2", "note": "x" }), false)
                .unwrap();

            db.drop_field("secrets", "note", true).unwrap();
            assert_eq!(
                db.table("secrets").unwrap().get("s1").unwrap().unwrap().get("token"),
                Some(&json!("hunter2"))
            );
            db.close().unwrap();
        }

        #[test]
        fn open_refuses_a_key_that_is_not_32_bytes() {
            let dir = tempdir().unwrap();
            let error = Database::open_with(
                dir.path(),
                OpenOptions {
                    cypher: Some(CipherOptions {
                        keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
                        key_id: None,
                        on_missing_key: None,
                    }),
                    ..OpenOptions::default()
                },
            )
            .map(|_| ())
            .unwrap_err();
            assert!(matches!(error, DatabaseError::Cipher(_)));
        }

        #[test]
        fn missing_key_policy_raw_hands_back_the_envelope_through_the_database() {
            let dir = tempdir().unwrap();
            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
            db.create_table("secrets", secrets_schema_json()).unwrap();
            db.table_mut("secrets")
                .unwrap()
                .put(&json!({ "id": "s1", "token": "hunter2" }), false)
                .unwrap();
            db.close().unwrap();

            let raw = Database::open_with(
                dir.path(),
                OpenOptions {
                    cypher: Some(CipherOptions {
                        keys: HashMap::from([("other".to_string(), key(2))]),
                        key_id: None,
                        on_missing_key: Some(MissingKeyPolicy::Raw),
                    }),
                    ..OpenOptions::default()
                },
            )
            .unwrap();
            let found = raw.table("secrets").unwrap().get("s1").unwrap().unwrap();
            assert!(found.get("token").unwrap().as_str().unwrap().contains("$cypher$v1$k1$"));
            raw.close().unwrap();
        }
    }

    mod partitions {
        use super::*;

        fn notes_schema_json(partitioned: bool) -> Value {
            let mut spec = json!({
                "schemaVersion": 1,
                "table": "notes",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "kind": { "type": "string", "required": true },
                    "title": { "type": "string", "required": true },
                    "body": { "type": "string" },
                },
            });
            if partitioned {
                spec.as_object_mut()
                    .unwrap()
                    .insert("partitionBy".into(), json!("kind"));
            }
            spec
        }

        fn lines_in(dir: &Path, part: &str) -> usize {
            fs::read_to_string(dir.join("notes").join(format!("{part}.jsonl")))
                .unwrap()
                .trim()
                .lines()
                .filter(|line| !line.is_empty())
                .count()
        }

        #[test]
        fn a_partitioned_table_survives_the_catalog_round_trip() {
            let dir = tempdir().unwrap();
            {
                let mut db = Database::open(dir.path()).unwrap();
                db.create_table("notes", notes_schema_json(true)).unwrap();
                let notes = db.table_mut("notes").unwrap();
                notes
                    .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
                    .unwrap();
                notes
                    .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
                    .unwrap();
                db.close().unwrap();
            }

            let db = Database::open(dir.path()).unwrap();
            let notes = db.table("notes").unwrap();
            assert_eq!(notes.schema().partition_by.as_deref(), Some("kind"));
            assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
            assert_eq!(lines_in(dir.path(), "note"), 1);
            assert_eq!(lines_in(dir.path(), "adr"), 1);
            assert!(!dir.path().join("notes.jsonl").exists());
            db.close().unwrap();
        }

        #[test]
        fn declaring_partitions_on_an_existing_table_moves_its_records_into_the_new_layout() {
            let dir = tempdir().unwrap();
            {
                let mut db = Database::open(dir.path()).unwrap();
                db.create_table("notes", notes_schema_json(false)).unwrap();
                let notes = db.table_mut("notes").unwrap();
                notes
                    .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
                    .unwrap();
                notes
                    .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
                    .unwrap();
                notes.delete("n1", false).unwrap();
                db.close().unwrap();
            }

            let mut db = Database::open(dir.path()).unwrap();
            let mut partitioned = notes_schema_json(true);
            partitioned.as_object_mut().unwrap().insert("revision".into(), json!(2));
            db.create_table("notes", partitioned).unwrap();

            let notes = db.table("notes").unwrap();
            assert_eq!(notes.get("n1").unwrap(), None);
            assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
            assert!(!dir.path().join("notes.jsonl").exists());
            assert_eq!(lines_in(dir.path(), "adr"), 1);
            db.close().unwrap();
        }

        #[test]
        fn dropping_the_field_the_table_partitions_by_is_refused() {
            let dir = tempdir().unwrap();
            let mut db = Database::open(dir.path()).unwrap();
            db.create_table("notes", notes_schema_json(true)).unwrap();
            db.table_mut("notes")
                .unwrap()
                .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
                .unwrap();

            let error = db.drop_field("notes", "kind", true).unwrap_err();
            assert!(error.to_string().contains("partitioned by"));
            assert_eq!(
                db.table("notes").unwrap().get("n1").unwrap().unwrap().get("kind"),
                Some(&json!("note"))
            );
            db.close().unwrap();
        }

        #[test]
        fn dropping_an_unrelated_field_keeps_every_partition_readable() {
            let dir = tempdir().unwrap();
            let mut db = Database::open(dir.path()).unwrap();
            db.create_table("notes", notes_schema_json(true)).unwrap();
            {
                let notes = db.table_mut("notes").unwrap();
                notes
                    .put(
                        &json!({ "id": "n1", "kind": "note", "title": "first", "body": "gone soon" }),
                        false,
                    )
                    .unwrap();
                notes
                    .put(&json!({ "id": "a1", "kind": "adr", "title": "kept" }), false)
                    .unwrap();
            }

            db.drop_field("notes", "body", true).unwrap();

            let notes = db.table("notes").unwrap();
            assert!(!notes.schema().fields.contains_key("body"));
            assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("first")));
            assert_eq!(notes.get("a1").unwrap().unwrap().get("kind"), Some(&json!("adr")));
            assert_eq!(lines_in(dir.path(), "note"), 1);
            assert_eq!(lines_in(dir.path(), "adr"), 1);
            db.close().unwrap();
        }

        #[test]
        fn a_unit_of_work_lands_in_two_partitions_of_one_table() {
            let dir = tempdir().unwrap();
            let mut db = Database::open(dir.path()).unwrap();
            db.create_table("notes", notes_schema_json(true)).unwrap();

            let mut unit = db.unit_of_work();
            unit.put("notes", json!({ "id": "n1", "kind": "note", "title": "from unit" }));
            unit.put("notes", json!({ "id": "a1", "kind": "adr", "title": "from unit too" }));
            let result = db.commit(&mut unit).unwrap();
            assert_eq!(result.operations, 2);

            let notes = db.table("notes").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!(lines_in(dir.path(), "note"), 1);
            assert_eq!(lines_in(dir.path(), "adr"), 1);
            db.close().unwrap();
        }
    }
    mod following_the_log {
        use super::*;

        fn readonly(follow: FollowOption) -> OpenOptions {
            OpenOptions {
                readonly: true,
                follow,
                ..Default::default()
            }
        }

        /// A writer and a reader on the same directory, as two separate handles — the shape of the
        /// TS `multiprocess.test.ts`, which runs the writer in a child process for the lock's sake.
        /// One process is enough here: the lock is only taken by the writable handle.
        fn writer_and_reader(dir: &Path, follow: FollowOption) -> (Database, Database) {
            let mut writer = Database::open(dir).unwrap();
            writer.create_table("users", users_schema_spec()).unwrap();
            writer
                .table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();
            let reader = Database::open_with(dir, readonly(follow)).unwrap();
            (writer, reader)
        }

        fn users_schema_spec() -> Value {
            json!({
                "schemaVersion": 1,
                "table": "users",
                "primaryKey": "id",
                "fields": {
                    "id": { "type": "string" },
                    "name": { "type": "string", "required": true },
                },
            })
        }

        fn put(writer: &mut Database, key: &str, name: &str) {
            writer
                .table_mut("users")
                .unwrap()
                .put(&json!({ "id": key, "name": name }), false)
                .unwrap();
        }

        #[test]
        fn a_readonly_open_follows_by_default_and_a_writable_one_does_not() {
            let dir = tempdir().unwrap();
            let mut writer = Database::open(dir.path()).unwrap();
            writer.create_table("users", users_schema_spec()).unwrap();

            assert_eq!(writer.follow(), None);
            assert_eq!(writer.table("users").unwrap().follow(), None);

            let reader = Database::open_with(dir.path(), readonly(FollowOption::Default)).unwrap();
            assert_eq!(
                reader.follow(),
                Some(FollowPolicy {
                    interval: DEFAULT_FOLLOW_INTERVAL
                })
            );
            assert_eq!(
                reader.table("users").unwrap().follow(),
                Some(FollowPolicy {
                    interval: DEFAULT_FOLLOW_INTERVAL
                })
            );
        }

        #[test]
        fn a_plain_get_on_a_following_reader_sees_what_the_writer_committed_after_it_opened() {
            let dir = tempdir().unwrap();
            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);

            for index in 2..=5 {
                put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
                let found = reader
                    .table_mut("users")
                    .unwrap()
                    .get_following(&format!("u{index}"))
                    .unwrap();
                assert_eq!(found.unwrap().get("name"), Some(&json!(format!("Osoba {index}"))));
            }
        }

        #[test]
        fn a_plain_query_on_a_following_reader_counts_what_the_writer_committed() {
            let dir = tempdir().unwrap();
            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);

            let spec = QueryBuilder::new().spec();
            assert_eq!(
                reader
                    .table_mut("users")
                    .unwrap()
                    .query_following(&spec)
                    .unwrap()
                    .matched,
                1
            );

            put(&mut writer, "u2", "Osoba 2");

            assert_eq!(
                reader
                    .table_mut("users")
                    .unwrap()
                    .query_following(&spec)
                    .unwrap()
                    .matched,
                2
            );
        }

        #[test]
        fn a_reader_with_follow_off_stands_still_until_refresh_is_called() {
            let dir = tempdir().unwrap();
            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Off);
            put(&mut writer, "u2", "Osoba 2");

            assert!(reader
                .table_mut("users")
                .unwrap()
                .get_following("u2")
                .unwrap()
                .is_none());
            assert!(reader.table_mut("users").unwrap().refresh().unwrap().changed());
            assert!(reader
                .table_mut("users")
                .unwrap()
                .get_following("u2")
                .unwrap()
                .is_some());
        }

        #[test]
        fn a_custom_interval_coalesces_the_freshness_check_and_lets_it_through_once_it_elapses() {
            let dir = tempdir().unwrap();
            let (mut writer, mut reader) =
                writer_and_reader(dir.path(), FollowOption::Every(Duration::from_millis(150)));

            // The first read is always due — nothing has been followed yet — and starts the clock.
            assert!(reader
                .table_mut("users")
                .unwrap()
                .get_following("u1")
                .unwrap()
                .is_some());
            put(&mut writer, "u2", "Osoba 2");
            assert!(reader
                .table_mut("users")
                .unwrap()
                .get_following("u2")
                .unwrap()
                .is_none());

            std::thread::sleep(Duration::from_millis(200));

            assert!(reader
                .table_mut("users")
                .unwrap()
                .get_following("u2")
                .unwrap()
                .is_some());
        }

        #[test]
        fn get_many_and_the_aggregates_follow_the_log_the_same_way_a_single_get_does() {
            let dir = tempdir().unwrap();
            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
            put(&mut writer, "u2", "Osoba 2");

            let keys = vec!["u1".to_string(), "u2".to_string()];
            let rows = reader.table_mut("users").unwrap().get_many_following(&keys).unwrap();
            assert!(rows.iter().all(Option::is_some), "{rows:?}");

            put(&mut writer, "u3", "Osoba 3");
            let spec = QueryBuilder::new().spec();
            let aggregate = crate::query::aggregate::AggregateSpec {
                count: true,
                ..Default::default()
            };
            let counted = reader
                .table_mut("users")
                .unwrap()
                .aggregate_following(&spec, &aggregate)
                .unwrap();
            assert_eq!(counted.count, 3);
        }

        #[test]
        fn a_following_reader_reloads_instead_of_reading_the_offsets_a_foreign_compaction_moved() {
            let dir = tempdir().unwrap();
            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
            for index in 2..=6 {
                put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
            }
            writer.table_mut("users").unwrap().delete("u5", false).unwrap();
            assert!(reader
                .table_mut("users")
                .unwrap()
                .get_following("u6")
                .unwrap()
                .is_some());

            writer.table_mut("users").unwrap().compact().unwrap();

            let users = reader.table_mut("users").unwrap();
            assert_eq!(
                users.get_following("u3").unwrap().unwrap().get("name"),
                Some(&json!("Osoba 3"))
            );
            assert!(users.get_following("u5").unwrap().is_none());
            assert_eq!(users.query_following(&QueryBuilder::new().spec()).unwrap().matched, 5);
        }
    }

    mod where_the_files_land {
        use super::*;
        use crate::database::LayoutKind;

        fn notes_schema_spec() -> Value {
            json!({
                "schemaVersion": 1,
                "table": "notes",
                "primaryKey": "id",
                "partitionBy": "kind",
                "fields": {
                    "id": { "type": "string" },
                    "kind": { "type": "string", "required": true, "index": "exact" },
                    "title": { "type": "string", "required": true },
                },
            })
        }

        fn nested(metadata_file: Option<&str>) -> OpenOptions {
            OpenOptions {
                layout: LayoutOptions {
                    kind: Some(LayoutKind::Nested),
                    metadata_file: metadata_file.map(str::to_string),
                    ..Default::default()
                },
                ..Default::default()
            }
        }

        #[test]
        fn the_default_open_writes_exactly_what_it_wrote_before_layouts_existed() {
            let dir = tempdir().unwrap();
            let mut db = Database::open(dir.path()).unwrap();
            db.create_table(
                "users",
                json!({
                    "schemaVersion": 1,
                    "table": "users",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
                }),
            )
            .unwrap();
            db.table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();
            db.close().unwrap();

            assert!(dir.path().join("hayson.json").is_file());
            assert!(dir.path().join("users.jsonl").is_file());
            assert!(!dir.path().join("tables").exists());
            assert!(!dir.path().join("runtime").exists());

            let db = Database::open(dir.path()).unwrap();
            assert_eq!(db.layout().kind(), LayoutKind::Flat);
            assert_eq!(db.layout().metadata_file(), "hayson.json");
            assert_eq!(db.layout().metadata_path(), dir.path().join("hayson.json"));
        }

        #[test]
        fn a_custom_metadata_file_holds_the_catalog_and_nothing_is_written_under_the_default_name() {
            let dir = tempdir().unwrap();
            let options = OpenOptions {
                layout: LayoutOptions {
                    metadata_file: Some("cerebro.json".into()),
                    ..Default::default()
                },
                ..Default::default()
            };

            let mut db = Database::open_with(dir.path(), options.clone()).unwrap();
            db.create_table(
                "users",
                json!({
                    "schemaVersion": 1,
                    "table": "users",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
                }),
            )
            .unwrap();
            db.table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();
            db.close().unwrap();

            assert!(dir.path().join("cerebro.json").is_file());
            assert!(!dir.path().join("hayson.json").exists());

            let db = Database::open_with(dir.path(), options).unwrap();
            assert_eq!(db.layout().metadata_file(), "cerebro.json");
            assert_eq!(db.table_names(), vec!["users"]);
            assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
        }

        #[test]
        fn the_deprecated_manifest_name_opens_what_the_new_one_wrote_and_the_other_way_round() {
            let dir = tempdir().unwrap();
            let legacy = OpenOptions {
                layout: LayoutOptions {
                    manifest: Some("cerebro.json".into()),
                    ..Default::default()
                },
                ..Default::default()
            };
            let mut db = Database::open_with(dir.path(), legacy).unwrap();
            db.create_table(
                "users",
                json!({
                    "schemaVersion": 1,
                    "table": "users",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" } },
                }),
            )
            .unwrap();
            db.close().unwrap();

            let current = OpenOptions {
                layout: LayoutOptions {
                    metadata_file: Some("cerebro.json".into()),
                    ..Default::default()
                },
                ..Default::default()
            };
            let db = Database::open_with(dir.path(), current).unwrap();
            assert_eq!(db.table_names(), vec!["users"]);
            assert_eq!(db.layout().metadata_path(), dir.path().join("cerebro.json"));
        }

        #[test]
        fn the_two_names_for_the_metadata_document_are_refused_when_they_disagree_touching_nothing() {
            let dir = tempdir().unwrap();
            let options = OpenOptions {
                layout: LayoutOptions {
                    metadata_file: Some("cerebro.json".into()),
                    manifest: Some("hayson.json".into()),
                    ..Default::default()
                },
                ..Default::default()
            };

            let Err(error) = Database::open_with(dir.path(), options) else {
                panic!("two disagreeing names for the metadata document were accepted")
            };

            assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
            let message = error.to_string();
            assert!(
                message.contains("metadataFile") && message.contains("manifest"),
                "{message}"
            );
            assert!(!dir.path().join("hayson.json").exists());
            assert!(!dir.path().join("cerebro.json").exists());
            assert!(!dir.path().join(crate::lock::LOCK_FILE).exists());
        }

        #[test]
        fn the_nested_layout_sorts_the_logs_the_snapshots_and_the_lock_into_their_subdirectories() {
            let dir = tempdir().unwrap();
            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
            db.create_table(
                "users",
                json!({
                    "schemaVersion": 1,
                    "table": "users",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
                }),
            )
            .unwrap();
            db.table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();

            assert!(dir.path().join("runtime").join(crate::lock::LOCK_FILE).is_file());
            db.close().unwrap();

            assert!(dir.path().join("hayson.json").is_file());
            assert!(dir.path().join("tables").join("users.jsonl").is_file());
            assert!(dir.path().join("index").join("users.idx").is_file());
            assert!(!dir.path().join("users.jsonl").exists());
        }

        #[test]
        fn a_partitioned_table_under_the_nested_layout_lands_under_tables_slash_its_own_name() {
            let dir = tempdir().unwrap();
            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
            db.create_table("notes", notes_schema_spec()).unwrap();
            db.table_mut("notes")
                .unwrap()
                .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
                .unwrap();
            db.close().unwrap();

            assert!(dir.path().join("tables").join("notes").join("note.jsonl").is_file());
        }

        #[test]
        fn a_nested_database_reopens_as_nested_without_being_told_and_is_refused_as_flat() {
            let dir = tempdir().unwrap();
            let mut db = Database::open_with(dir.path(), nested(Some("cerebro.json"))).unwrap();
            db.create_table(
                "users",
                json!({
                    "schemaVersion": 1,
                    "table": "users",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
                }),
            )
            .unwrap();
            db.table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();
            db.close().unwrap();

            let named_only = OpenOptions {
                layout: LayoutOptions {
                    metadata_file: Some("cerebro.json".into()),
                    ..Default::default()
                },
                ..Default::default()
            };
            let db = Database::open_with(dir.path(), named_only).unwrap();
            assert_eq!(db.layout().kind(), LayoutKind::Nested);
            assert_eq!(
                db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
                Some(&json!("Ania"))
            );
            db.close().unwrap();

            let as_flat = OpenOptions {
                layout: LayoutOptions {
                    kind: Some(LayoutKind::Flat),
                    metadata_file: Some("cerebro.json".into()),
                    ..Default::default()
                },
                ..Default::default()
            };
            let Err(error) = Database::open_with(dir.path(), as_flat) else {
                panic!("a nested database was opened as flat")
            };
            assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
            assert!(error.to_string().contains("nested"), "{error}");
        }

        #[test]
        fn a_unit_of_work_under_the_nested_layout_journals_into_runtime_and_reopens_whole() {
            let dir = tempdir().unwrap();
            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
            db.create_table(
                "users",
                json!({
                    "schemaVersion": 1,
                    "table": "users",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
                }),
            )
            .unwrap();

            let mut unit = db.unit_of_work();
            unit.put("users", json!({ "id": "u1", "name": "Ania" }));
            unit.put("users", json!({ "id": "u2", "name": "Bartek" }));
            db.commit(&mut unit).unwrap();
            db.close().unwrap();

            assert!(!dir.path().join("runtime").join(".hayson.txn").exists());
            let db = Database::open_with(dir.path(), nested(None)).unwrap();
            assert_eq!(db.table("users").unwrap().len(), 2);
        }
    }

    /// F8: a catalog naming a table whose log is not on disk. A writable handle makes the log it
    /// misses, so the two open modes owe the caller opposite answers here.
    mod a_declared_table_whose_log_is_not_on_disk {
        use super::*;

        fn notes_spec() -> Value {
            json!({
                "schemaVersion": 1,
                "table": "notes",
                "primaryKey": "id",
                "partitionBy": "kind",
                "fields": {
                    "id": { "type": "string" },
                    "kind": { "type": "string", "required": true },
                    "title": { "type": "string", "required": true },
                },
            })
        }

        fn readonly() -> OpenOptions {
            OpenOptions {
                readonly: true,
                ..Default::default()
            }
        }

        fn users_with_a_record(dir: &Path) {
            let mut db = Database::open(dir).unwrap();
            db.create_table(
                "users",
                json!({
                    "schemaVersion": 1,
                    "table": "users",
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
                }),
            )
            .unwrap();
            db.table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();
            db.close().unwrap();
        }

        fn notes_with_a_record(dir: &Path) {
            let mut db = Database::open(dir).unwrap();
            db.create_table("notes", notes_spec()).unwrap();
            db.table_mut("notes")
                .unwrap()
                .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
                .unwrap();
            db.close().unwrap();
        }

        fn refusal(dir: &Path) -> DatabaseError {
            match Database::open_with(dir, readonly()) {
                Err(error) => error,
                Ok(_) => panic!("expected \"{}\" to be refused", dir.display()),
            }
        }

        fn schema_refusal(error: DatabaseError) -> String {
            match error {
                DatabaseError::Schema(HaysonSchemaError(message)) => message,
                other => panic!("expected a schema refusal, got {other:?}"),
            }
        }

        #[test]
        fn a_readonly_open_refuses_naming_the_table_and_the_log_without_advising_create_table() {
            let dir = tempdir().unwrap();
            users_with_a_record(dir.path());
            let log = log_path(dir.path(), "users");
            fs::remove_file(&log).unwrap();

            let message = schema_refusal(refusal(dir.path()));

            assert!(message.contains("\"users\""), "{message}");
            assert!(message.contains(&log.display().to_string()), "{message}");
            assert!(message.contains("readonly"), "{message}");
            assert!(!message.contains("createTable"), "{message}");
            assert!(!log.exists(), "the refused open must not have created the log");
        }

        #[test]
        fn a_writable_open_of_the_same_directory_creates_the_log_and_serves_an_empty_table() {
            let dir = tempdir().unwrap();
            users_with_a_record(dir.path());
            fs::remove_file(log_path(dir.path(), "users")).unwrap();

            let mut db = Database::open(dir.path()).unwrap();

            assert_eq!(db.table_names(), vec!["users"]);
            assert!(log_path(dir.path(), "users").is_file());
            assert_eq!(db.table("users").unwrap().len(), 0);
            db.table_mut("users")
                .unwrap()
                .put(&json!({ "id": "u1", "name": "Ania" }), false)
                .unwrap();
            assert_eq!(db.table("users").unwrap().len(), 1);
            db.close().unwrap();
        }

        // NOTE: an empty log is a table nothing was written to; a missing one is a table gone.
        #[test]
        fn a_readonly_open_accepts_a_log_that_is_present_and_empty() {
            let dir = tempdir().unwrap();
            users_with_a_record(dir.path());
            fs::write(log_path(dir.path(), "users"), "").unwrap();

            let db = Database::open_with(dir.path(), readonly()).unwrap();

            assert_eq!(db.table_names(), vec!["users"]);
            assert_eq!(db.table("users").unwrap().len(), 0);
        }

        #[test]
        fn a_readonly_open_refuses_a_partitioned_table_whose_recorded_partition_is_gone() {
            let dir = tempdir().unwrap();
            notes_with_a_record(dir.path());
            let partition = dir.path().join("notes").join("note.jsonl");
            assert!(
                index_path(dir.path(), "notes").is_file(),
                "the refusal reads the snapshot"
            );
            fs::remove_dir_all(dir.path().join("notes")).unwrap();

            let message = schema_refusal(refusal(dir.path()));

            assert!(message.contains("\"notes\""), "{message}");
            assert!(message.contains(&partition.display().to_string()), "{message}");
            assert!(message.contains("readonly"), "{message}");
            assert!(!message.contains("createTable"), "{message}");
        }

        #[test]
        fn a_readonly_open_accepts_a_partitioned_table_that_was_never_written_to() {
            let dir = tempdir().unwrap();
            let mut db = Database::open(dir.path()).unwrap();
            db.create_table("notes", notes_spec()).unwrap();
            db.close().unwrap();

            let db = Database::open_with(dir.path(), readonly()).unwrap();

            assert_eq!(db.table_names(), vec!["notes"]);
            assert_eq!(db.table("notes").unwrap().len(), 0);
        }

        /// A partition the listing shows but the filesystem cannot deliver is corruption, not a
        /// table that opens empty — and the log set would otherwise create a fresh file over it.
        #[cfg(unix)]
        #[test]
        fn a_readonly_open_reports_a_partition_that_cannot_be_opened_against_its_own_path() {
            let dir = tempdir().unwrap();
            notes_with_a_record(dir.path());
            let partition = dir.path().join("notes").join("note.jsonl");
            fs::remove_file(&partition).unwrap();
            std::os::unix::fs::symlink(dir.path().join("notes").join("gone.jsonl"), &partition).unwrap();

            let error = refusal(dir.path());

            match error {
                DatabaseError::Log(path, inner) => {
                    assert_eq!(path, partition);
                    assert_eq!(inner.kind(), io::ErrorKind::NotFound);
                }
                other => panic!("expected a log refusal naming the partition, got {other:?}"),
            }
            assert!(!dir.path().join("notes").join("gone.jsonl").exists());
        }
    }

    mod describing_one_table {
        use super::*;

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

        /// A handle over both tables, with the catalog on disk moved on under it for `users`
        /// alone — the drift every gate here has to tell apart from a healthy `orders`.
        fn users_drifted_under(dir: &Path) -> Database {
            write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
            let mut db = Database::open(dir).unwrap();
            write_catalog_to(dir, &catalog_of(&[users_with_city(), orders_schema()]));
            assert!(db.refresh().unwrap());
            db
        }

        fn refusal_of(answered: Result<Value, DatabaseError>) -> String {
            match answered {
                Err(DatabaseError::Schema(error)) => error.to_string(),
                other => panic!("expected a schema refusal, got {other:?}"),
            }
        }

        #[test]
        fn describes_a_table_of_its_own_while_another_one_has_drifted() {
            let dir = tempdir().unwrap();
            let db = users_drifted_under(dir.path());

            let document = db.describe_table("orders").unwrap();

            assert_eq!(document["table"], json!("orders"));
            assert_eq!(document["revision"], json!(1));
            assert!(db.catalog().is_err());
        }

        #[test]
        fn refuses_the_table_that_drifted_itself() {
            let dir = tempdir().unwrap();
            let db = users_drifted_under(dir.path());

            let message = refusal_of(db.describe_table("users"));

            assert!(message.contains("\"users\""), "{message}");
            assert!(message.contains("revision 2"), "{message}");
            // NOTE: refusing replaces the reload, so the stale handle stays stale.
            assert_eq!(db.table("users").unwrap().schema().revision, 1);
        }

        #[test]
        fn describes_the_index_set_the_handle_serves_and_leaves_the_named_ones_out() {
            let dir = tempdir().unwrap();
            let mut catalog = catalog_of(&[users_schema()]);
            let entry = catalog.tables.get_mut("users").unwrap();
            entry.indexes = Some(vec![IndexEntry {
                name: "by_name".to_string(),
                fields: vec!["name".to_string()],
                kind: crate::catalog::CatalogIndexKind::Exact,
                unique: false,
                where_: None,
                expire_after_seconds: None,
            }]);
            write_catalog_to(dir.path(), &catalog);
            let db = Database::open(dir.path()).unwrap();

            let document = db.describe_table("users").unwrap();

            assert_eq!(
                document["fields"]["name"],
                json!({ "type": "string", "required": true })
            );
            assert!(document.as_object().unwrap().get("indexes").is_none());
            assert!(db.catalog().is_ok());
        }

        #[test]
        fn refuses_a_table_this_handle_holds_nowhere() {
            let dir = tempdir().unwrap();
            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
            let db = Database::open(dir.path()).unwrap();

            let message = refusal_of(db.describe_table("ghosts"));

            assert!(message.contains("declares no table \"ghosts\""), "{message}");
        }
    }
}