Skip to main content

hayson_engine/database/
database.rs

1use std::collections::{BTreeMap, HashMap, HashSet};
2use std::fmt;
3use std::fs;
4use std::io;
5use std::path::{Path, PathBuf};
6use std::time::Duration;
7
8use serde_json::Value;
9
10use crate::atomic::AtomicTemporaryFileError;
11use crate::catalog::{
12    check_integrity, declare_relation as declare_relation_in, empty_catalog, expand_indexes, index_identity,
13    parse_catalog, reference_of, relations_touching, write_catalog, Catalog, CatalogCompaction, CatalogLayout,
14    ExpandedIndex, IntegrityReport, Relation, RelationAction, TableEntry, TableOptions, TriggerEntry,
15};
16use crate::checkpoint::{read_snapshot, RestoreOutcome};
17use crate::crypto::{CipherOptions, FieldCipher};
18use crate::errors::{HaysonCipherError, HaysonRelationRestrictError, HaysonRelationViolationError, HaysonSchemaError};
19use crate::lock::{DirLock, DirLockError, LockOptions};
20use crate::query::builder::QueryBuilder;
21use crate::schema::{is_partitioned, parse_schema, FieldSpec, HaysonRecord, TableSchema};
22use crate::storage::{partition_names_in, LOG_SUFFIX};
23use crate::table::{describe_table_document, CachePolicy, FollowPolicy, Table, TableError};
24use crate::txn::{apply_prepared, recover_incomplete, PreparedUnit, TxnError, UnitOfWork, UnitOfWorkResult};
25
26use super::layout::{resolve_kind, stored_kind_of, Layout, LayoutKind, LayoutOptions, DEFAULT_METADATA_FILE};
27/// Matches the TS `DEFAULT_SNAPSHOT_EVERY` (`packages/hayson/src/database/database.ts`) — ops
28/// applied to a table between snapshots before the next commit writes one unprompted.
29const DEFAULT_SNAPSHOT_EVERY: u64 = 1000;
30/// Matches the TS `DEFAULT_COMPACTION` (`packages/hayson/src/database/database.ts`) — the garbage
31/// a table's log has to hold before a write rewrites it on its own.
32const DEFAULT_COMPACTION: CatalogCompaction = CatalogCompaction::Enabled {
33    dead_bytes_ratio: 0.4,
34    min_dead_bytes: 64 * 1024,
35};
36const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
37const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
38
39#[derive(Debug)]
40pub enum DatabaseError {
41    Lock(DirLockError),
42    Schema(HaysonSchemaError),
43    /// The handle was opened readonly and `operation` needs the writer. Its own variant rather than
44    /// a `Schema` refusal so a binding can tell it apart without reading the message.
45    ReadOnly {
46        dir: PathBuf,
47        operation: String,
48    },
49    /// `drop_field` refused because live records still carry the field. `records` is how many were
50    /// found, which is what a caller weighing `force` needs.
51    FieldInUse {
52        table: String,
53        field: String,
54        records: usize,
55    },
56    Catalog(String),
57    Io(io::Error),
58    /// An IO failure a caller can only act on knowing which file it was about: `io::Error` carries
59    /// no path, and a readonly open that cannot deliver one log has to name it.
60    Log(PathBuf, io::Error),
61    Table(String, TableError),
62    Txn(TxnError),
63    /// Boxed for the same reason as `TableError::Cipher` — the `open()`-time keyring is the only
64    /// thing that ever builds one, so nothing else should pay for its size at rest.
65    Cipher(Box<HaysonCipherError>),
66}
67
68impl fmt::Display for DatabaseError {
69    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
70        match self {
71            DatabaseError::Lock(error) => error.fmt(f),
72            DatabaseError::Schema(error) => error.fmt(f),
73            DatabaseError::ReadOnly { dir, operation } => {
74                write!(f, "database \"{}\" is readonly — {operation} is not available", dir.display())
75            }
76            DatabaseError::FieldInUse { table, field, records } => write!(
77                f,
78                "field \"{field}\" of table \"{table}\" still holds a value in {records} live record(s) — pass force to drop it anyway"
79            ),
80            DatabaseError::Catalog(message) => write!(f, "{message}"),
81            DatabaseError::Io(error) => error.fmt(f),
82            DatabaseError::Log(path, error) => write!(f, "{error}, open \"{}\"", path.display()),
83            DatabaseError::Table(name, error) => write!(f, "table \"{name}\": {error}"),
84            DatabaseError::Txn(error) => error.fmt(f),
85            DatabaseError::Cipher(error) => error.fmt(f),
86        }
87    }
88}
89
90impl std::error::Error for DatabaseError {}
91
92impl From<io::Error> for DatabaseError {
93    fn from(value: io::Error) -> Self {
94        DatabaseError::Io(value)
95    }
96}
97
98impl From<DirLockError> for DatabaseError {
99    fn from(value: DirLockError) -> Self {
100        DatabaseError::Lock(value)
101    }
102}
103
104impl From<HaysonSchemaError> for DatabaseError {
105    fn from(value: HaysonSchemaError) -> Self {
106        DatabaseError::Schema(value)
107    }
108}
109
110impl From<TxnError> for DatabaseError {
111    fn from(value: TxnError) -> Self {
112        DatabaseError::Txn(value)
113    }
114}
115
116impl From<HaysonCipherError> for DatabaseError {
117    fn from(value: HaysonCipherError) -> Self {
118        DatabaseError::Cipher(Box::new(value))
119    }
120}
121
122/// Whether the tables of this database catch up with foreign writes on their own — the port of the
123/// TS `OpenOptions.follow`, which takes `false`, `true` or `{ intervalMs }`.
124///
125/// [`FollowOption::Default`] resolves by `readonly`, exactly as the TS `resolveFollow` does: a
126/// readonly handle follows (at [`DEFAULT_FOLLOW_INTERVAL`], which is zero — check on every read), a
127/// writable one does not, since it is the only writer the directory allows and already sees its own
128/// writes.
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
130pub enum FollowOption {
131    #[default]
132    Default,
133    /// Never catch up unprompted; only an explicit [`Table::refresh`]/[`Database::refresh`] does.
134    Off,
135    /// Catch up on a read, at most once per this interval. `Duration::ZERO` means every read.
136    Every(Duration),
137}
138
139/// The interval a `follow` left at its default resolves to — the TS `DEFAULT_FOLLOW_INTERVAL_MS`,
140/// which is `0`: check on every read, which is one `stat` per log file when nothing changed.
141pub const DEFAULT_FOLLOW_INTERVAL: Duration = Duration::ZERO;
142
143impl FollowOption {
144    fn resolve(self, readonly: bool) -> Option<FollowPolicy> {
145        match self {
146            FollowOption::Off => None,
147            FollowOption::Every(interval) => Some(FollowPolicy { interval }),
148            FollowOption::Default if readonly => Some(FollowPolicy {
149                interval: DEFAULT_FOLLOW_INTERVAL,
150            }),
151            FollowOption::Default => None,
152        }
153    }
154}
155
156/// Whether `Database::open` takes the writer lock at all, and on what terms when it does.
157///
158/// [`LockMode::Skip`] is the port of the TS `open(dir, { lock: false })` and is dangerous by
159/// design: a handle that skipped the lock writes the catalog it holds in memory over whatever
160/// another writer declared meanwhile, because the lock is the only guard there. See
161/// `docs/format.md` and the TS `deklarowanie-indeksu-wspolbieznie.test.ts` for the exact window.
162#[derive(Debug, Clone, Copy)]
163pub enum LockMode {
164    /// One writer per directory (ADR-hjxp/0006), on this lease tuning.
165    Enabled(LockOptions),
166    /// No lock at all — mutual exclusion between writers of this directory is not enforced.
167    Skip,
168}
169
170impl Default for LockMode {
171    fn default() -> Self {
172        LockMode::Enabled(LockOptions::default())
173    }
174}
175
176impl LockMode {
177    /// The lease tuning of an enabled lock — `None` when the caller asked to skip locking, which
178    /// leaves nothing to tune.
179    pub fn options(&self) -> Option<&LockOptions> {
180        match self {
181            LockMode::Enabled(options) => Some(options),
182            LockMode::Skip => None,
183        }
184    }
185
186    /// The mutable half of [`LockMode::options`], for a host adjusting the defaults it opens with.
187    pub fn options_mut(&mut self) -> Option<&mut LockOptions> {
188        match self {
189            LockMode::Enabled(options) => Some(options),
190            LockMode::Skip => None,
191        }
192    }
193}
194
195/// How `Database::open` acquires the writer lock and whether it opens for writing at all.
196#[derive(Debug, Clone)]
197pub struct OpenOptions {
198    pub readonly: bool,
199    pub lock: LockMode,
200    /// Ops applied to a table between periodic `<table>.idx` snapshots — see [`Table::snapshot`].
201    /// `0` disables periodic snapshotting; defaults to [`DEFAULT_SNAPSHOT_EVERY`].
202    pub snapshot_every: u64,
203    /// Overrides the automatic-compaction policy of every table for this handle — the TS
204    /// `compaction` option, where `Disabled` is its `false`. `None` leaves the choice to the
205    /// catalog: the table's own `options.compaction`, then `defaults.compaction`, then
206    /// [`DEFAULT_COMPACTION`].
207    pub compaction: Option<CatalogCompaction>,
208    /// Keys for the fields a schema marks `cypher`. Without them such fields cannot be written.
209    pub cypher: Option<CipherOptions>,
210    /// Whether reads catch up with what other processes appended, without an explicit refresh.
211    pub follow: FollowOption,
212    /// How many parsed documents each table of this handle keeps in memory in front of its log —
213    /// see [`CachePolicy`]. `Disabled` by default, and named here rather than in the catalog on
214    /// purpose: it costs memory in this process only, so it is the opener's call and not a property
215    /// of the database on disk.
216    pub cache: CachePolicy,
217    /// Where in the directory the metadata document and the table files live. Left at its default,
218    /// this is the flat arrangement every database used before layouts existed, under
219    /// `hayson.json` — and an existing database is opened in the arrangement it is stored in,
220    /// whether or not the caller names one.
221    pub layout: LayoutOptions,
222}
223
224impl Default for OpenOptions {
225    fn default() -> Self {
226        OpenOptions {
227            readonly: false,
228            lock: LockMode::default(),
229            snapshot_every: DEFAULT_SNAPSHOT_EVERY,
230            compaction: None,
231            cypher: None,
232            follow: FollowOption::default(),
233            cache: CachePolicy::default(),
234            layout: LayoutOptions::default(),
235        }
236    }
237}
238
239fn now_ms() -> u64 {
240    chrono::Utc::now().timestamp_millis().max(0) as u64
241}
242
243/// A scalar reference value as plain text — a string reads back verbatim, not JSON-quoted, so it
244/// matches what the caller actually wrote. `reference_of` only ever hands this a `String`,
245/// `Number` or `Bool` (never `Null`/`Array`/`Object`), so `Value::to_string()` is the fallback for
246/// a variant that cannot occur here rather than a case this needs to get right.
247fn scalar_as_text(value: &Value) -> String {
248    match value {
249        Value::String(s) => s.clone(),
250        other => other.to_string(),
251    }
252}
253
254fn fnv1a(bytes: &[u8]) -> String {
255    let mut hash = FNV_OFFSET;
256    for byte in bytes {
257        hash ^= u64::from(*byte);
258        hash = hash.wrapping_mul(FNV_PRIME);
259    }
260    format!("{hash:016x}")
261}
262
263/// Identifies a `TableSchema` and the index set it is opened with for
264/// `checkpoint::restore_or_rebuild`'s pairing check — cheap and deterministic, not cryptographic:
265/// it only has to change whenever either does. The index set belongs in it because a snapshot
266/// entry carries the projection the registry held when it was written: a snapshot taken before an
267/// index was declared projects none of that index's fields, so restoring from it would rebuild the
268/// new index empty and answer queries against it with silence. An empty `declared` appends
269/// nothing, mirroring the TS `schemaFingerprint`.
270fn schema_fingerprint(schema: &TableSchema, declared: &[ExpandedIndex]) -> String {
271    if declared.is_empty() {
272        return fnv1a(format!("{schema:?}").as_bytes());
273    }
274    let mut described: Vec<String> = declared.iter().map(|index| format!("{index:?}")).collect();
275    described.sort();
276    fnv1a(format!("{schema:?}|{}", described.join("|")).as_bytes())
277}
278
279/// The inverse of `catalog::table_entry_of`: rebuilds the `TableSchema` a `TableEntry` describes,
280/// so a table can be opened straight from the catalog without a `<table>.schema.json` file. No
281/// such conversion exists in `catalog/` yet (format 1's `describeTableEntry` was never ported), so
282/// it lives here — the only place format 2 without format 1 needs it.
283fn table_schema_of(name: &str, entry: &TableEntry) -> TableSchema {
284    let fields: BTreeMap<String, FieldSpec> = entry
285        .fields
286        .iter()
287        .map(|(field, spec)| (field.clone(), spec.spec.clone()))
288        .collect();
289    TableSchema {
290        table: name.to_string(),
291        revision: entry.revision as u32,
292        primary_key: entry.primary_key.clone(),
293        partition_by: entry.partition_by.clone(),
294        fields,
295    }
296}
297
298/// The trigger declarations of `catalog` naming `name` as their source — what `Table::set_triggers`
299/// takes.
300fn triggers_of(catalog: &Catalog, name: &str) -> Vec<TriggerEntry> {
301    let Some(declared) = catalog.triggers.as_ref() else {
302        return Vec::new();
303    };
304    declared
305        .iter()
306        .filter(|trigger| trigger.table == name)
307        .cloned()
308        .collect()
309}
310
311/// Hands every open table the trigger declarations naming it as their source (`Table::
312/// set_triggers`'s contract) — called after `open()` rebuilds `tables` from the catalog, and again
313/// after `declare_trigger` changes `catalog.triggers`, so a live table is never out of step with
314/// what is on disk. A table this handle reopens on its own gets its set from
315/// [`open_catalog_table`] instead, which covers only the one table it brings up.
316fn sync_table_triggers(catalog: &Catalog, tables: &mut HashMap<String, Table>) {
317    for (name, table) in tables.iter_mut() {
318        table.set_triggers(triggers_of(catalog, name));
319    }
320}
321
322/// One table option as the catalog resolves it: what the table entry declares beats what
323/// `catalog.defaults` declares for the whole database, the same order [`compaction_of`] spells out
324/// for the two sources it shares with this.
325fn option_of<T>(catalog: &Catalog, entry: &TableEntry, pick: impl Fn(&TableOptions) -> Option<T>) -> Option<T> {
326    entry
327        .options
328        .as_ref()
329        .and_then(&pick)
330        .or_else(|| catalog.defaults.as_ref().and_then(&pick))
331}
332
333/// Applies the `imageDir` [`option_of`] resolves, against the database directory, to a just-opened
334/// table — left at its own computed default (a `<table>.images/` sibling of the log file) when
335/// neither the entry nor `catalog.defaults` names one.
336fn apply_image_dir(dir: &Path, catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
337    if let Some(image_dir) = option_of(catalog, entry, |options| options.image_dir.clone()) {
338        table.set_image_dir(dir.join(image_dir));
339    }
340}
341
342/// Applies the `sync` [`option_of`] resolves to a just-opened table — left at `Table`'s own
343/// `Commit` default when neither the entry nor `catalog.defaults` names one, which is the
344/// behaviour every table had while the field was parsed but never read (`docs/format.md`).
345fn apply_sync(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
346    if let Some(sync) = option_of(catalog, entry, |options| options.sync) {
347        table.set_sync(sync);
348    }
349}
350
351/// Applies the table entry's `validation` to a just-opened table — left at `Table`'s own `Strict`
352/// default when the entry declares none, which is the behaviour every table had while the field was
353/// parsed but never read (`docs/format.md`).
354///
355/// [`option_of`] does not apply here: `validation` is a key of the table section itself, and
356/// neither `tables.*.options` nor `catalog.defaults` carries it, so there is no database-wide
357/// declaration to fall back to.
358fn apply_validation(entry: &TableEntry, table: &mut Table) {
359    if let Some(validation) = entry.validation {
360        table.set_validation(validation);
361    }
362}
363
364/// Hands a just-opened table the fields its entry marks `deprecated`, each against its note (never
365/// empty — `parse_catalog` refuses `deprecated: ""`) — left empty when the entry marks none.
366///
367/// The mark lives on the catalog field spec, not on `schema::FieldSpec`, so the schema
368/// `table_schema_of` rebuilds cannot carry it and the table has to be told separately — the same
369/// way [`apply_validation`] tells it about a key of the table section.
370fn apply_deprecated(entry: &TableEntry, table: &mut Table) {
371    let marked: BTreeMap<String, String> = entry
372        .fields
373        .iter()
374        .filter_map(|(name, field)| field.deprecated.as_ref().map(|note| (name.clone(), note.clone())))
375        .collect();
376    if !marked.is_empty() {
377        table.set_deprecated_fields(marked);
378    }
379}
380
381/// Applies the `linkParsing` [`option_of`] resolves to a just-opened table — left at `Table`'s own
382/// `Eager` default when neither the entry nor `catalog.defaults` names one, per the "default eager,
383/// configurable to lazy" call.
384fn apply_link_parsing(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
385    if let Some(link_parsing) = option_of(catalog, entry, |options| options.link_parsing) {
386        table.set_link_parsing(link_parsing);
387    }
388}
389
390/// Arms a just-(re)opened table's automatic compaction, resolving the four places a policy can
391/// come from in the order the catalog documents: what `open()` named beats what the table entry
392/// declares, which beats `catalog.defaults`, which beats [`DEFAULT_COMPACTION`].
393fn compaction_of(named: Option<CatalogCompaction>, catalog: &Catalog, entry: Option<&TableEntry>) -> CatalogCompaction {
394    named
395        .or_else(|| {
396            entry
397                .and_then(|entry| entry.options.as_ref())
398                .and_then(|options| options.compaction)
399        })
400        .or_else(|| catalog.defaults.as_ref().and_then(|defaults| defaults.compaction))
401        .unwrap_or(DEFAULT_COMPACTION)
402}
403
404/// Arms a just-(re)opened table with the policy [`compaction_of`] resolved — the thresholds
405/// `Table::compact_if_due` measures each partition against.
406fn apply_compaction(compaction: CatalogCompaction, table: &mut Table) {
407    table.set_compaction(compaction);
408}
409
410/// Hands a just-(re)opened table the keyring `options.cypher` resolved into a `FieldCipher` — left
411/// at `None` when `open()` named no keys, which is fine for a table whose schema declares no
412/// `cypher` field and an error the moment one tries to write or read one that does.
413fn apply_cipher(cipher: Option<&FieldCipher>, table: &mut Table) {
414    table.set_cipher(cipher.cloned());
415}
416
417/// Counts live records of `open_table` still carrying `field` and refuses when any is left — the
418/// half of `Database::drop_field` that `force` waives, kept apart from the catalog half so the two
419/// can be sequenced independently.
420fn assert_no_residual_values(name: &str, open_table: &Table, field: &str) -> Result<(), DatabaseError> {
421    let all = open_table
422        .query(&QueryBuilder::new().spec())
423        .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
424    let mut residual = 0usize;
425    for key in &all.keys {
426        let record = open_table
427            .get(key)
428            .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
429        if record.is_some_and(|record| record.contains_key(field)) {
430            residual += 1;
431        }
432    }
433    if residual > 0 {
434        return Err(DatabaseError::FieldInUse {
435            table: name.to_string(),
436            field: field.to_string(),
437            records: residual,
438        });
439    }
440    Ok(())
441}
442
443/// Sorts a failed catalog write into the variant that keeps the most of it. `write_catalog` erases
444/// its refusals into `Box<dyn Error>`, and one of them — the `.tmp` path that could not be claimed,
445/// raised as an [`AtomicTemporaryFileError`] inside an `io::Error` — names a file the caller has to
446/// remove before retrying. Flattening it into `Catalog(String)` would leave that path recoverable
447/// only from the message text; everything else has nothing typed to keep and stays a catalog error.
448fn catalog_write_error(error: Box<dyn std::error::Error>) -> DatabaseError {
449    match error.downcast::<io::Error>() {
450        Ok(io) if AtomicTemporaryFileError::downcast(&io).is_some() => DatabaseError::Io(*io),
451        Ok(io) => DatabaseError::Catalog(io.to_string()),
452        Err(other) => DatabaseError::Catalog(other.to_string()),
453    }
454}
455
456/// Directory a partitioned table keeps its `<value>.jsonl` files in — the same one `Table::open`
457/// derives from the log path it is handed, spelled out here because this check runs before it.
458fn partition_dir_of(layout: &Layout, table: &str) -> PathBuf {
459    layout.tables().join(table)
460}
461
462/// Whether anything at all sits at `path`, without following a link: a dangling symlink is present
463/// here, so that it is the open — not this check — that reports what is wrong with it.
464fn on_disk(path: &Path) -> bool {
465    fs::symlink_metadata(path).is_ok()
466}
467
468/// The log files the catalog says this table has and the disk does not — port of the TS
469/// `missingLogsOf` (`packages/hayson/src/database/database.ts`).
470///
471/// An unpartitioned table has exactly one log, so its absence is the whole answer. A partitioned
472/// one has as many as its index snapshot claims: a table nothing was ever written to legitimately
473/// holds no file at all, and only a partition the snapshot recorded but the directory no longer
474/// lists counts as gone.
475fn missing_logs_of(layout: &Layout, index_path: &Path, schema: &TableSchema) -> io::Result<Vec<PathBuf>> {
476    let flat = layout.log_path(&schema.table);
477    if !is_partitioned(schema) {
478        return Ok(if on_disk(&flat) { Vec::new() } else { vec![flat] });
479    }
480    let dir = partition_dir_of(layout, &schema.table);
481    let present = partition_names_in(&dir);
482    // NOTE: a flat log under a partitioned schema is the move `open_logs` finishes, not a gap.
483    if present.is_empty() && on_disk(&flat) {
484        return Ok(Vec::new());
485    }
486    let claimed = read_snapshot(index_path)?
487        .and_then(|snapshot| snapshot.header.parts)
488        .unwrap_or_default();
489    Ok(claimed
490        .into_iter()
491        .filter(|part| !present.contains(&part.part))
492        .map(|part| dir.join(format!("{}{LOG_SUFFIX}", part.part)))
493        .collect())
494}
495
496/// The log files a readonly open of this table is about to take on, as the disk lists them.
497fn present_logs_of(layout: &Layout, schema: &TableSchema) -> Vec<PathBuf> {
498    let flat = layout.log_path(&schema.table);
499    if !is_partitioned(schema) {
500        return if on_disk(&flat) { vec![flat] } else { Vec::new() };
501    }
502    let dir = partition_dir_of(layout, &schema.table);
503    let present = partition_names_in(&dir);
504    if present.is_empty() {
505        return if on_disk(&flat) { vec![flat] } else { Vec::new() };
506    }
507    present
508        .iter()
509        .map(|part| dir.join(format!("{part}{LOG_SUFFIX}")))
510        .collect()
511}
512
513/// How one open table differs from the section the catalog on disk now holds for it — the port of
514/// the TS `TableDrift` (`packages/hayson/src/database/database.ts`, before the binding cutover).
515enum TableDrift {
516    Absent,
517    Schema { revision: u64 },
518    Indexes { unserved: Vec<String>, stale: Vec<String> },
519}
520
521struct DriftedTable {
522    name: String,
523    drift: TableDrift,
524}
525
526/// Names in `expected` that `present` holds no counterpart for, matched by [`index_identity`]
527/// rather than by name — a renamed index is the same index, and must not read as drift.
528fn index_names_missing_from(expected: &[ExpandedIndex], present: &[ExpandedIndex]) -> Vec<String> {
529    let identities: HashSet<String> = present.iter().map(index_identity).collect();
530    let mut names: Vec<String> = expected
531        .iter()
532        .filter(|index| !identities.contains(&index_identity(index)))
533        .map(|index| index.name.clone())
534        .collect();
535    names.sort();
536    names
537}
538
539/// The schema without the three field modifiers the index half of the drift check already covers
540/// in full: `unique` implies an exact index (`schema::parse`), and `index`/`fulltext` are what
541/// `sugar_indexes_of` expands, so nothing is lost by dropping them here.
542///
543/// They have to go, because a schema rebuilt from the catalog (`table_schema_of`) carries only the
544/// field modifiers the entry itself spells out, while one parsed from a schema document carries
545/// whatever `schema::parse` derived — comparing them would report a handle as drifted from its own
546/// write.
547fn shape_of(schema: &TableSchema) -> TableSchema {
548    let mut bare = schema.clone();
549    for spec in bare.fields.values_mut() {
550        spec.index = None;
551        spec.unique = false;
552        spec.fulltext = false;
553    }
554    bare
555}
556
557fn quoted(names: &[String]) -> String {
558    names
559        .iter()
560        .map(|name| format!("\"{name}\""))
561        .collect::<Vec<_>>()
562        .join(", ")
563}
564
565fn stored_side_of(drift: &TableDrift) -> String {
566    match drift {
567        TableDrift::Absent => "declares no table at all".to_string(),
568        TableDrift::Schema { revision } => format!("declares revision {revision}"),
569        TableDrift::Indexes { unserved, stale } => {
570            let mut halves = Vec::new();
571            if !unserved.is_empty() {
572                halves.push(format!(
573                    "declares indexes this handle does not serve: {}",
574                    quoted(unserved)
575                ));
576            }
577            if !stale.is_empty() {
578                halves.push(format!(
579                    "no longer declares indexes this handle still serves: {}",
580                    quoted(stale)
581                ));
582            }
583            halves.join(", and ")
584        }
585    }
586}
587
588fn missing_logs_error(dir: &Path, name: &str, missing: &[PathBuf]) -> HaysonSchemaError {
589    let listed = missing
590        .iter()
591        .map(|path| format!("\"{}\"", path.display()))
592        .collect::<Vec<_>>()
593        .join(", ");
594    let one = missing.len() == 1;
595    let (its, is, it, file, log) = if one {
596        ("its log".to_string(), "is", "it", "file", "log")
597    } else {
598        (format!("{} of its logs", missing.len()), "are", "them", "files", "logs")
599    };
600    HaysonSchemaError(format!(
601        "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",
602        dir.display()
603    ))
604}
605
606/// The guard a readonly open owes every table the catalog declares: a log the catalog names but the
607/// disk does not hold is a refusal, not a table that opens empty, because nothing on this handle
608/// will ever write it (F8, TS `#openIfOnDisk`).
609///
610/// A writable open needs none of this — it creates the logs it misses on the way in.
611///
612/// The two halves answer two different questions. The first is what the catalog promised and the
613/// listing denies; the second is whether what the listing does show can actually be opened, since
614/// `LogSet` opens its files with `create(true)` and would otherwise silently make a fresh log out
615/// of a path that is broken — a dangling link being the case that shows it.
616fn refuse_unreadable_logs(layout: &Layout, name: &str, entry: &TableEntry) -> Result<(), DatabaseError> {
617    let schema = table_schema_of(name, entry);
618    let missing = missing_logs_of(layout, &layout.index_path(name), &schema)?;
619    if !missing.is_empty() {
620        return Err(DatabaseError::Schema(missing_logs_error(layout.root(), name, &missing)));
621    }
622    for path in present_logs_of(layout, &schema) {
623        fs::File::open(&path).map_err(|error| DatabaseError::Log(path, error))?;
624    }
625    Ok(())
626}
627
628/// Brings up one table exactly as the catalog describes it: the schema rebuilt from its entry, its
629/// snapshot paired against its log, the catalog's full expanded index list, and the options
630/// (`imageDir`, `linkParsing`, `sync`, `validation`, `compaction`, the keyring, the follow policy,
631/// the triggers naming it as their source, the fields its entry marks `deprecated`) applied on
632/// top. Every caller that swaps a live handle out — `open`, [`Database::refresh`],
633/// [`Database::create_table`], [`Database::drop_field`], [`Database::declare_index`] — goes through
634/// here, so the set of things a reopened table carries has one place to grow.
635#[allow(clippy::too_many_arguments)]
636fn open_catalog_table(
637    layout: &Layout,
638    name: &str,
639    catalog: &Catalog,
640    entry: &TableEntry,
641    snapshot_every: u64,
642    cipher: Option<&FieldCipher>,
643    follow: Option<FollowPolicy>,
644    readonly: bool,
645    compaction: CatalogCompaction,
646    cache: CachePolicy,
647) -> Result<(Table, RestoreOutcome), DatabaseError> {
648    let schema = table_schema_of(name, entry);
649    let declared = expand_indexes(name, entry);
650    let hash = schema_fingerprint(&schema, &declared);
651    let (mut table, outcome) = Table::open_with_snapshot_and_declared_indexes(
652        schema,
653        layout.log_path(name),
654        layout.index_path(name),
655        &hash,
656        snapshot_every,
657        readonly,
658        &declared,
659    )
660    .map_err(|error| DatabaseError::Table(name.to_string(), TableError::Io(error)))?;
661    apply_image_dir(layout.root(), catalog, entry, &mut table);
662    apply_link_parsing(catalog, entry, &mut table);
663    apply_sync(catalog, entry, &mut table);
664    apply_validation(entry, &mut table);
665    apply_deprecated(entry, &mut table);
666    apply_compaction(compaction, &mut table);
667    table.set_cache(cache);
668    apply_cipher(cipher, &mut table);
669    table.set_follow(follow);
670    table.set_triggers(triggers_of(catalog, name));
671    Ok((table, outcome))
672}
673
674/// The first table log in `layout`, looked for where the layout puts them: directly in the tables
675/// directory, and one level down inside it, which is where a partitioned table keeps its per-value
676/// logs. Sorted, so the same directory always names the same log in a refusal.
677fn first_table_log(layout: &Layout) -> Option<PathBuf> {
678    let has_log_suffix = |path: &Path| {
679        path.extension()
680            .is_some_and(|ext| format!(".{}", ext.to_string_lossy()) == LOG_SUFFIX)
681    };
682    let entries_of = |dir: &Path| -> Vec<PathBuf> {
683        let Ok(read) = fs::read_dir(dir) else { return Vec::new() };
684        let mut found: Vec<PathBuf> = read.filter_map(Result::ok).map(|entry| entry.path()).collect();
685        found.sort();
686        found
687    };
688
689    let top = entries_of(layout.tables());
690    if let Some(log) = top.iter().find(|path| path.is_file() && has_log_suffix(path)) {
691        return Some(log.clone());
692    }
693    top.iter().filter(|path| path.is_dir()).find_map(|dir| {
694        entries_of(dir)
695            .into_iter()
696            .find(|path| path.is_file() && has_log_suffix(path))
697    })
698}
699
700/// Refuses to treat a directory that already holds table logs as a brand new database.
701///
702/// Without `hayson.json` there is nothing that says which schema those logs were written under, so
703/// the only safe answer is to stop: opening would otherwise write a fresh empty catalog next to real
704/// data, and the next write would append records of a foreign schema onto a log nobody can read back.
705/// A `.gitignore` that swallowed the catalog, or a restore that copied only the data, both look
706/// exactly like this. See `packages/hayson/test/katalog-bez-metadanych.test.ts`.
707fn refuse_orphan_logs(layout: &Layout) -> Result<(), HaysonSchemaError> {
708    let Some(log) = first_table_log(layout) else {
709        return Ok(());
710    };
711    Err(HaysonSchemaError(format!(
712        "\"{}\" 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",
713        layout.root().display(),
714        layout.metadata_file(),
715        log.display(),
716        layout.metadata_file(),
717    )))
718}
719
720fn read_catalog(path: &Path) -> Result<Option<Catalog>, DatabaseError> {
721    let raw = match fs::read_to_string(path) {
722        Ok(raw) => raw,
723        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
724        Err(error) => return Err(error.into()),
725    };
726    let value: Value = serde_json::from_str(&raw)
727        .map_err(|error| DatabaseError::Catalog(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
728    Ok(Some(parse_catalog(&value)?))
729}
730
731/// Owns the writer lock for the length of one operation and no longer: every refusal after
732/// `DirLock::acquire` unwinds through this `Drop` and takes the lock file with it, while a
733/// successful open moves the lock out with [`HeldLock::into_held`] and keeps it for the handle.
734///
735/// `DirLock` itself does not clean up on drop (its `release` is the only place that removes files),
736/// so without this a refused open would leave a lock file naming a process that is still alive —
737/// which no staleness rule ever reclaims, and which locks the directory out for good.
738///
739/// Shared with `migrate_layout`, which holds the same lock for the length of a move and wants the
740/// same guarantee about a refusal halfway through it.
741pub(super) struct HeldLock(pub(super) Option<DirLock>);
742
743impl HeldLock {
744    fn into_held(mut self) -> Option<DirLock> {
745        self.0.take()
746    }
747}
748
749impl Drop for HeldLock {
750    fn drop(&mut self) {
751        if let Some(lock) = self.0.as_mut() {
752            lock.release();
753        }
754    }
755}
756
757/// A directory of tables: the JSONL logs, their index snapshots, the metadata catalog
758/// (`hayson.json`) describing them, and one writer lock (`lock::DirLock`, ADR-hjxp/0006 — one
759/// writer per database directory).
760pub struct Database {
761    layout: Layout,
762    lock: Option<DirLock>,
763    catalog: Catalog,
764    tables: HashMap<String, Table>,
765    restore_outcomes: HashMap<String, RestoreOutcome>,
766    readonly: bool,
767    snapshot_every: u64,
768    /// What `open()` named, not what any table resolved to — kept so a table opened later by
769    /// `refresh`/`createTable` is armed on the same terms as the ones opened at startup.
770    compaction: Option<CatalogCompaction>,
771    cipher: Option<FieldCipher>,
772    follow: Option<FollowPolicy>,
773    /// What `open()` named, for the same reason as `compaction`: a table opened later by
774    /// `refresh`/`createTable` caches on the same terms as the ones opened at startup.
775    cache: CachePolicy,
776}
777
778impl Database {
779    pub fn open(dir: &Path) -> Result<Database, DatabaseError> {
780        Database::open_with(dir, OpenOptions::default())
781    }
782
783    /// Opens (and creates when missing) a Hayson database directory:
784    ///
785    /// 1. locks the directory for writing, unless `options.readonly` or `options.lock` is
786    ///    [`LockMode::Skip`] (ADR-hjxp/0006) — at the path the arrangement on disk puts the lock,
787    ///    settled under the lock itself (see [`Database::hold_for_open`]);
788    /// 2. reads `hayson.json`, or starts an empty catalog when the directory is new;
789    /// 3. for each declared table, opens its log and restores its registry from `<table>.idx`
790    ///    via `checkpoint::restore_or_rebuild` rather than always scanning the whole log;
791    /// 4. finishes an unfinished `.hayson.txn` journal, if one is on disk, before returning —
792    ///    so no write reaches a table until whatever a prior crash left half-committed is done.
793    ///
794    /// A refusal at any step after the lock was taken releases it again before returning, so a
795    /// directory this call rejected is left no more locked than it was found.
796    pub fn open_with(dir: &Path, options: OpenOptions) -> Result<Database, DatabaseError> {
797        let (layout, lock) = match options.lock {
798            LockMode::Enabled(lock) if !options.readonly => Database::hold_for_open(dir, &options.layout, lock)?,
799            LockMode::Enabled(_) | LockMode::Skip => (Database::resolve_layout(dir, &options.layout)?, HeldLock(None)),
800        };
801
802        if !options.readonly {
803            for directory in layout.directories() {
804                fs::create_dir_all(directory)?;
805            }
806        }
807
808        Database::open_locked(layout, options, lock)
809    }
810
811    /// Takes the writer lock where the arrangement on disk puts it, and settles which arrangement
812    /// that is only once the lock is held.
813    ///
814    /// The arrangement has to be probed before the lock can be taken, because the lock file lives
815    /// where the arrangement says (`.hayson.lock` in the root of a flat database, `runtime/` in a
816    /// nested one). A `migrate_layout` that ran between the probe and the lock moved every file and
817    /// released the lock of the old arrangement — so the lock taken on the probe's word is one
818    /// nobody contends for, and the paths the probe named point at files that are no longer there:
819    /// a writer that went on would open empty logs beside the real ones, and the next move would
820    /// not see it. So the probe is read again under the lock, the way `migrate_layout` reads it
821    /// again under its own; when it disagrees, the lock is released and taken again where the
822    /// arrangement now on disk puts it.
823    ///
824    /// Bounded rather than open-ended: the arrangement can only change while the lock it names is
825    /// free, so a second disagreement means something is moving the directory without the lock,
826    /// and that is reported, not waited out.
827    ///
828    /// Only the lock's own directory is created before the lock is held (as in `hold_for_move`):
829    /// the rest of the arrangement is created by the caller once it is settled, so a probe that was
830    /// already stale does not leave an empty `tables/` and `index/` in a flat database.
831    fn hold_for_open(
832        dir: &Path,
833        options: &LayoutOptions,
834        lock: LockOptions,
835    ) -> Result<(Layout, HeldLock), DatabaseError> {
836        const SETTLE_ATTEMPTS: usize = 3;
837        let mut probed = Database::resolve_layout(dir, options)?;
838        for _ in 0..SETTLE_ATTEMPTS {
839            fs::create_dir_all(probed.runtime())?;
840            let held = HeldLock(Some(DirLock::acquire(probed.lock_path(), lock)?));
841            let settled = Database::resolve_layout(dir, options)?;
842            if settled.kind() == probed.kind() {
843                return Ok((settled, held));
844            }
845            // NOTE: release before the cleanup — a nested lock file lives in the directory removed.
846            drop(held);
847            if probed.kind() == LayoutKind::Nested {
848                let _ = fs::remove_dir(probed.runtime());
849            }
850            probed = settled;
851        }
852        Err(HaysonSchemaError(format!(
853            "\"{}\" changed arrangement {SETTLE_ATTEMPTS} times while its writer lock was being taken — something keeps moving it without the lock; open it again once the moves are over",
854            dir.display()
855        ))
856        .into())
857    }
858
859    /// Works out where this database's files are before anything is opened or locked: the metadata
860    /// document name the options ask for (refusing `metadataFile` and `manifest` naming two
861    /// different files), then the arrangement — whatever the document on disk records, or the
862    /// requested one when there is no document yet.
863    ///
864    /// Runs before `create_dir_all`, so a refusal here leaves the directory exactly as it found it.
865    /// Public for that same reason: a caller that only needs to know where the writer lock or the
866    /// metadata document of a directory sits gets the paths without opening the database, taking
867    /// the lock, or creating anything.
868    ///
869    /// @param dir the database directory
870    /// @param options the `layout`/`metadataFile`/`manifest` triple of `OpenOptions`, unresolved
871    /// @returns where this database's files are, or a refusal to name the metadata document twice
872    pub fn resolve_layout(dir: &Path, options: &LayoutOptions) -> Result<Layout, DatabaseError> {
873        let metadata_file = options.metadata_file()?.unwrap_or(DEFAULT_METADATA_FILE);
874        let stored = stored_kind_of(&dir.join(metadata_file));
875        let kind = resolve_kind(dir, stored, options.kind)?;
876        Ok(Layout::new(dir, kind, metadata_file))
877    }
878
879    fn open_locked(layout: Layout, options: OpenOptions, lock: HeldLock) -> Result<Database, DatabaseError> {
880        let found = read_catalog(layout.metadata_path())?;
881        let catalog = match found {
882            Some(catalog) => catalog,
883            None => {
884                refuse_orphan_logs(&layout)?;
885                let catalog = empty_catalog(CatalogLayout::from(layout.kind()), now_ms());
886                if !options.readonly {
887                    write_catalog(layout.metadata_path(), &catalog)
888                        .map_err(|error| DatabaseError::Catalog(error.to_string()))?;
889                }
890                catalog
891            }
892        };
893
894        let cipher = options.cypher.clone().map(FieldCipher::new).transpose()?;
895        let follow = options.follow.resolve(options.readonly);
896
897        let mut tables = HashMap::new();
898        let mut restore_outcomes = HashMap::new();
899        for (name, entry) in &catalog.tables {
900            if options.readonly {
901                refuse_unreadable_logs(&layout, name, entry)?;
902            }
903            let (table, outcome) = open_catalog_table(
904                &layout,
905                name,
906                &catalog,
907                entry,
908                options.snapshot_every,
909                cipher.as_ref(),
910                follow,
911                options.readonly,
912                compaction_of(options.compaction, &catalog, Some(entry)),
913                options.cache,
914            )?;
915            tables.insert(name.clone(), table);
916            restore_outcomes.insert(name.clone(), outcome);
917        }
918
919        if !options.readonly {
920            recover_incomplete(&layout.journal_path(), &mut tables)?;
921        }
922
923        sync_table_triggers(&catalog, &mut tables);
924
925        Ok(Database {
926            layout,
927            lock: lock.into_held(),
928            catalog,
929            tables,
930            restore_outcomes,
931            readonly: options.readonly,
932            snapshot_every: options.snapshot_every,
933            compaction: options.compaction,
934            cipher,
935            follow,
936            cache: options.cache,
937        })
938    }
939
940    pub fn dir(&self) -> &Path {
941        self.layout.root()
942    }
943
944    /// Where this handle's files are on disk — the arrangement and the metadata document name it
945    /// resolved at open, the read side of the TS `Database.layout` property.
946    pub fn layout(&self) -> &Layout {
947        &self.layout
948    }
949
950    /// The follow policy every table of this database was opened with — `None` when reads never
951    /// catch up unprompted, which is what a writable open resolves to.
952    pub fn follow(&self) -> Option<FollowPolicy> {
953        self.follow
954    }
955
956    pub fn table_names(&self) -> Vec<&str> {
957        let mut names: Vec<&str> = self.tables.keys().map(String::as_str).collect();
958        names.sort_unstable();
959        names
960    }
961
962    pub fn table(&self, name: &str) -> Option<&Table> {
963        self.tables.get(name)
964    }
965
966    /// Handing out a writable table is the write path this type cannot see the end of, so the
967    /// lease is renewed here as well — a host that never calls [`DirLock::tick`] still keeps its
968    /// lock as long as it keeps writing. A lock already lost cannot be reported through an
969    /// `Option`; the next call that returns a `Result` — `commit` and every schema change — raises
970    /// [`DatabaseError::Lock`] instead.
971    pub fn table_mut(&mut self, name: &str) -> Option<&mut Table> {
972        let _ = self.renew_lock();
973        self.tables.get_mut(name)
974    }
975
976    /// Every `enforce: true` relation with `from.table == table`, checked against `values` (a `put`
977    /// document or a `patch`'s changes — either is a JSON object, and only the fields it actually
978    /// names are checked). A field the relation names but `values` does not touch is left alone: for
979    /// `put` that never happens (the whole document is `values`), and for `patch` it means the
980    /// existing value stands, already valid the last time it was written under enforcement — or
981    /// grandfathered in from before `enforce` was declared, which is [`Database::declare_relation`]'s
982    /// call to make, not this one's to re-litigate on every unrelated patch.
983    ///
984    /// `to.field` is guaranteed to be `to.table`'s primary key or a `unique` field (the relation
985    /// parser refuses anything else), so [`Table::contains_value`] answers in O(1) off the registry
986    /// — no scan, unlike [`Database::check_integrity`].
987    ///
988    /// `HaysonRelationViolationError::value` reports the offending value as plain text via
989    /// [`scalar_as_text`], not `Value::to_string()` — the latter would quote a string scalar
990    /// (`"b_gone"`), which every other typed error field on this boundary (e.g.
991    /// `HaysonUniqueViolationError::holder`) reports unquoted, and a caller comparing it against
992    /// the value it wrote should not have to know the difference.
993    fn assert_relations_hold(&self, table: &str, values: &Value) -> Result<(), DatabaseError> {
994        for relation in relations_touching(&self.catalog, Some(table)) {
995            if relation.from.table != table || relation.enforce != Some(true) {
996                continue;
997            }
998            let Some(value) = reference_of(values.get(&relation.from.field)) else {
999                continue;
1000            };
1001            let Some(to_table) = self.tables.get(&relation.to.table) else {
1002                continue;
1003            };
1004            if !to_table.contains_value(&relation.to.field, &value) {
1005                return Err(DatabaseError::Table(
1006                    table.to_string(),
1007                    TableError::Relation(HaysonRelationViolationError {
1008                        relation: relation.name.clone(),
1009                        table: table.to_string(),
1010                        field: relation.from.field.clone(),
1011                        value: scalar_as_text(&value),
1012                    }),
1013                ));
1014            }
1015        }
1016        Ok(())
1017    }
1018
1019    /// [`Table::put`], refusing first if `document` breaks an `enforce: true` relation declared on
1020    /// `table`. `table_mut(name).put(...)` skips this — every caller that wants enforcement honoured
1021    /// has to come in through here, because [`Table`] itself cannot see another table's registry to
1022    /// check against (only [`Database`] holds every table at once).
1023    pub fn put_checked(&mut self, table: &str, document: &Value, force: bool) -> Result<HaysonRecord, DatabaseError> {
1024        self.assert_relations_hold(table, document)?;
1025        let root = self.layout.root().display().to_string();
1026        let handle = self
1027            .table_mut(table)
1028            .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1029        handle
1030            .put(document, force)
1031            .map_err(|error| DatabaseError::Table(table.to_string(), error))
1032    }
1033
1034    /// [`Table::patch`], refusing first if `changes` breaks an `enforce: true` relation declared on
1035    /// `table`. See [`Database::put_checked`] for why this exists alongside `table_mut(...).patch`.
1036    pub fn patch_checked(
1037        &mut self,
1038        table: &str,
1039        key: &str,
1040        changes: &Value,
1041        force: bool,
1042    ) -> Result<HaysonRecord, DatabaseError> {
1043        self.assert_relations_hold(table, changes)?;
1044        let root = self.layout.root().display().to_string();
1045        let handle = self
1046            .table_mut(table)
1047            .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1048        handle
1049            .patch(key, changes, force)
1050            .map_err(|error| DatabaseError::Table(table.to_string(), error))
1051    }
1052
1053    /// [`Table::stage_put`], refusing first if `document` breaks an `enforce: true` relation declared
1054    /// on `table`. Same rationale as [`Database::put_checked`]; the napi/wasm bindings call this
1055    /// instead of `table_mut(name).stage_put(...)` because their staged writes still run synchronously
1056    /// with full [`Database`] access at the call site, same as a direct `put`.
1057    pub fn stage_put_checked(
1058        &mut self,
1059        table: &str,
1060        document: &Value,
1061        force: bool,
1062    ) -> Result<HaysonRecord, DatabaseError> {
1063        self.assert_relations_hold(table, document)?;
1064        let root = self.layout.root().display().to_string();
1065        let handle = self
1066            .table_mut(table)
1067            .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1068        handle
1069            .stage_put(document, force)
1070            .map_err(|error| DatabaseError::Table(table.to_string(), error))
1071    }
1072
1073    /// [`Table::stage_patch`], refusing first if `changes` breaks an `enforce: true` relation declared
1074    /// on `table`. See [`Database::stage_put_checked`].
1075    pub fn stage_patch_checked(
1076        &mut self,
1077        table: &str,
1078        key: &str,
1079        changes: &Value,
1080        force: bool,
1081    ) -> Result<HaysonRecord, DatabaseError> {
1082        self.assert_relations_hold(table, changes)?;
1083        let root = self.layout.root().display().to_string();
1084        let handle = self
1085            .table_mut(table)
1086            .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1087        handle
1088            .stage_patch(key, changes, force)
1089            .map_err(|error| DatabaseError::Table(table.to_string(), error))
1090    }
1091
1092    /// Deletes `key` from `table`, honouring every `on_delete` relation naming `table` as the `to`
1093    /// side — `restrict` refuses if a referencing record still exists, `cascade` deletes every
1094    /// referencing record (recursively, watching for a cycle in the relation graph), `set-null`
1095    /// clears the referencing field instead of deleting the row, and `no-action`/undeclared leaves
1096    /// referencing records untouched exactly as `Table::delete` always has (checked, not fixed —
1097    /// [`Database::check_integrity`] still finds what this deliberately leaves behind).
1098    ///
1099    /// A table nothing points at (no relation names it as `to`) skips the plan and goes straight
1100    /// through `Table::delete` — the common case pays nothing extra, and `force` there behaves as
1101    /// it always has. Every effect a cascade needs lands in one [`UnitOfWork`]
1102    /// ([`Database::commit`]), so a `restrict` found three tables deep aborts the whole delete with
1103    /// nothing written, not a half-cascaded one. `force` has no equivalent inside that unit — a
1104    /// fresh unit has no pending per-key conflict to force past the way a plain `Table::delete`
1105    /// does — so it is honoured only on the no-relation fast path.
1106    pub fn delete_checked(&mut self, table: &str, key: &str, force: bool) -> Result<bool, DatabaseError> {
1107        if self.readonly {
1108            return Err(DatabaseError::ReadOnly {
1109                dir: self.layout.root().to_path_buf(),
1110                operation: "delete".to_string(),
1111            });
1112        }
1113        let root = self.layout.root().display().to_string();
1114        if self.tables.get(table).is_none() {
1115            return Err(DatabaseError::Catalog(format!(
1116                "table \"{table}\" does not exist in \"{root}\""
1117            )));
1118        }
1119        let incoming = relations_touching(&self.catalog, Some(table))
1120            .into_iter()
1121            .any(|relation| relation.to.table == table);
1122        if !incoming {
1123            self.renew_lock()?;
1124            let handle = self.tables.get_mut(table).expect("checked above");
1125            return handle
1126                .delete(key, force)
1127                .map_err(|error| DatabaseError::Table(table.to_string(), error));
1128        }
1129
1130        let mut unit = UnitOfWork::new();
1131        let mut visited: HashSet<(String, String)> = HashSet::new();
1132        let existed = self.plan_delete_cascade(table, key, &mut visited, &mut unit)?;
1133        if !existed {
1134            return Ok(false);
1135        }
1136        self.commit(&mut unit)?;
1137        Ok(true)
1138    }
1139
1140    /// Stages `table`.`key`'s own delete plus every `cascade`/`set-null` effect an `on_delete`
1141    /// relation pointing at it requires, recursing into further cascades. Returns whether `key`
1142    /// was there to delete (mirrors `Table::delete`'s own answer) — `false` short-circuits before
1143    /// staging anything, matching `Table::delete`'s no-op on a missing key.
1144    ///
1145    /// `visited` guards a cycle in the cascade graph (A cascades into B which cascades back into
1146    /// A): a `(table, key)` already staged in this walk is not staged or walked a second time —
1147    /// the plan's own recursion is the only thing this needs to protect against, not a live-data
1148    /// cycle, since the walk never revisits a pair it has already resolved.
1149    fn plan_delete_cascade(
1150        &self,
1151        table: &str,
1152        key: &str,
1153        visited: &mut HashSet<(String, String)>,
1154        unit: &mut UnitOfWork,
1155    ) -> Result<bool, DatabaseError> {
1156        if !visited.insert((table.to_string(), key.to_string())) {
1157            return Ok(true);
1158        }
1159        let handle = self
1160            .tables
1161            .get(table)
1162            .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist")))?;
1163        let Some(record) = handle
1164            .get(key)
1165            .map_err(|error| DatabaseError::Table(table.to_string(), error))?
1166        else {
1167            return Ok(false);
1168        };
1169
1170        for relation in relations_touching(&self.catalog, Some(table)) {
1171            if relation.to.table != table {
1172                continue;
1173            }
1174            let action = relation.on_delete.unwrap_or(RelationAction::NoAction);
1175            if matches!(action, RelationAction::NoAction) {
1176                continue;
1177            }
1178            let Some(value) = reference_of(record.get(&relation.to.field)) else {
1179                continue;
1180            };
1181            let Some(from_table) = self.tables.get(&relation.from.table) else {
1182                continue;
1183            };
1184            let referencing = from_table
1185                .keys_referencing(&relation.from.field, &value)
1186                .map_err(|error| DatabaseError::Table(relation.from.table.clone(), error))?;
1187            if referencing.is_empty() {
1188                continue;
1189            }
1190
1191            match action {
1192                RelationAction::Restrict => {
1193                    return Err(DatabaseError::Table(
1194                        table.to_string(),
1195                        TableError::RelationRestrict(HaysonRelationRestrictError {
1196                            relation: relation.name.clone(),
1197                            table: table.to_string(),
1198                            key: key.to_string(),
1199                            referencing_table: relation.from.table.clone(),
1200                            referencing_field: relation.from.field.clone(),
1201                            referencing: referencing.len(),
1202                        }),
1203                    ));
1204                }
1205                RelationAction::Cascade => {
1206                    for referencing_key in &referencing {
1207                        self.plan_delete_cascade(&relation.from.table, referencing_key, visited, unit)?;
1208                    }
1209                }
1210                RelationAction::SetNull => {
1211                    for referencing_key in &referencing {
1212                        unit.patch(
1213                            relation.from.table.as_str(),
1214                            referencing_key.clone(),
1215                            Value::Object(serde_json::Map::from_iter([(relation.from.field.clone(), Value::Null)])),
1216                        );
1217                    }
1218                }
1219                RelationAction::NoAction => unreachable!("filtered above"),
1220            }
1221        }
1222
1223        unit.delete(table, key.to_string());
1224        Ok(true)
1225    }
1226
1227    /// Renews the writer lease from outside — [`DirLock::tick`] reached through the lock this
1228    /// handle owns privately. A host without background threads (`wasm32-wasip1`) has no heartbeat
1229    /// of its own, so this is what a timer on its side drives; call it at least twice per
1230    /// `LockOptions::heartbeat_ms`. Does nothing on a readonly handle, which holds no lock.
1231    pub fn tick(&self) -> Result<(), DatabaseError> {
1232        self.renew_lock()
1233    }
1234
1235    fn renew_lock(&self) -> Result<(), DatabaseError> {
1236        match &self.lock {
1237            Some(lock) => Ok(lock.tick()?),
1238            None => Ok(()),
1239        }
1240    }
1241
1242    /// How the table named `name` was brought up at open — `Restored` from `<table>.idx` or
1243    /// `Rebuilt` by a full scan of its log. `None` when the database holds no such table.
1244    pub fn restore_outcome(&self, name: &str) -> Option<RestoreOutcome> {
1245        self.restore_outcomes.get(name).copied()
1246    }
1247
1248    /// The metadata catalog as this handle last read it, without the drift gate — every table it
1249    /// describes may be one this handle does not serve. `catalog` is what a caller outside the
1250    /// engine should ask for.
1251    pub fn catalog_unchecked(&self) -> &Catalog {
1252        &self.catalog
1253    }
1254
1255    /// The metadata catalog as this handle last read it — the parsed form of `hayson.json`, which
1256    /// `catalog::catalog_document` turns back into the document a caller can serialize.
1257    ///
1258    /// Refuses when any table open here has drifted from what the catalog now declares, because
1259    /// handing the document over would describe tables this handle does not serve. Refusing is all
1260    /// that happens: the handle is never reloaded behind the caller's back, so the schema and the
1261    /// index set every open table was opened with stay exactly as they were.
1262    pub fn catalog(&self) -> Result<&Catalog, DatabaseError> {
1263        let drifted = self.drifted_tables();
1264        if drifted.is_empty() {
1265            return Ok(&self.catalog);
1266        }
1267        Err(DatabaseError::Schema(self.drift_error(&drifted)))
1268    }
1269
1270    /// The schema of one table as a standalone document, the shape `create_table` accepts — what
1271    /// the TS `describeTable` answers with.
1272    ///
1273    /// Read off the live handle (`table::describe_table_document`), not off the catalog entry, so
1274    /// it describes the schema and the index set the table is actually being served with — every
1275    /// table the catalog declares is open here, both after `open` and after
1276    /// [`Database::refresh`], so there is no catalog-only table to fall back to.
1277    ///
1278    /// The drift gate here is the one table named, unlike [`Database::catalog`], whose document
1279    /// covers every table and so has to answer for every table: a table of its own that moved on
1280    /// says nothing about this one.
1281    pub fn describe_table(&self, name: &str) -> Result<Value, DatabaseError> {
1282        if let Some(drift) = self.drift_of(name) {
1283            let drifted = [DriftedTable {
1284                name: name.to_string(),
1285                drift,
1286            }];
1287            return Err(DatabaseError::Schema(self.drift_error(&drifted)));
1288        }
1289        let table = self.tables.get(name).ok_or_else(|| {
1290            DatabaseError::Schema(HaysonSchemaError(format!("the catalog declares no table \"{name}\"")))
1291        })?;
1292        Ok(describe_table_document(table))
1293    }
1294
1295    fn drift_of(&self, name: &str) -> Option<TableDrift> {
1296        let table = self.tables.get(name)?;
1297        let Some(entry) = self.catalog.tables.get(name) else {
1298            return Some(TableDrift::Absent);
1299        };
1300        if shape_of(&table_schema_of(name, entry)) != shape_of(table.schema()) {
1301            return Some(TableDrift::Schema {
1302                revision: entry.revision,
1303            });
1304        }
1305        let declared = expand_indexes(name, entry);
1306        let served = table.declared_indexes();
1307        let unserved = index_names_missing_from(&declared, served);
1308        let stale = index_names_missing_from(served, &declared);
1309        if unserved.is_empty() && stale.is_empty() {
1310            return None;
1311        }
1312        Some(TableDrift::Indexes { unserved, stale })
1313    }
1314
1315    fn drifted_tables(&self) -> Vec<DriftedTable> {
1316        let mut names: Vec<&String> = self.tables.keys().collect();
1317        names.sort();
1318        names
1319            .into_iter()
1320            .filter_map(|name| {
1321                self.drift_of(name).map(|drift| DriftedTable {
1322                    name: name.clone(),
1323                    drift,
1324                })
1325            })
1326            .collect()
1327    }
1328
1329    fn drift_error(&self, drifted: &[DriftedTable]) -> HaysonSchemaError {
1330        let detail = drifted
1331            .iter()
1332            .map(|DriftedTable { name, drift }| {
1333                let revision = self.tables.get(name).map(|table| table.schema().revision);
1334                let open = revision.map_or_else(|| "not open".to_string(), |at| format!("revision {at}"));
1335                format!(
1336                    "\"{name}\" is open here at {open} while the catalog {}",
1337                    stored_side_of(drift)
1338                )
1339            })
1340            .collect::<Vec<_>>()
1341            .join("; ");
1342        HaysonSchemaError(format!(
1343            "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",
1344            self.layout.root().display()
1345        ))
1346    }
1347
1348    /// The relations the catalog declares.
1349    ///
1350    /// @param table When given, keeps only the relations touching this table, on either end.
1351    pub fn relations(&self, table: Option<&str>) -> Vec<Relation> {
1352        relations_touching(&self.catalog, table)
1353    }
1354
1355    /// Walks every declared relation and reports the records whose reference has no counterpart —
1356    /// a diagnosis, not a gate; nothing is repaired, removed or written.
1357    pub fn check_integrity(&self) -> io::Result<IntegrityReport> {
1358        let refs: HashMap<String, &Table> = self.tables.iter().map(|(name, table)| (name.clone(), table)).collect();
1359        check_integrity(&self.catalog, &refs)
1360    }
1361
1362    /// Catches up with everything other processes did to this directory since this handle last
1363    /// looked — the port of the TS `Database.refresh()`, in the same two steps and the same order:
1364    ///
1365    /// 1. `hayson.json` is read again, and every table it now declares that this handle does not
1366    ///    hold is opened, through the very code path `open` uses per table;
1367    /// 2. every table already open here is caught up through [`Table::refresh`].
1368    ///
1369    /// Answers whether anything at all changed. A catalog that lost a table on disk does not close
1370    /// the one already open here: a live table is not dropped out from under a caller holding it.
1371    pub fn refresh(&mut self) -> Result<bool, DatabaseError> {
1372        let mut changed = false;
1373        if let Some(catalog) = read_catalog(self.layout.metadata_path())? {
1374            if catalog != self.catalog {
1375                self.catalog = catalog;
1376                changed = true;
1377            }
1378        }
1379
1380        let missing: Vec<String> = self
1381            .catalog
1382            .tables
1383            .keys()
1384            .filter(|name| !self.tables.contains_key(*name))
1385            .cloned()
1386            .collect();
1387        for name in missing {
1388            let Some(entry) = self.catalog.tables.get(&name) else {
1389                continue;
1390            };
1391            let (table, outcome) = open_catalog_table(
1392                &self.layout,
1393                &name,
1394                &self.catalog,
1395                entry,
1396                self.snapshot_every,
1397                self.cipher.as_ref(),
1398                self.follow,
1399                self.readonly,
1400                compaction_of(self.compaction, &self.catalog, Some(entry)),
1401                self.cache,
1402            )?;
1403            self.tables.insert(name.clone(), table);
1404            self.restore_outcomes.insert(name, outcome);
1405            changed = true;
1406        }
1407
1408        for (name, table) in self.tables.iter_mut() {
1409            let outcome = table
1410                .refresh()
1411                .map_err(|error| DatabaseError::Table(name.clone(), TableError::Io(error)))?;
1412            changed |= outcome.changed();
1413        }
1414
1415        // NOTE: triggers are resynced after the catalog re-read — a table opened above has none yet.
1416        sync_table_triggers(&self.catalog, &mut self.tables);
1417        Ok(changed)
1418    }
1419
1420    /// Starts a unit of work: stage `put`/`patch`/`delete` across several tables, then pass it to
1421    /// [`Database::commit`], which lands it whole or not at all.
1422    pub fn unit_of_work(&mut self) -> UnitOfWork {
1423        UnitOfWork::new()
1424    }
1425
1426    /// Runs every operation staged in `unit`, across the tables it names, through the six-step
1427    /// commit `txn::UnitOfWork::commit` documents (resolve, barrier, prepare, journal, write,
1428    /// cleanup) — refused when the database is readonly.
1429    ///
1430    /// The unit is borrowed, not consumed: a commit refused before anything was journalled leaves
1431    /// it staged and retriable, exactly as the TS engine leaves it.
1432    pub fn commit(&mut self, unit: &mut UnitOfWork) -> Result<UnitOfWorkResult, DatabaseError> {
1433        let prepared = self.prepare_commit(unit)?;
1434        unit.journal(&prepared, &self.layout.journal_path())?;
1435        self.apply_commit(prepared)
1436    }
1437
1438    /// The first phase of [`Database::commit`] — the readonly refusal, the lock renewal and
1439    /// [`UnitOfWork::prepare`]. Nothing is appended and no file is written, so a refusal from here
1440    /// leaves `unit` staged and retriable exactly as a refused `commit` does.
1441    ///
1442    /// Split out for a caller that keeps the database behind a lock and wants to release it across
1443    /// the journal write, which touches no table: prepare here, [`UnitOfWork::journal`] with the
1444    /// database let go, then [`Database::apply_commit`] once it is held again. The caller must stay
1445    /// the only writer across all three, or the burned `_seq` numbers would land out of order.
1446    pub fn prepare_commit(&mut self, unit: &mut UnitOfWork) -> Result<PreparedUnit, DatabaseError> {
1447        if self.readonly {
1448            return Err(DatabaseError::ReadOnly {
1449                dir: self.layout.root().to_path_buf(),
1450                operation: "commit".to_string(),
1451            });
1452        }
1453        self.renew_lock()?;
1454        Ok(unit.prepare(&mut self.tables)?)
1455    }
1456
1457    /// The last phase of [`Database::commit`]: every prepared entry appended and the journal
1458    /// removed. See [`Database::prepare_commit`] for the contract a caller running the phases apart
1459    /// has to keep.
1460    pub fn apply_commit(&mut self, prepared: PreparedUnit) -> Result<UnitOfWorkResult, DatabaseError> {
1461        let journal_path = self.layout.journal_path();
1462        Ok(apply_prepared(prepared, &mut self.tables, &journal_path)?)
1463    }
1464
1465    /// Refuses the operation when this handle is not the writer, and renews the lease while it is
1466    /// — so a lock stolen out from under a live handle stops the next write instead of letting it
1467    /// through into a directory somebody else now owns.
1468    fn assert_writable(&self, operation: &str) -> Result<(), DatabaseError> {
1469        if self.readonly {
1470            return Err(DatabaseError::ReadOnly {
1471                dir: self.layout.root().to_path_buf(),
1472                operation: operation.to_string(),
1473            });
1474        }
1475        self.renew_lock()
1476    }
1477
1478    fn write_catalog_to_disk(&self, catalog: &Catalog) -> Result<(), DatabaseError> {
1479        write_catalog(self.layout.metadata_path(), catalog).map_err(catalog_write_error)
1480    }
1481
1482    /// Creates a table, or brings an existing one up to a newer (additive) schema — `catalog::build`
1483    /// does not check revision compatibility yet (see its own doc comment), so a genuinely breaking
1484    /// change is not refused here either; that check is still on the TODO list this method inherits.
1485    ///
1486    /// Disk first: the catalog is written before `self.catalog`/`self.tables` change, so a failure
1487    /// past that point is the only way this call can leave the two out of step with each other.
1488    ///
1489    /// The handle comes from `open_catalog_table` — the same helper `open()`/`refresh()` use — so a
1490    /// table migrated here carries exactly what it would carry after a restart, triggers included.
1491    ///
1492    /// `name` is the key the table lands under in the catalog, and the schema's own `table` is
1493    /// carried into every log line and index snapshot written for it. The two disagreeing is
1494    /// refused rather than resolved: the catalog key wins wherever they are read back, so accepting
1495    /// the spec would silently drop the name its author wrote.
1496    pub fn create_table(&mut self, name: &str, schema_spec: Value) -> Result<(), DatabaseError> {
1497        self.assert_writable("createTable")?;
1498        let schema = parse_schema(&schema_spec)?;
1499        if schema.table != name {
1500            return Err(DatabaseError::Schema(HaysonSchemaError(format!(
1501                "schema declares table \"{}\" but it is being created as \"{name}\" — name them the same",
1502                schema.table
1503            ))));
1504        }
1505        let now = now_ms();
1506        let next_catalog = crate::catalog::build::create_table(&self.catalog, name, &schema, now);
1507        self.write_catalog_to_disk(&next_catalog)?;
1508
1509        let next_entry = &next_catalog.tables[name];
1510        let (table, outcome) = open_catalog_table(
1511            &self.layout,
1512            name,
1513            &next_catalog,
1514            next_entry,
1515            self.snapshot_every,
1516            self.cipher.as_ref(),
1517            self.follow,
1518            self.readonly,
1519            compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1520            self.cache,
1521        )?;
1522
1523        self.catalog = next_catalog;
1524        self.tables.insert(name.to_string(), table);
1525        self.restore_outcomes.insert(name.to_string(), outcome);
1526        Ok(())
1527    }
1528
1529    /// Takes a field off the schema of a table without touching its log — records written before
1530    /// the drop keep carrying the field on disk, reads just stop handing it back.
1531    ///
1532    /// Refuses (leaving disk and `self` untouched) when the field is the primary key, is what the
1533    /// table partitions by, or an index/check/relation of the catalog still depends on it
1534    /// (`catalog::build::drop_field`, unconditional on `force`) — and, unless `force` is `true`,
1535    /// when a live record still holds a value for it (checked here, against the open table, per the
1536    /// TODO `catalog::build::drop_field` leaves for this layer).
1537    ///
1538    /// The declaration check runs first on purpose: with both refusals due the caller is told about
1539    /// the one `force` cannot get them past, and the error class matches what the field's
1540    /// dependents are — see `packages/hayson/test/drop-pola-kolejnosc-odmowy.test.ts`.
1541    pub fn drop_field(&mut self, table: &str, field: &str, force: bool) -> Result<(), DatabaseError> {
1542        self.assert_writable("dropField")?;
1543        self.flush_table(table)?;
1544        let existing = self.tables.get(table).ok_or_else(|| {
1545            DatabaseError::Catalog(format!(
1546                "table \"{table}\" does not exist in \"{}\"",
1547                self.layout.root().display()
1548            ))
1549        })?;
1550
1551        let now = now_ms();
1552        let (next_catalog, next_entry) = crate::catalog::build::drop_field(&self.catalog, table, field, now)?;
1553
1554        if !force {
1555            assert_no_residual_values(table, existing, field)?;
1556        }
1557
1558        self.write_catalog_to_disk(&next_catalog)?;
1559
1560        let (mut table_handle, _) = open_catalog_table(
1561            &self.layout,
1562            table,
1563            &next_catalog,
1564            &next_entry,
1565            self.snapshot_every,
1566            self.cipher.as_ref(),
1567            self.follow,
1568            self.readonly,
1569            compaction_of(self.compaction, &next_catalog, Some(&next_entry)),
1570            self.cache,
1571        )?;
1572        // NOTE: the .idx on disk still projects the dropped field — rewrite it before anyone restores from it.
1573        table_handle
1574            .snapshot()
1575            .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1576
1577        self.catalog = next_catalog;
1578        self.tables.insert(table.to_string(), table_handle);
1579        Ok(())
1580    }
1581
1582    /// Adds a named index to a table of the catalog — a unique index is refused whole if a live
1583    /// record already shares a value on it (`catalog::declare_index`, checked against the open
1584    /// table). The reopened table's registry is built from the catalog's full expanded index list
1585    /// (`catalog::expand_indexes`), so a composite or `where`-scoped index is enforced/served right
1586    /// away, same as the single-field unconditional case `catalog::expand_indexes` derives as sugar.
1587    ///
1588    /// Reopened through `open_catalog_table` — the same helper `open()`/`refresh()` use — so the
1589    /// handle keeps its snapshot pairing, and the stale `.idx` is rewritten before this returns.
1590    pub fn declare_index(&mut self, table: &str, index_spec: Value) -> Result<(), DatabaseError> {
1591        self.assert_writable("declareIndex")?;
1592        self.flush_table(table)?;
1593        let existing = self.tables.get(table).ok_or_else(|| {
1594            DatabaseError::Catalog(format!(
1595                "table \"{table}\" does not exist in \"{}\"",
1596                self.layout.root().display()
1597            ))
1598        })?;
1599
1600        let now = now_ms();
1601        let next_catalog = crate::catalog::declare_index(&self.catalog, table, &index_spec, existing, now)?;
1602        self.write_catalog_to_disk(&next_catalog)?;
1603
1604        let next_entry = &next_catalog.tables[table];
1605        let (mut table_handle, _) = open_catalog_table(
1606            &self.layout,
1607            table,
1608            &next_catalog,
1609            next_entry,
1610            self.snapshot_every,
1611            self.cipher.as_ref(),
1612            self.follow,
1613            self.readonly,
1614            compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1615            self.cache,
1616        )?;
1617        // NOTE: the .idx on disk projects nothing of the new index — rewrite it before anyone restores from it.
1618        table_handle
1619            .snapshot()
1620            .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1621
1622        self.catalog = next_catalog;
1623        self.tables.insert(table.to_string(), table_handle);
1624        Ok(())
1625    }
1626
1627    /// Adds a named relation to the catalog (`catalog::declare_relation`). With `enforce: true`,
1628    /// refuses whole if a live record already breaks it — unless `force` is `true`, which declares
1629    /// it anyway and lets enforcement start from here on, existing orphans left as they are.
1630    ///
1631    /// Unlike `declare_index`, nothing here reopens a table: a relation is not read by a table's own
1632    /// registry the way an index is, so no live handle changes shape. `from`/`to` are read straight
1633    /// out of `relation_spec` first, ahead of the real parse inside `catalog::declare_relation`, only
1634    /// to know which two tables to flush and hand it — a malformed spec still gets the full,
1635    /// authoritative error from the parser, this is not validation.
1636    pub fn declare_relation(&mut self, relation_spec: Value, force: bool) -> Result<(), DatabaseError> {
1637        self.assert_writable("declareRelation")?;
1638
1639        let from_table = relation_spec
1640            .get("from")
1641            .and_then(|end| end.get("table"))
1642            .and_then(Value::as_str)
1643            .ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"from.table\"".to_string()))?
1644            .to_string();
1645        let to_table = relation_spec
1646            .get("to")
1647            .and_then(|end| end.get("table"))
1648            .and_then(Value::as_str)
1649            .ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"to.table\"".to_string()))?
1650            .to_string();
1651
1652        self.flush_table(&from_table)?;
1653        self.flush_table(&to_table)?;
1654        let from_handle = self.tables.get(&from_table).ok_or_else(|| {
1655            DatabaseError::Catalog(format!(
1656                "table \"{from_table}\" does not exist in \"{}\"",
1657                self.layout.root().display()
1658            ))
1659        })?;
1660        let to_handle = self.tables.get(&to_table).ok_or_else(|| {
1661            DatabaseError::Catalog(format!(
1662                "table \"{to_table}\" does not exist in \"{}\"",
1663                self.layout.root().display()
1664            ))
1665        })?;
1666
1667        let now = now_ms();
1668        let next_catalog = declare_relation_in(&self.catalog, &relation_spec, from_handle, to_handle, now, force)?;
1669        self.write_catalog_to_disk(&next_catalog)?;
1670        self.catalog = next_catalog;
1671        Ok(())
1672    }
1673
1674    /// Declares an AFTER trigger (`catalog::build::declare_trigger`, which refuses a spec naming a
1675    /// table the catalog does not declare) and refreshes every open table's live trigger set from
1676    /// the result — both the source table `trigger_spec.table` names and the effect table
1677    /// `trigger_spec.effect.table` names, since either could be a live table this call did not
1678    /// itself just open. Disk first, same discipline as `create_table`/`declare_index`.
1679    pub fn declare_trigger(&mut self, trigger_spec: Value) -> Result<(), DatabaseError> {
1680        self.assert_writable("declareTrigger")?;
1681
1682        let now = now_ms();
1683        let next_catalog = crate::catalog::declare_trigger(&self.catalog, &trigger_spec, now)?;
1684        self.write_catalog_to_disk(&next_catalog)?;
1685
1686        self.catalog = next_catalog;
1687        sync_table_triggers(&self.catalog, &mut self.tables);
1688        Ok(())
1689    }
1690
1691    /// Appends every write staged against any table of this database and not yet committed — the
1692    /// TS `Database.flush()`. A no-op for a table with nothing in flight, which is every table of a
1693    /// database whose writes all went through `put`/`patch`/`delete`.
1694    pub fn flush(&mut self) -> Result<(), DatabaseError> {
1695        for (name, table) in &mut self.tables {
1696            table
1697                .flush()
1698                .map_err(|error| DatabaseError::Table(name.clone(), error))?;
1699        }
1700        Ok(())
1701    }
1702
1703    /// Commits what is staged against one table. Every path that reopens a table over its log, or
1704    /// rewrites that log, has to run this first: the handle it replaces takes its staging window
1705    /// with it.
1706    fn flush_table(&mut self, table: &str) -> Result<(), DatabaseError> {
1707        let Some(handle) = self.tables.get_mut(table) else {
1708            return Ok(());
1709        };
1710        handle
1711            .flush()
1712            .map_err(|error| DatabaseError::Table(table.to_string(), error))
1713    }
1714
1715    /// Appends whatever every table still holds staged, snapshots the ones with writes since their
1716    /// last snapshot (see [`Table::needs_snapshot`]), then releases the writer lock, if this handle
1717    /// holds one — mirrors the TS `Database.close()`. The catalog needs no flush of its own: it is
1718    /// already durable (`fsync`) by the time the call that wrote it returns.
1719    ///
1720    /// The lock is released whether or not a table refused its flush or snapshot: the handle is
1721    /// consumed either way, and a lock file left behind names a live pid that no staleness rule
1722    /// reclaims, so keeping it would lock this very process out of its own directory. The first
1723    /// refusal is what the caller gets, after every other table had its turn.
1724    pub fn close(mut self) -> io::Result<()> {
1725        let flushed = if self.readonly {
1726            Ok(())
1727        } else {
1728            self.flush_and_snapshot_all()
1729        };
1730        if let Some(mut lock) = self.lock.take() {
1731            lock.release();
1732        }
1733        flushed
1734    }
1735
1736    fn flush_and_snapshot_all(&mut self) -> io::Result<()> {
1737        let mut first_refusal = None;
1738        for table in self.tables.values_mut() {
1739            // NOTE: flush before the snapshot — a staged write is in neither the log nor it.
1740            let outcome = table.flush().map_err(io::Error::from).and_then(|()| {
1741                if table.needs_snapshot() {
1742                    table.snapshot()
1743                } else {
1744                    Ok(())
1745                }
1746            });
1747            if let Err(error) = outcome {
1748                first_refusal.get_or_insert(error);
1749            }
1750        }
1751        match first_refusal {
1752            Some(error) => Err(error),
1753            None => Ok(()),
1754        }
1755    }
1756}
1757
1758#[cfg(test)]
1759mod tests {
1760    use serde_json::json;
1761    use tempfile::tempdir;
1762
1763    use super::*;
1764    use crate::catalog::create_table;
1765    use crate::catalog::types::{CatalogIndexKind, IndexEntry};
1766    use crate::checkpoint::write_snapshot;
1767    use crate::registry::table_registry::TableRegistry;
1768    use crate::schema::parse_schema;
1769    use crate::storage::{serialize_entry, LogMeta, LogOp};
1770    use crate::txn::journal::{write_journal, Journal, JournalEntry, JournalHeader, JOURNAL_VERSION};
1771    use crate::txn::JOURNAL_FILE;
1772
1773    const MANIFEST_FILE: &str = DEFAULT_METADATA_FILE;
1774
1775    fn flat(dir: &Path) -> Layout {
1776        Layout::new(dir, crate::database::LayoutKind::Flat, MANIFEST_FILE)
1777    }
1778
1779    fn log_path(dir: &Path, table: &str) -> std::path::PathBuf {
1780        flat(dir).log_path(table)
1781    }
1782
1783    /// The binding above this crate keeps the database behind an `RwLock` shared across the libuv
1784    /// pool, which needs `Sync` as well as `Send`. Nothing here runs: the call fails to compile the
1785    /// moment an interior-mutability cell that is not thread-safe reappears anywhere below.
1786    #[test]
1787    fn a_database_and_its_tables_can_be_shared_across_threads() {
1788        fn shareable<T: Send + Sync>() {}
1789        shareable::<Database>();
1790        shareable::<Table>();
1791        shareable::<crate::registry::fulltext_index::FulltextIndex>();
1792    }
1793
1794    fn index_path(dir: &Path, table: &str) -> std::path::PathBuf {
1795        flat(dir).index_path(table)
1796    }
1797
1798    fn users_schema() -> TableSchema {
1799        parse_schema(&json!({
1800            "schemaVersion": 1,
1801            "table": "users",
1802            "primaryKey": "id",
1803            "fields": {
1804                "id": { "type": "string" },
1805                "name": { "type": "string", "required": true },
1806            },
1807        }))
1808        .unwrap()
1809    }
1810
1811    fn orders_schema() -> TableSchema {
1812        parse_schema(&json!({
1813            "schemaVersion": 1,
1814            "table": "orders",
1815            "primaryKey": "id",
1816            "fields": {
1817                "id": { "type": "string" },
1818                "userId": { "type": "string", "required": true },
1819            },
1820        }))
1821        .unwrap()
1822    }
1823
1824    fn catalog_of(schemas: &[TableSchema]) -> Catalog {
1825        let mut catalog = empty_catalog(CatalogLayout::Flat, 1_000);
1826        for schema in schemas {
1827            catalog = create_table(&catalog, &schema.table, schema, 1_000);
1828        }
1829        catalog
1830    }
1831
1832    fn write_catalog_to(dir: &Path, catalog: &Catalog) {
1833        write_catalog(&dir.join(MANIFEST_FILE), catalog).unwrap();
1834    }
1835
1836    #[test]
1837    fn open_builds_a_database_with_every_declared_table_available() {
1838        let dir = tempdir().unwrap();
1839        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1840
1841        let mut db = Database::open(dir.path()).unwrap();
1842        assert_eq!(db.table_names(), vec!["orders", "users"]);
1843
1844        db.table_mut("users")
1845            .unwrap()
1846            .put(&json!({ "id": "u1", "name": "Ania" }), false)
1847            .unwrap();
1848        assert_eq!(
1849            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1850            Some(&json!("Ania"))
1851        );
1852        db.close().unwrap();
1853    }
1854
1855    #[test]
1856    fn open_restores_the_registry_from_a_paired_snapshot_instead_of_scanning_the_log() {
1857        let dir = tempdir().unwrap();
1858        let catalog = catalog_of(&[users_schema()]);
1859        write_catalog_to(dir.path(), &catalog);
1860
1861        let schema = table_schema_of("users", &catalog.tables["users"]);
1862        let hash = schema_fingerprint(&schema, &expand_indexes("users", &catalog.tables["users"]));
1863        let log_path = log_path(dir.path(), "users");
1864        let mut log = crate::storage::LogFile::open(&log_path, false).unwrap();
1865        log.append(&[serialize_entry(
1866            &LogMeta {
1867                key: "u1".into(),
1868                op: LogOp::Put,
1869                seq: 1,
1870                ts: 1,
1871            },
1872            Some(&{
1873                let mut record = crate::schema::HaysonRecord::new();
1874                record.insert("id".into(), json!("u1"));
1875                record.insert("name".into(), json!("Ania"));
1876                record
1877            }),
1878        )])
1879        .unwrap();
1880        let registry = TableRegistry::build_from_log(schema.clone(), &log).unwrap();
1881        write_snapshot(&index_path(dir.path(), "users"), &registry, "users", &hash, 0, &log).unwrap();
1882
1883        let db = Database::open(dir.path()).unwrap();
1884        assert_eq!(db.restore_outcome("users"), Some(RestoreOutcome::Restored));
1885        assert_eq!(
1886            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1887            Some(&json!("Ania"))
1888        );
1889        db.close().unwrap();
1890    }
1891
1892    #[test]
1893    fn open_finishes_an_unfinished_journal_before_returning() {
1894        let dir = tempdir().unwrap();
1895        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1896
1897        {
1898            let mut users = Table::open(users_schema(), log_path(dir.path(), "users")).unwrap();
1899            let mut orders = Table::open(orders_schema(), log_path(dir.path(), "orders")).unwrap();
1900            let user_entries = users
1901                .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
1902                .unwrap();
1903            let order_entries = orders
1904                .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "o1", "userId": "u1" }))])
1905                .unwrap();
1906
1907            let mut entries = Vec::new();
1908            for entry in &user_entries {
1909                entries.push(JournalEntry {
1910                    t: "users".into(),
1911                    s: entry.seq(),
1912                    l: entry.line().to_string(),
1913                });
1914            }
1915            for entry in &order_entries {
1916                entries.push(JournalEntry {
1917                    t: "orders".into(),
1918                    s: entry.seq(),
1919                    l: entry.line().to_string(),
1920                });
1921            }
1922            write_journal(
1923                &dir.path().join(JOURNAL_FILE),
1924                &Journal {
1925                    header: JournalHeader {
1926                        journal_version: JOURNAL_VERSION,
1927                        kind: "unit".into(),
1928                        id: "crash".into(),
1929                        started_at: 0,
1930                        tables: vec!["users".into(), "orders".into()],
1931                    },
1932                    entries,
1933                },
1934            )
1935            .unwrap();
1936            // process dies here: prepared and journalled, never applied to either log.
1937        }
1938
1939        assert!(dir.path().join(JOURNAL_FILE).exists());
1940        let db = Database::open(dir.path()).unwrap();
1941        assert!(!dir.path().join(JOURNAL_FILE).exists());
1942        assert_eq!(
1943            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1944            Some(&json!("Ania"))
1945        );
1946        assert_eq!(
1947            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
1948            Some(&json!("u1"))
1949        );
1950        db.close().unwrap();
1951    }
1952
1953    #[test]
1954    fn a_second_open_of_the_same_directory_is_denied_while_the_first_still_holds_the_lock() {
1955        let dir = tempdir().unwrap();
1956        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1957
1958        let _first = Database::open(dir.path()).unwrap();
1959        let second = Database::open_with(
1960            dir.path(),
1961            OpenOptions {
1962                readonly: false,
1963                lock: LockMode::Enabled(LockOptions {
1964                    timeout_ms: Some(0),
1965                    ..Default::default()
1966                }),
1967                ..Default::default()
1968            },
1969        );
1970        assert!(matches!(second, Err(DatabaseError::Lock(_))));
1971    }
1972
1973    fn lock_path_of(dir: &Path) -> std::path::PathBuf {
1974        dir.join(crate::lock::LOCK_FILE)
1975    }
1976
1977    fn open_refusing(dir: &Path) -> DatabaseError {
1978        Database::open_with(
1979            dir,
1980            OpenOptions {
1981                lock: LockMode::Enabled(LockOptions {
1982                    timeout_ms: Some(0),
1983                    ..Default::default()
1984                }),
1985                ..Default::default()
1986            },
1987        )
1988        .map(|_| ())
1989        .expect_err("this directory must not open")
1990    }
1991
1992    #[test]
1993    fn a_directory_with_logs_but_no_catalog_is_refused_instead_of_opening_empty() {
1994        let dir = tempdir().unwrap();
1995        let mut db = Database::open(dir.path()).unwrap();
1996        db.create_table(
1997            "users",
1998            json!({
1999                "schemaVersion": 1,
2000                "table": "users",
2001                "primaryKey": "id",
2002                "fields": {
2003                    "id": { "type": "string" },
2004                    "name": { "type": "string", "required": true },
2005                },
2006            }),
2007        )
2008        .unwrap();
2009        db.table_mut("users")
2010            .unwrap()
2011            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2012            .unwrap();
2013        db.close().unwrap();
2014
2015        let log = fs::read(dir.path().join("users.jsonl")).unwrap();
2016        fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
2017
2018        let refusal = open_refusing(dir.path()).to_string();
2019        assert!(
2020            refusal.contains("users.jsonl"),
2021            "the refusal has to name the log it found: {refusal}"
2022        );
2023        assert!(
2024            refusal.contains(MANIFEST_FILE),
2025            "the refusal has to name the catalog: {refusal}"
2026        );
2027
2028        // The refusal is worth nothing if it already wrote a fresh catalog over the data it refused.
2029        assert!(
2030            !dir.path().join(MANIFEST_FILE).exists(),
2031            "a refused open must not leave a catalog"
2032        );
2033        assert_eq!(
2034            fs::read(dir.path().join("users.jsonl")).unwrap(),
2035            log,
2036            "the log must be untouched"
2037        );
2038    }
2039
2040    #[test]
2041    fn an_empty_directory_and_one_holding_no_logs_still_open_as_a_new_database() {
2042        let empty = tempdir().unwrap();
2043        assert!(
2044            Database::open(empty.path()).is_ok(),
2045            "an empty directory is still a new database"
2046        );
2047
2048        let littered = tempdir().unwrap();
2049        fs::write(littered.path().join("README.md"), "# not a table\n").unwrap();
2050        fs::create_dir(littered.path().join("users-backup")).unwrap();
2051        assert!(Database::open(littered.path()).is_ok(), "nothing here is a table log");
2052    }
2053
2054    #[test]
2055    fn the_logs_of_a_partitioned_table_are_found_one_level_down_too() {
2056        let dir = tempdir().unwrap();
2057        fs::create_dir_all(dir.path().join("notes")).unwrap();
2058        fs::write(dir.path().join("notes").join("note.jsonl"), "").unwrap();
2059
2060        let refusal = open_refusing(dir.path()).to_string();
2061        assert!(
2062            refusal.contains("note.jsonl"),
2063            "a partitioned log still counts as data: {refusal}"
2064        );
2065    }
2066
2067    /// Every way the catalog read under the lock can refuse. The lock is taken before any of them
2068    /// runs (`stored_kind_of` is deliberately lenient), so each one is a chance to strand the lock
2069    /// file on a process that is still alive — which no staleness rule would ever reclaim.
2070    #[test]
2071    fn a_directory_refused_at_open_is_left_unlocked_whichever_way_the_catalog_was_broken() {
2072        for (case, document) in [
2073            ("not JSON at all", "{ this is not json".to_string()),
2074            (
2075                "no formatVersion",
2076                json!({ "catalogVersion": 2, "tables": {} }).to_string(),
2077            ),
2078            (
2079                "the format-1 marker",
2080                json!({ "hayson": 1, "formatVersion": 2, "tables": {} }).to_string(),
2081            ),
2082            (
2083                "a formatVersion from the future",
2084                json!({ "catalogVersion": 2, "formatVersion": 3, "tables": {} }).to_string(),
2085            ),
2086        ] {
2087            let dir = tempdir().unwrap();
2088            std::fs::write(dir.path().join(MANIFEST_FILE), &document).unwrap();
2089
2090            open_refusing(dir.path());
2091
2092            assert!(
2093                !lock_path_of(dir.path()).exists(),
2094                "{case}: the refused open left its lock file behind"
2095            );
2096        }
2097    }
2098
2099    /// The point of releasing on refusal: the directory has to stay openable. Nothing else reclaims
2100    /// such a lock — its owner is the very process that got the refusal, and it is still running.
2101    #[test]
2102    fn a_directory_opens_normally_after_an_open_that_was_refused_a_broken_catalog() {
2103        let dir = tempdir().unwrap();
2104        std::fs::write(dir.path().join(MANIFEST_FILE), "{ this is not json").unwrap();
2105
2106        open_refusing(dir.path());
2107
2108        std::fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
2109        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2110
2111        let mut db = Database::open(dir.path()).unwrap();
2112        db.table_mut("users")
2113            .unwrap()
2114            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2115            .unwrap();
2116        db.close().unwrap();
2117    }
2118
2119    /// A refusal raised past the catalog — the cipher is built after it parsed — has to unwind the
2120    /// same way, which is what putting the lock behind a guard instead of one `release` call buys.
2121    #[test]
2122    fn a_directory_refused_for_a_bad_cipher_key_is_left_unlocked_too() {
2123        let dir = tempdir().unwrap();
2124        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2125
2126        let error = Database::open_with(
2127            dir.path(),
2128            OpenOptions {
2129                cypher: Some(CipherOptions {
2130                    keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
2131                    key_id: None,
2132                    on_missing_key: None,
2133                }),
2134                ..OpenOptions::default()
2135            },
2136        )
2137        .map(|_| ())
2138        .unwrap_err();
2139
2140        assert!(matches!(error, DatabaseError::Cipher(_)));
2141        assert!(!lock_path_of(dir.path()).exists());
2142        Database::open(dir.path()).unwrap().close().unwrap();
2143    }
2144
2145    /// The other half of the guard: a successful open must still be holding the lock, not have
2146    /// released it on the way out of `open_locked`.
2147    #[test]
2148    fn an_open_that_succeeded_still_holds_the_lock_until_the_handle_closes() {
2149        let dir = tempdir().unwrap();
2150        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2151
2152        let held = Database::open(dir.path()).unwrap();
2153        assert!(lock_path_of(dir.path()).exists());
2154        assert!(matches!(open_refusing(dir.path()), DatabaseError::Lock(_)));
2155
2156        held.close().unwrap();
2157        assert!(!lock_path_of(dir.path()).exists());
2158    }
2159
2160    /// A second writer of a directory somebody else already holds — the port of the TS
2161    /// `open(dir, { lock: false })`, deliberately dangerous and only ever reached on request.
2162    fn open_skipping_the_lock(dir: &Path) -> Database {
2163        Database::open_with(
2164            dir,
2165            OpenOptions {
2166                lock: LockMode::Skip,
2167                ..Default::default()
2168            },
2169        )
2170        .unwrap()
2171    }
2172
2173    fn declared_on_disk(dir: &Path, table: &str) -> Vec<String> {
2174        let catalog = read_catalog(&dir.join(MANIFEST_FILE))
2175            .unwrap()
2176            .expect("a catalog on disk");
2177        let mut names: Vec<String> = catalog.tables[table]
2178            .indexes
2179            .iter()
2180            .flatten()
2181            .map(|index| index.name.clone())
2182            .collect();
2183        names.sort();
2184        names
2185    }
2186
2187    /// The fixture the three lock-skipping tests share: a directory with two tables, two records,
2188    /// and no live handle — the `usersWithTwoRecords` of the TS
2189    /// `deklarowanie-indeksu-wspolbieznie.test.ts`.
2190    fn two_tables_with_records(dir: &Path) {
2191        write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
2192        let mut db = Database::open(dir).unwrap();
2193        db.table_mut("users")
2194            .unwrap()
2195            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2196            .unwrap();
2197        db.table_mut("users")
2198            .unwrap()
2199            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
2200            .unwrap();
2201        db.close().unwrap();
2202    }
2203
2204    #[test]
2205    fn a_second_writer_that_skips_the_lock_opens_and_writes_beside_the_one_holding_it() {
2206        let dir = tempdir().unwrap();
2207        two_tables_with_records(dir.path());
2208
2209        let mut locked = Database::open(dir.path()).unwrap();
2210        let mut unlocked = open_skipping_the_lock(dir.path());
2211
2212        // Both are writers, on their own table each — the lock never came into it.
2213        locked
2214            .table_mut("users")
2215            .unwrap()
2216            .put(&json!({ "id": "u3", "name": "Kasia" }), false)
2217            .unwrap();
2218        unlocked
2219            .table_mut("orders")
2220            .unwrap()
2221            .put(&json!({ "id": "o1", "userId": "u1" }), false)
2222            .unwrap();
2223
2224        // While a third handle asking for the lock is still refused, so nothing was released.
2225        let third = Database::open_with(
2226            dir.path(),
2227            OpenOptions {
2228                lock: LockMode::Enabled(LockOptions {
2229                    timeout_ms: Some(0),
2230                    ..Default::default()
2231                }),
2232                ..Default::default()
2233            },
2234        );
2235        assert!(matches!(third, Err(DatabaseError::Lock(_))));
2236
2237        unlocked.close().unwrap();
2238        locked.close().unwrap();
2239    }
2240
2241    /// The window `declare_index` documents: it writes the catalog it holds in memory, with no
2242    /// check against the disk, and the writer lock is the only guard. A handle that skipped the
2243    /// lock therefore overwrites what the other one declared. This is the TS behaviour, asserted
2244    /// so a change to it is a decision and not an accident.
2245    #[test]
2246    fn a_lock_skipping_handle_that_never_refreshed_loses_the_declaration_of_the_locked_one() {
2247        let dir = tempdir().unwrap();
2248        two_tables_with_records(dir.path());
2249
2250        let mut locked = Database::open(dir.path()).unwrap();
2251        let mut unlocked = open_skipping_the_lock(dir.path());
2252
2253        locked
2254            .declare_index(
2255                "users",
2256                json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
2257            )
2258            .unwrap();
2259        unlocked
2260            .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
2261            .unwrap();
2262
2263        unlocked.close().unwrap();
2264        locked.close().unwrap();
2265
2266        assert_eq!(declared_on_disk(dir.path(), "users"), vec!["by_id".to_string()]);
2267    }
2268
2269    #[test]
2270    fn a_lock_skipping_handle_that_refreshes_first_keeps_both_declarations() {
2271        let dir = tempdir().unwrap();
2272        two_tables_with_records(dir.path());
2273
2274        let mut locked = Database::open(dir.path()).unwrap();
2275        let mut unlocked = open_skipping_the_lock(dir.path());
2276
2277        locked
2278            .declare_index(
2279                "users",
2280                json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
2281            )
2282            .unwrap();
2283        assert!(unlocked.refresh().unwrap(), "the catalog changed under it");
2284        unlocked
2285            .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
2286            .unwrap();
2287
2288        unlocked.close().unwrap();
2289        locked.close().unwrap();
2290
2291        assert_eq!(
2292            declared_on_disk(dir.path(), "users"),
2293            vec!["by_id".to_string(), "by_name".to_string()]
2294        );
2295    }
2296
2297    /// A reader of the same directory as the live writer: readonly, so it takes no lock and the
2298    /// writer keeps its own — the shape a follower process has.
2299    fn open_follower(dir: &Path) -> Database {
2300        Database::open_with(
2301            dir,
2302            OpenOptions {
2303                readonly: true,
2304                lock: LockMode::default(),
2305                ..Default::default()
2306            },
2307        )
2308        .unwrap()
2309    }
2310
2311    #[test]
2312    fn refresh_with_nothing_new_on_disk_reports_no_change() {
2313        let dir = tempdir().unwrap();
2314        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2315        let mut writer = Database::open(dir.path()).unwrap();
2316        writer
2317            .table_mut("users")
2318            .unwrap()
2319            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2320            .unwrap();
2321        let mut follower = open_follower(dir.path());
2322        follower.refresh().unwrap();
2323
2324        assert!(!follower.refresh().unwrap());
2325        writer.close().unwrap();
2326    }
2327
2328    #[test]
2329    fn refresh_picks_up_both_a_table_and_the_records_another_process_wrote() {
2330        let dir = tempdir().unwrap();
2331        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2332        let mut writer = Database::open(dir.path()).unwrap();
2333        let mut follower = open_follower(dir.path());
2334        assert_eq!(follower.table_names(), vec!["users"]);
2335
2336        writer
2337            .table_mut("users")
2338            .unwrap()
2339            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2340            .unwrap();
2341        writer
2342            .create_table(
2343                "orders",
2344                json!({
2345                    "schemaVersion": 1,
2346                    "table": "orders",
2347                    "primaryKey": "id",
2348                    "fields": { "id": { "type": "string" }, "userId": { "type": "string", "required": true } },
2349                }),
2350            )
2351            .unwrap();
2352        writer
2353            .table_mut("orders")
2354            .unwrap()
2355            .put(&json!({ "id": "o1", "userId": "u1" }), false)
2356            .unwrap();
2357
2358        assert!(follower.refresh().unwrap());
2359
2360        assert_eq!(follower.table_names(), vec!["orders", "users"]);
2361        assert_eq!(
2362            follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2363            Some(&json!("Ania"))
2364        );
2365        assert_eq!(
2366            follower
2367                .table("orders")
2368                .unwrap()
2369                .get("o1")
2370                .unwrap()
2371                .unwrap()
2372                .get("userId"),
2373            Some(&json!("u1"))
2374        );
2375        writer.close().unwrap();
2376    }
2377
2378    #[test]
2379    fn refresh_survives_a_compaction_run_by_the_writer_process() {
2380        let dir = tempdir().unwrap();
2381        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2382        let mut writer = Database::open(dir.path()).unwrap();
2383        let users = writer.table_mut("users").unwrap();
2384        users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
2385        users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
2386        users.patch("u1", &json!({ "name": "Anna" }), false).unwrap();
2387        let mut follower = open_follower(dir.path());
2388        assert_eq!(follower.table("users").unwrap().len(), 2);
2389
2390        writer.table_mut("users").unwrap().compact().unwrap();
2391        assert!(follower.refresh().unwrap());
2392
2393        assert_eq!(follower.table("users").unwrap().len(), 2);
2394        assert_eq!(
2395            follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2396            Some(&json!("Anna"))
2397        );
2398        assert_eq!(follower.table("users").unwrap().stats().gen, 1);
2399        writer.close().unwrap();
2400    }
2401
2402    #[test]
2403    fn unit_of_work_commits_across_two_tables_and_the_database_sees_both() {
2404        let dir = tempdir().unwrap();
2405        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2406        let mut db = Database::open(dir.path()).unwrap();
2407
2408        let mut unit = db.unit_of_work();
2409        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
2410        unit.put("orders", json!({ "id": "o1", "userId": "u1" }));
2411        let result = db.commit(&mut unit).unwrap();
2412        assert_eq!(result.operations, 2);
2413
2414        assert_eq!(
2415            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2416            Some(&json!("Ania"))
2417        );
2418        assert_eq!(
2419            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2420            Some(&json!("u1"))
2421        );
2422        db.close().unwrap();
2423    }
2424
2425    #[test]
2426    fn create_table_on_an_open_database_adds_a_new_working_table_immediately() {
2427        let dir = tempdir().unwrap();
2428        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2429        let mut db = Database::open(dir.path()).unwrap();
2430
2431        db.create_table(
2432            "orders",
2433            json!({
2434                "schemaVersion": 1,
2435                "table": "orders",
2436                "primaryKey": "id",
2437                "fields": {
2438                    "id": { "type": "string" },
2439                    "userId": { "type": "string", "required": true },
2440                },
2441            }),
2442        )
2443        .unwrap();
2444
2445        assert_eq!(db.table_names(), vec!["orders", "users"]);
2446        assert!(db.catalog.tables.contains_key("orders"));
2447
2448        db.table_mut("orders")
2449            .unwrap()
2450            .put(&json!({ "id": "o1", "userId": "u1" }), false)
2451            .unwrap();
2452        assert_eq!(
2453            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2454            Some(&json!("u1"))
2455        );
2456
2457        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2458        assert!(reread.tables.contains_key("orders"));
2459        db.close().unwrap();
2460    }
2461
2462    #[test]
2463    fn create_table_refuses_a_schema_naming_a_table_other_than_the_one_being_created() {
2464        let dir = tempdir().unwrap();
2465        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2466        let mut db = Database::open(dir.path()).unwrap();
2467        let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2468
2469        let error = db
2470            .create_table(
2471                "orders",
2472                json!({
2473                    "schemaVersion": 1,
2474                    "table": "invoices",
2475                    "primaryKey": "id",
2476                    "fields": { "id": { "type": "string" } },
2477                }),
2478            )
2479            .expect_err("a schema naming another table is refused");
2480
2481        assert!(matches!(error, DatabaseError::Schema(_)), "got {error:?}");
2482        let message = error.to_string();
2483        assert!(
2484            message.contains("invoices"),
2485            "the refusal names the schema's table: {message}"
2486        );
2487        assert!(
2488            message.contains("orders"),
2489            "the refusal names the key it was created under: {message}"
2490        );
2491
2492        assert_eq!(db.table_names(), vec!["users"]);
2493        assert!(!db.catalog.tables.contains_key("orders"));
2494        assert!(!db.catalog.tables.contains_key("invoices"));
2495        assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2496        db.close().unwrap();
2497    }
2498
2499    #[test]
2500    fn create_table_still_takes_a_schema_naming_exactly_the_table_it_creates() {
2501        let dir = tempdir().unwrap();
2502        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2503        let mut db = Database::open(dir.path()).unwrap();
2504
2505        db.create_table(
2506            "orders",
2507            json!({
2508                "schemaVersion": 1,
2509                "table": "orders",
2510                "primaryKey": "id",
2511                "fields": { "id": { "type": "string" } },
2512            }),
2513        )
2514        .unwrap();
2515
2516        assert_eq!(db.table("orders").unwrap().schema().table, "orders");
2517        db.close().unwrap();
2518    }
2519
2520    #[test]
2521    fn create_table_refuses_on_a_readonly_database_and_changes_nothing() {
2522        let dir = tempdir().unwrap();
2523        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2524        // NOTE: a readonly open refuses a catalog whose logs are not on disk yet (F8) — make them.
2525        Database::open(dir.path()).unwrap().close().unwrap();
2526        let mut db = Database::open_with(
2527            dir.path(),
2528            OpenOptions {
2529                readonly: true,
2530                lock: LockMode::default(),
2531                ..Default::default()
2532            },
2533        )
2534        .unwrap();
2535
2536        let error = db
2537            .create_table(
2538                "orders",
2539                json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2540            )
2541            .unwrap_err();
2542        assert!(
2543            matches!(&error, DatabaseError::ReadOnly { dir: refused, operation } if refused == dir.path() && operation == "createTable"),
2544            "expected a typed readonly refusal, got {error:?}"
2545        );
2546        assert_eq!(db.table_names(), vec!["users"]);
2547    }
2548
2549    #[test]
2550    fn create_table_refused_by_a_directory_on_the_catalog_temporary_path_keeps_the_typed_refusal() {
2551        let dir = tempdir().unwrap();
2552        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2553        let mut db = Database::open(dir.path()).unwrap();
2554        let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2555        // NOTE: a directory on the .tmp path fails the write with EISDIR whatever the uid is.
2556        let temporary = dir.path().join(format!("{MANIFEST_FILE}.tmp"));
2557        fs::create_dir(&temporary).unwrap();
2558
2559        let error = db
2560            .create_table(
2561                "orders",
2562                json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2563            )
2564            .unwrap_err();
2565
2566        let refusal = match &error {
2567            DatabaseError::Io(inner) => AtomicTemporaryFileError::downcast(inner)
2568                .expect("a .tmp path that could not be claimed stays a temporary-file refusal"),
2569            other => panic!("expected an io error keeping the typed refusal, got {other:?}"),
2570        };
2571        assert_eq!(refusal.path, temporary);
2572        assert!(refusal.reason.contains("could not be cleared"), "{}", refusal.reason);
2573        assert_eq!(db.table_names(), vec!["users"]);
2574        assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2575    }
2576
2577    #[test]
2578    fn a_catalog_write_that_fails_for_any_other_reason_is_still_a_catalog_refusal() {
2579        let unreadable = catalog_write_error(Box::new(HaysonSchemaError("nope".into())));
2580
2581        assert!(matches!(unreadable, DatabaseError::Catalog(_)), "got {unreadable:?}");
2582        assert!(matches!(
2583            catalog_write_error(Box::new(io::Error::from(io::ErrorKind::PermissionDenied))),
2584            DatabaseError::Catalog(_)
2585        ));
2586    }
2587
2588    #[test]
2589    fn drop_field_without_dependents_removes_it_from_disk_and_from_memory() {
2590        let dir = tempdir().unwrap();
2591        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2592        let mut db = Database::open(dir.path()).unwrap();
2593        db.table_mut("users")
2594            .unwrap()
2595            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2596            .unwrap();
2597        db.table_mut("users").unwrap().delete("u1", false).unwrap();
2598
2599        db.drop_field("users", "name", false).unwrap();
2600
2601        assert!(!db.table("users").unwrap().schema().fields.contains_key("name"));
2602        assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2603
2604        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2605        assert!(!reread.tables["users"].fields.contains_key("name"));
2606        db.close().unwrap();
2607    }
2608
2609    #[test]
2610    fn drop_field_refuses_a_live_residual_field_without_force() {
2611        let dir = tempdir().unwrap();
2612        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2613        let mut db = Database::open(dir.path()).unwrap();
2614        db.table_mut("users")
2615            .unwrap()
2616            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2617            .unwrap();
2618
2619        let error = db.drop_field("users", "name", false);
2620        assert!(error.is_err());
2621        assert!(db.catalog.tables["users"].fields.contains_key("name"));
2622        assert!(db.table("users").unwrap().schema().fields.contains_key("name"));
2623
2624        db.drop_field("users", "name", true).unwrap();
2625        assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2626        db.close().unwrap();
2627    }
2628
2629    fn users_catalog_indexing_name() -> Catalog {
2630        let mut catalog = catalog_of(&[users_schema()]);
2631        catalog.tables.get_mut("users").unwrap().indexes = Some(vec![IndexEntry {
2632            name: "by_name".to_string(),
2633            fields: vec!["name".to_string()],
2634            kind: CatalogIndexKind::Exact,
2635            unique: false,
2636            where_: None,
2637            expire_after_seconds: None,
2638        }]);
2639        catalog
2640    }
2641
2642    #[test]
2643    fn drop_field_names_the_declaration_when_residual_records_also_block_it() {
2644        let dir = tempdir().unwrap();
2645        write_catalog_to(dir.path(), &users_catalog_indexing_name());
2646        let mut db = Database::open(dir.path()).unwrap();
2647        db.table_mut("users")
2648            .unwrap()
2649            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2650            .unwrap();
2651
2652        let error = db.drop_field("users", "name", false).unwrap_err();
2653
2654        assert!(
2655            matches!(error, DatabaseError::Schema(_)),
2656            "expected a schema refusal, got {error:?}"
2657        );
2658        let message = error.to_string();
2659        assert!(message.contains("by_name"), "{message}");
2660        assert!(!message.contains("live record(s)"), "{message}");
2661        assert!(db.catalog.tables["users"].fields.contains_key("name"));
2662        db.close().unwrap();
2663    }
2664
2665    #[test]
2666    fn drop_field_names_the_declaration_when_it_is_the_only_thing_blocking_it() {
2667        let dir = tempdir().unwrap();
2668        write_catalog_to(dir.path(), &users_catalog_indexing_name());
2669        let mut db = Database::open(dir.path()).unwrap();
2670
2671        let error = db.drop_field("users", "name", false).unwrap_err();
2672
2673        assert!(
2674            matches!(error, DatabaseError::Schema(_)),
2675            "expected a schema refusal, got {error:?}"
2676        );
2677        assert!(error.to_string().contains("by_name"));
2678        assert!(db.catalog.tables["users"].fields.contains_key("name"));
2679        db.close().unwrap();
2680    }
2681
2682    #[test]
2683    fn drop_field_names_the_residual_records_when_no_declaration_blocks_it() {
2684        let dir = tempdir().unwrap();
2685        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2686        let mut db = Database::open(dir.path()).unwrap();
2687        db.table_mut("users")
2688            .unwrap()
2689            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2690            .unwrap();
2691
2692        let error = db.drop_field("users", "name", false).unwrap_err();
2693
2694        assert!(
2695            matches!(
2696                &error,
2697                DatabaseError::FieldInUse { table, field, records }
2698                    if table == "users" && field == "name" && *records == 1
2699            ),
2700            "expected a typed field-in-use refusal, got {error:?}"
2701        );
2702        assert!(db.catalog.tables["users"].fields.contains_key("name"));
2703        db.close().unwrap();
2704    }
2705
2706    #[test]
2707    fn drop_field_keeps_refusing_a_declared_dependency_even_with_force() {
2708        let dir = tempdir().unwrap();
2709        write_catalog_to(dir.path(), &users_catalog_indexing_name());
2710        let mut db = Database::open(dir.path()).unwrap();
2711        db.table_mut("users")
2712            .unwrap()
2713            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2714            .unwrap();
2715
2716        let error = db.drop_field("users", "name", true).unwrap_err();
2717
2718        assert!(
2719            matches!(error, DatabaseError::Schema(_)),
2720            "expected a schema refusal, got {error:?}"
2721        );
2722        assert!(error.to_string().contains("by_name"));
2723        db.close().unwrap();
2724    }
2725
2726    #[test]
2727    fn declare_index_with_a_conflicting_unique_value_refuses_and_changes_nothing() {
2728        let dir = tempdir().unwrap();
2729        let schema = parse_schema(&json!({
2730            "schemaVersion": 1,
2731            "table": "users",
2732            "primaryKey": "id",
2733            "fields": {
2734                "id": { "type": "string" },
2735                "email": { "type": "string" },
2736            },
2737        }))
2738        .unwrap();
2739        write_catalog_to(dir.path(), &catalog_of(&[schema]));
2740        let mut db = Database::open(dir.path()).unwrap();
2741        db.table_mut("users")
2742            .unwrap()
2743            .put(&json!({ "id": "u1", "email": "same@example.com" }), false)
2744            .unwrap();
2745        db.table_mut("users")
2746            .unwrap()
2747            .put(&json!({ "id": "u2", "email": "same@example.com" }), false)
2748            .unwrap();
2749
2750        let error = db.declare_index(
2751            "users",
2752            json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2753        );
2754        assert!(error.is_err());
2755        assert!(db.catalog.tables["users"].indexes.is_none());
2756
2757        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2758        assert!(reread.tables["users"].indexes.is_none());
2759        db.close().unwrap();
2760    }
2761
2762    fn orders_user_relation(enforce: bool) -> Value {
2763        json!({
2764            "name": "orders_user",
2765            "from": { "table": "orders", "field": "userId" },
2766            "to": { "table": "users", "field": "id" },
2767            "cardinality": "many-to-one",
2768            "enforce": enforce,
2769        })
2770    }
2771
2772    #[test]
2773    fn declare_relation_adds_a_relation_to_the_catalog_and_persists_it() {
2774        let dir = tempdir().unwrap();
2775        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2776        let mut db = Database::open(dir.path()).unwrap();
2777
2778        db.declare_relation(orders_user_relation(false), false).unwrap();
2779
2780        assert_eq!(db.catalog.relations.as_ref().unwrap()[0].name, "orders_user");
2781        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2782        assert_eq!(reread.relations.unwrap()[0].name, "orders_user");
2783        db.close().unwrap();
2784    }
2785
2786    #[test]
2787    fn declare_relation_with_enforce_true_refuses_over_a_live_orphan_and_force_overrides() {
2788        let dir = tempdir().unwrap();
2789        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2790        let mut db = Database::open(dir.path()).unwrap();
2791        db.table_mut("orders")
2792            .unwrap()
2793            .put(&json!({ "id": "o1", "userId": "ghost" }), false)
2794            .unwrap();
2795
2796        let error = db.declare_relation(orders_user_relation(true), false).unwrap_err();
2797        assert!(error.to_string().contains("orders_user"));
2798        assert!(db.catalog.relations.is_none());
2799
2800        db.declare_relation(orders_user_relation(true), true).unwrap();
2801        assert_eq!(db.catalog.relations.as_ref().unwrap()[0].enforce, Some(true));
2802        db.close().unwrap();
2803    }
2804
2805    #[test]
2806    fn put_checked_refuses_a_value_with_no_counterpart_on_an_enforced_relation() {
2807        let dir = tempdir().unwrap();
2808        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2809        let mut db = Database::open(dir.path()).unwrap();
2810        db.declare_relation(orders_user_relation(true), false).unwrap();
2811
2812        let error = db
2813            .put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
2814            .unwrap_err();
2815
2816        assert!(error.to_string().contains("orders_user"));
2817        assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
2818        db.close().unwrap();
2819    }
2820
2821    #[test]
2822    fn put_checked_allows_a_value_with_a_live_counterpart() {
2823        let dir = tempdir().unwrap();
2824        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2825        let mut db = Database::open(dir.path()).unwrap();
2826        db.declare_relation(orders_user_relation(true), false).unwrap();
2827        db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2828            .unwrap();
2829
2830        db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2831            .unwrap();
2832
2833        assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
2834        db.close().unwrap();
2835    }
2836
2837    #[test]
2838    fn put_checked_ignores_a_relation_that_does_not_declare_enforce() {
2839        let dir = tempdir().unwrap();
2840        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2841        let mut db = Database::open(dir.path()).unwrap();
2842        db.declare_relation(orders_user_relation(false), false).unwrap();
2843
2844        db.put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
2845            .unwrap();
2846
2847        assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
2848        db.close().unwrap();
2849    }
2850
2851    #[test]
2852    fn patch_checked_refuses_changing_the_field_to_a_value_with_no_counterpart() {
2853        let dir = tempdir().unwrap();
2854        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2855        let mut db = Database::open(dir.path()).unwrap();
2856        db.declare_relation(orders_user_relation(true), false).unwrap();
2857        db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2858            .unwrap();
2859        db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2860            .unwrap();
2861
2862        let error = db
2863            .patch_checked("orders", "o1", &json!({ "userId": "ghost" }), false)
2864            .unwrap_err();
2865
2866        assert!(error.to_string().contains("orders_user"));
2867        assert_eq!(
2868            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2869            Some(&json!("u1"))
2870        );
2871        db.close().unwrap();
2872    }
2873
2874    #[test]
2875    fn patch_checked_leaves_the_field_alone_when_changes_does_not_touch_it() {
2876        let dir = tempdir().unwrap();
2877        let orders_named = parse_schema(&json!({
2878            "schemaVersion": 1,
2879            "table": "orders",
2880            "primaryKey": "id",
2881            "fields": {
2882                "id": { "type": "string" },
2883                "userId": { "type": "string", "required": true },
2884                "note": { "type": "string" },
2885            },
2886        }))
2887        .unwrap();
2888        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_named]));
2889        let mut db = Database::open(dir.path()).unwrap();
2890        db.declare_relation(orders_user_relation(true), false).unwrap();
2891        db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2892            .unwrap();
2893        db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2894            .unwrap();
2895
2896        db.patch_checked("orders", "o1", &json!({ "note": "gift-wrapped" }), false)
2897            .unwrap();
2898
2899        assert_eq!(
2900            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("note"),
2901            Some(&json!("gift-wrapped"))
2902        );
2903        db.close().unwrap();
2904    }
2905
2906    #[test]
2907    fn declare_index_without_a_conflict_covers_records_written_before_the_declaration() {
2908        let dir = tempdir().unwrap();
2909        let schema = parse_schema(&json!({
2910            "schemaVersion": 1,
2911            "table": "users",
2912            "primaryKey": "id",
2913            "fields": {
2914                "id": { "type": "string" },
2915                "email": { "type": "string" },
2916            },
2917        }))
2918        .unwrap();
2919        write_catalog_to(dir.path(), &catalog_of(&[schema]));
2920        let mut db = Database::open(dir.path()).unwrap();
2921        db.table_mut("users")
2922            .unwrap()
2923            .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
2924            .unwrap();
2925        db.table_mut("users")
2926            .unwrap()
2927            .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
2928            .unwrap();
2929        db.table_mut("users")
2930            .unwrap()
2931            .put(&json!({ "id": "u3", "email": "kasia@example.com" }), false)
2932            .unwrap();
2933
2934        db.declare_index(
2935            "users",
2936            json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2937        )
2938        .unwrap();
2939
2940        assert_eq!(db.catalog.tables["users"].indexes.as_ref().unwrap().len(), 1);
2941
2942        let spec = QueryBuilder::new()
2943            .r#where(
2944                "email",
2945                crate::query::predicates::ComparisonOp::Eq,
2946                json!("piotr@example.com"),
2947            )
2948            .spec();
2949        let found = db.table("users").unwrap().query(&spec).unwrap();
2950        assert_eq!(found.keys, vec!["u2".to_string()]);
2951
2952        let conflicting = db
2953            .table_mut("users")
2954            .unwrap()
2955            .put(&json!({ "id": "u4", "email": "piotr@example.com" }), false);
2956        assert!(conflicting.is_err());
2957        db.close().unwrap();
2958    }
2959
2960    #[test]
2961    fn declare_index_snapshots_the_reopened_table_and_keeps_snapshotting_it_afterwards() {
2962        let dir = tempdir().unwrap();
2963        let schema = parse_schema(&json!({
2964            "schemaVersion": 1,
2965            "table": "users",
2966            "primaryKey": "id",
2967            "fields": {
2968                "id": { "type": "string" },
2969                "email": { "type": "string" },
2970            },
2971        }))
2972        .unwrap();
2973        write_catalog_to(dir.path(), &catalog_of(&[schema]));
2974        let mut db = Database::open(dir.path()).unwrap();
2975        db.table_mut("users")
2976            .unwrap()
2977            .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
2978            .unwrap();
2979
2980        db.declare_index(
2981            "users",
2982            json!({ "name": "users_email_exact", "fields": ["email"], "kind": "exact" }),
2983        )
2984        .unwrap();
2985
2986        assert!(
2987            index_path(dir.path(), "users").exists(),
2988            "declare_index owes the table a snapshot carrying the new index's projection"
2989        );
2990
2991        db.table_mut("users")
2992            .unwrap()
2993            .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
2994            .unwrap();
2995        assert!(
2996            db.table("users").unwrap().needs_snapshot(),
2997            "the handle declare_index left behind must still be able to snapshot itself"
2998        );
2999        db.close().unwrap();
3000
3001        let reopened = Database::open(dir.path()).unwrap();
3002        assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
3003        let spec = QueryBuilder::new()
3004            .r#where(
3005                "email",
3006                crate::query::predicates::ComparisonOp::Eq,
3007                json!("piotr@example.com"),
3008            )
3009            .spec();
3010        assert_eq!(
3011            reopened.table("users").unwrap().query(&spec).unwrap().keys,
3012            vec!["u2".to_string()]
3013        );
3014        assert_eq!(
3015            declared_on_disk(dir.path(), "users"),
3016            vec!["users_email_exact".to_string()]
3017        );
3018        reopened.close().unwrap();
3019    }
3020
3021    #[test]
3022    fn a_database_wide_sync_default_reaches_a_table_declaring_none_of_its_own() {
3023        let dir = tempdir().unwrap();
3024        let mut catalog = catalog_of(&[users_schema()]);
3025        catalog.defaults = Some(crate::catalog::TableOptions {
3026            sync: Some(crate::catalog::CatalogSyncMode::Never),
3027            ..Default::default()
3028        });
3029        write_catalog_to(dir.path(), &catalog);
3030
3031        let db = Database::open(dir.path()).unwrap();
3032        assert_eq!(
3033            db.table("users").unwrap().sync_mode(),
3034            crate::catalog::CatalogSyncMode::Never
3035        );
3036        db.close().unwrap();
3037    }
3038
3039    #[test]
3040    fn a_table_declaring_its_own_sync_mode_beats_the_database_wide_default() {
3041        let dir = tempdir().unwrap();
3042        let mut catalog = catalog_of(&[users_schema(), orders_schema()]);
3043        catalog.defaults = Some(crate::catalog::TableOptions {
3044            sync: Some(crate::catalog::CatalogSyncMode::Never),
3045            ..Default::default()
3046        });
3047        catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
3048            sync: Some(crate::catalog::CatalogSyncMode::Commit),
3049            ..Default::default()
3050        });
3051        write_catalog_to(dir.path(), &catalog);
3052
3053        let db = Database::open(dir.path()).unwrap();
3054        assert_eq!(
3055            db.table("users").unwrap().sync_mode(),
3056            crate::catalog::CatalogSyncMode::Commit
3057        );
3058        assert_eq!(
3059            db.table("orders").unwrap().sync_mode(),
3060            crate::catalog::CatalogSyncMode::Never
3061        );
3062        db.close().unwrap();
3063    }
3064
3065    #[test]
3066    fn a_table_declaring_validation_warn_takes_a_write_with_a_stray_field_and_stores_it_without_that_field() {
3067        let dir = tempdir().unwrap();
3068        let mut catalog = catalog_of(&[users_schema()]);
3069        catalog.tables.get_mut("users").unwrap().validation = Some(crate::catalog::CatalogValidation::Warn);
3070        write_catalog_to(dir.path(), &catalog);
3071
3072        let mut db = Database::open(dir.path()).unwrap();
3073        let document = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });
3074        let table = db.table_mut("users").unwrap();
3075        let record = table.put(&document, false).unwrap();
3076
3077        assert_eq!(record.get("nickname"), None);
3078        assert_eq!(
3079            table.dropped_field_warnings(&document),
3080            vec!["field \"nickname\": is not declared in the schema, dropped from the write".to_string()]
3081        );
3082        db.close().unwrap();
3083
3084        let reopened = Database::open(dir.path()).unwrap();
3085        let stored = reopened.table("users").unwrap().get("u1").unwrap().unwrap();
3086        assert_eq!(stored.get("name"), Some(&json!("Ania")));
3087        assert_eq!(stored.get("nickname"), None);
3088        reopened.close().unwrap();
3089    }
3090
3091    #[test]
3092    fn a_table_declaring_no_validation_refuses_a_write_with_a_stray_field() {
3093        let dir = tempdir().unwrap();
3094        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3095
3096        let mut db = Database::open(dir.path()).unwrap();
3097        assert_eq!(
3098            db.table("users").unwrap().validation(),
3099            crate::catalog::CatalogValidation::Strict
3100        );
3101        let error = db
3102            .table_mut("users")
3103            .unwrap()
3104            .put(&json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }), false)
3105            .unwrap_err();
3106
3107        assert!(error.to_string().contains("is not declared in the schema"), "{error}");
3108        db.close().unwrap();
3109    }
3110
3111    fn deprecating_age(dir: &Path, validation: Option<crate::catalog::CatalogValidation>, note: &str) {
3112        let mut catalog = catalog_of(&[users_with_age_schema()]);
3113        let entry = catalog.tables.get_mut("users").unwrap();
3114        entry.validation = validation;
3115        entry.fields.get_mut("age").unwrap().deprecated = Some(note.to_string());
3116        write_catalog_to(dir, &catalog);
3117    }
3118
3119    fn users_with_age_schema() -> TableSchema {
3120        parse_schema(&json!({
3121            "schemaVersion": 1,
3122            "table": "users",
3123            "primaryKey": "id",
3124            "fields": {
3125                "id": { "type": "string" },
3126                "name": { "type": "string", "required": true },
3127                "age": { "type": "int" },
3128            },
3129        }))
3130        .unwrap()
3131    }
3132
3133    #[test]
3134    fn a_write_touching_a_field_the_catalog_marks_deprecated_warns_with_the_field_and_the_note() {
3135        let dir = tempdir().unwrap();
3136        deprecating_age(dir.path(), None, "use \"bornAt\" instead");
3137
3138        let mut db = Database::open(dir.path()).unwrap();
3139        let table = db.table_mut("users").unwrap();
3140        let touching = json!({ "id": "u1", "name": "Ania", "age": 30 });
3141        let record = table.put(&touching, false).unwrap();
3142
3143        assert_eq!(record.get("age"), Some(&json!(30)));
3144        assert_eq!(
3145            table.deprecated_field_warnings(&touching),
3146            vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()]
3147        );
3148        assert!(table
3149            .deprecated_field_warnings(&json!({ "id": "u2", "name": "Ola" }))
3150            .is_empty());
3151        db.close().unwrap();
3152    }
3153
3154    #[test]
3155    fn the_deprecated_mark_reaches_a_table_in_either_validation_mode_and_on_every_write() {
3156        let expected = vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()];
3157        let document = json!({ "id": "u1", "name": "Ania", "age": 30 });
3158
3159        for validation in [None, Some(crate::catalog::CatalogValidation::Warn)] {
3160            let dir = tempdir().unwrap();
3161            deprecating_age(dir.path(), validation, "use \"bornAt\" instead");
3162
3163            let mut db = Database::open(dir.path()).unwrap();
3164            let table = db.table_mut("users").unwrap();
3165            table.put(&document, false).unwrap();
3166            assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
3167            table.put(&document, false).unwrap();
3168            assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
3169            db.close().unwrap();
3170
3171            let reopened = Database::open(dir.path()).unwrap();
3172            assert_eq!(
3173                reopened.table("users").unwrap().deprecated_field_warnings(&document),
3174                expected
3175            );
3176            reopened.close().unwrap();
3177        }
3178    }
3179
3180    #[test]
3181    fn a_catalog_marking_no_field_deprecated_leaves_every_table_without_the_warning() {
3182        let dir = tempdir().unwrap();
3183        write_catalog_to(dir.path(), &catalog_of(&[users_with_age_schema()]));
3184
3185        let db = Database::open(dir.path()).unwrap();
3186        let table = db.table("users").unwrap();
3187
3188        assert!(table.deprecated_fields().is_empty());
3189        assert!(table
3190            .deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania", "age": 30 }))
3191            .is_empty());
3192        db.close().unwrap();
3193    }
3194
3195    #[test]
3196    fn a_catalog_naming_no_sync_anywhere_leaves_every_table_committing() {
3197        let dir = tempdir().unwrap();
3198        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3199
3200        let db = Database::open(dir.path()).unwrap();
3201        assert_eq!(
3202            db.table("users").unwrap().sync_mode(),
3203            crate::catalog::CatalogSyncMode::Commit
3204        );
3205        db.close().unwrap();
3206    }
3207
3208    #[test]
3209    fn a_sync_never_table_survives_a_close_and_reopen_with_its_writes_and_its_mode() {
3210        let dir = tempdir().unwrap();
3211        let mut catalog = catalog_of(&[users_schema()]);
3212        catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
3213            sync: Some(crate::catalog::CatalogSyncMode::Never),
3214            ..Default::default()
3215        });
3216        write_catalog_to(dir.path(), &catalog);
3217
3218        {
3219            let mut db = Database::open(dir.path()).unwrap();
3220            db.table_mut("users")
3221                .unwrap()
3222                .put(&json!({ "id": "u1", "name": "Ania" }), false)
3223                .unwrap();
3224            db.close().unwrap();
3225        }
3226
3227        let db = Database::open(dir.path()).unwrap();
3228        assert_eq!(
3229            db.table("users").unwrap().sync_mode(),
3230            crate::catalog::CatalogSyncMode::Never
3231        );
3232        assert_eq!(
3233            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3234            Some(&json!("Ania"))
3235        );
3236        db.close().unwrap();
3237    }
3238
3239    #[test]
3240    fn a_database_wide_default_names_the_link_parsing_mode_a_table_declaring_none_opens_with() {
3241        let dir = tempdir().unwrap();
3242        let pages_schema = parse_schema(&json!({
3243            "schemaVersion": 1,
3244            "table": "pages",
3245            "primaryKey": "id",
3246            "fields": {
3247                "id": { "type": "string" },
3248                "page": { "type": "markdown" },
3249            },
3250        }))
3251        .unwrap();
3252        let mut catalog = catalog_of(&[pages_schema]);
3253        catalog.defaults = Some(crate::catalog::TableOptions {
3254            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3255            ..Default::default()
3256        });
3257        write_catalog_to(dir.path(), &catalog);
3258
3259        let db = Database::open(dir.path()).unwrap();
3260        assert_eq!(
3261            db.table("pages").unwrap().link_parsing(),
3262            crate::catalog::LinkParsing::Lazy
3263        );
3264        db.close().unwrap();
3265    }
3266
3267    #[test]
3268    fn a_table_declaring_its_own_link_parsing_mode_beats_the_database_wide_default() {
3269        let dir = tempdir().unwrap();
3270        let pages_schema = parse_schema(&json!({
3271            "schemaVersion": 1,
3272            "table": "pages",
3273            "primaryKey": "id",
3274            "fields": {
3275                "id": { "type": "string" },
3276                "page": { "type": "markdown" },
3277            },
3278        }))
3279        .unwrap();
3280        let mut catalog = catalog_of(&[pages_schema]);
3281        catalog.defaults = Some(crate::catalog::TableOptions {
3282            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3283            ..Default::default()
3284        });
3285        catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
3286            link_parsing: Some(crate::catalog::LinkParsing::Eager),
3287            ..Default::default()
3288        });
3289        write_catalog_to(dir.path(), &catalog);
3290
3291        let db = Database::open(dir.path()).unwrap();
3292        assert_eq!(
3293            db.table("pages").unwrap().link_parsing(),
3294            crate::catalog::LinkParsing::Eager
3295        );
3296        db.close().unwrap();
3297    }
3298
3299    #[test]
3300    fn a_database_wide_default_names_the_image_dir_a_table_declaring_none_writes_under() {
3301        let dir = tempdir().unwrap();
3302        let docs_schema = parse_schema(&json!({
3303            "schemaVersion": 1,
3304            "table": "docs",
3305            "primaryKey": "id",
3306            "fields": {
3307                "id": { "type": "string" },
3308                "cover": { "type": "image" },
3309            },
3310        }))
3311        .unwrap();
3312        let mut catalog = catalog_of(&[docs_schema]);
3313        catalog.defaults = Some(crate::catalog::TableOptions {
3314            image_dir: Some("shared-images".to_string()),
3315            ..Default::default()
3316        });
3317        write_catalog_to(dir.path(), &catalog);
3318
3319        let mut db = Database::open(dir.path()).unwrap();
3320        let record = db
3321            .table_mut("docs")
3322            .unwrap()
3323            .put(
3324                &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
3325                false,
3326            )
3327            .unwrap();
3328        let physical_path = record
3329            .get("cover")
3330            .unwrap()
3331            .get("physicalPath")
3332            .unwrap()
3333            .as_str()
3334            .unwrap();
3335
3336        assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
3337        assert!(dir.path().join("shared-images").join(physical_path).exists());
3338        db.close().unwrap();
3339    }
3340
3341    #[test]
3342    fn declare_index_of_a_composite_field_pair_serves_a_query_matching_both_fields() {
3343        let dir = tempdir().unwrap();
3344        let schema = parse_schema(&json!({
3345            "schemaVersion": 1,
3346            "table": "orders",
3347            "primaryKey": "id",
3348            "fields": {
3349                "id": { "type": "string" },
3350                "userId": { "type": "string" },
3351                "status": { "type": "string" },
3352            },
3353        }))
3354        .unwrap();
3355        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3356        let mut db = Database::open(dir.path()).unwrap();
3357        db.table_mut("orders")
3358            .unwrap()
3359            .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
3360            .unwrap();
3361        db.table_mut("orders")
3362            .unwrap()
3363            .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
3364            .unwrap();
3365        db.table_mut("orders")
3366            .unwrap()
3367            .put(&json!({ "id": "o3", "userId": "u2", "status": "open" }), false)
3368            .unwrap();
3369
3370        db.declare_index(
3371            "orders",
3372            json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
3373        )
3374        .unwrap();
3375
3376        let spec = QueryBuilder::new()
3377            .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
3378            .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("open"))
3379            .spec();
3380        let found = db.table("orders").unwrap().query(&spec).unwrap();
3381        assert_eq!(found.keys, vec!["o1".to_string()]);
3382        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3383        assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
3384        assert_eq!(found.plan.candidates, 1);
3385        db.close().unwrap();
3386    }
3387
3388    #[test]
3389    fn a_where_scoped_index_answers_only_a_query_that_repeats_its_predicate() {
3390        let dir = tempdir().unwrap();
3391        let schema = parse_schema(&json!({
3392            "schemaVersion": 1,
3393            "table": "users",
3394            "primaryKey": "id",
3395            "fields": {
3396                "id": { "type": "string" },
3397                "role": { "type": "string" },
3398                "status": { "type": "string" },
3399            },
3400        }))
3401        .unwrap();
3402        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3403        let mut db = Database::open(dir.path()).unwrap();
3404        db.table_mut("users")
3405            .unwrap()
3406            .put(&json!({ "id": "u1", "role": "admin", "status": "active" }), false)
3407            .unwrap();
3408        db.table_mut("users")
3409            .unwrap()
3410            .put(&json!({ "id": "u2", "role": "admin", "status": "inactive" }), false)
3411            .unwrap();
3412
3413        db.declare_index(
3414            "users",
3415            json!({
3416                "name": "users_role_active",
3417                "fields": ["role"],
3418                "kind": "exact",
3419                "where": [{ "field": "status", "op": "=", "value": "active" }],
3420            }),
3421        )
3422        .unwrap();
3423
3424        let eq = crate::query::predicates::ComparisonOp::Eq;
3425        let omitted = QueryBuilder::new().r#where("role", eq, json!("admin")).spec();
3426        let found = db.table("users").unwrap().query(&omitted).unwrap();
3427        assert_eq!(
3428            found.keys,
3429            vec!["u1".to_string(), "u2".to_string()],
3430            "the index holds u1 alone, but the query never asked for active records only"
3431        );
3432        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Scan);
3433
3434        let repeated = QueryBuilder::new()
3435            .r#where("role", eq, json!("admin"))
3436            .r#where("status", eq, json!("active"))
3437            .spec();
3438        let narrowed = db.table("users").unwrap().query(&repeated).unwrap();
3439        assert_eq!(narrowed.keys, vec!["u1".to_string()]);
3440        assert_eq!(narrowed.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3441        assert_eq!(narrowed.plan.index, Some("users_role_active".to_string()));
3442        assert_eq!(narrowed.plan.candidates, 1);
3443        db.close().unwrap();
3444    }
3445
3446    #[test]
3447    fn a_declared_composite_index_still_serves_queries_after_closing_and_reopening_the_database() {
3448        let dir = tempdir().unwrap();
3449        let schema = parse_schema(&json!({
3450            "schemaVersion": 1,
3451            "table": "orders",
3452            "primaryKey": "id",
3453            "fields": {
3454                "id": { "type": "string" },
3455                "userId": { "type": "string" },
3456                "status": { "type": "string" },
3457            },
3458        }))
3459        .unwrap();
3460        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3461        let mut db = Database::open(dir.path()).unwrap();
3462        db.table_mut("orders")
3463            .unwrap()
3464            .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
3465            .unwrap();
3466        db.table_mut("orders")
3467            .unwrap()
3468            .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
3469            .unwrap();
3470        db.declare_index(
3471            "orders",
3472            json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
3473        )
3474        .unwrap();
3475        db.close().unwrap();
3476
3477        let reopened = Database::open(dir.path()).unwrap();
3478        let spec = QueryBuilder::new()
3479            .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
3480            .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("closed"))
3481            .spec();
3482        let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3483        assert_eq!(found.keys, vec!["o2".to_string()]);
3484        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3485        assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
3486        reopened.close().unwrap();
3487    }
3488
3489    fn add_index_to_catalog_on_disk(dir: &Path, table: &str, index: IndexEntry) {
3490        let mut catalog = read_catalog(&dir.join(MANIFEST_FILE)).unwrap().unwrap();
3491        let entry = catalog.tables.get_mut(table).unwrap();
3492        entry.indexes = Some(vec![index]);
3493        write_catalog_to(dir, &catalog);
3494    }
3495
3496    fn sorted_index_entry(name: &str, fields: &[&str]) -> IndexEntry {
3497        IndexEntry {
3498            name: name.to_string(),
3499            fields: fields.iter().map(|field| field.to_string()).collect(),
3500            kind: CatalogIndexKind::Sorted,
3501            unique: false,
3502            where_: None,
3503            expire_after_seconds: None,
3504        }
3505    }
3506
3507    /// A snapshot entry carries the projection the registry held when it was written, so a snapshot
3508    /// taken before an index was declared knows none of that index's fields. Restoring from it
3509    /// would leave the new index empty and answer every query against it with silence — the
3510    /// fingerprint has to count the index set, not the schema alone, so this pairing is stale.
3511    #[test]
3512    fn a_snapshot_written_before_an_index_was_declared_is_rebuilt_rather_than_restored_into_it() {
3513        let dir = tempdir().unwrap();
3514        let schema = parse_schema(&json!({
3515            "schemaVersion": 1,
3516            "table": "orders",
3517            "primaryKey": "id",
3518            "fields": {
3519                "id": { "type": "string" },
3520                "userId": { "type": "string" },
3521                "total": { "type": "int" },
3522            },
3523        }))
3524        .unwrap();
3525        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3526        let mut db = Database::open(dir.path()).unwrap();
3527        let orders = db.table_mut("orders").unwrap();
3528        orders
3529            .put(&json!({ "id": "a", "userId": "u1", "total": 10 }), false)
3530            .unwrap();
3531        orders.put(&json!({ "id": "b", "userId": "u1" }), false).unwrap();
3532        orders
3533            .put(&json!({ "id": "c", "userId": "u0", "total": 5 }), false)
3534            .unwrap();
3535        db.close().unwrap();
3536        assert!(
3537            index_path(dir.path(), "orders").exists(),
3538            "the fixture needs a snapshot to restore from"
3539        );
3540
3541        add_index_to_catalog_on_disk(
3542            dir.path(),
3543            "orders",
3544            sorted_index_entry("orders_user_total", &["userId", "total"]),
3545        );
3546
3547        let reopened = Database::open(dir.path()).unwrap();
3548        assert_eq!(reopened.restore_outcome("orders"), Some(RestoreOutcome::Rebuilt));
3549
3550        let spec = QueryBuilder::new()
3551            .r#where("userId", crate::query::predicates::ComparisonOp::Gte, json!("u1"))
3552            .spec();
3553        let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3554        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Sorted);
3555        assert_eq!(found.plan.index, Some("orders_user_total".to_string()));
3556        let mut keys = found.keys.clone();
3557        keys.sort();
3558        assert_eq!(
3559            keys,
3560            vec!["a".to_string(), "b".to_string()],
3561            "b has no total, but it still has a userId"
3562        );
3563        reopened.close().unwrap();
3564    }
3565
3566    /// The registry spans every partition of a table, so a composite unique index does too — but
3567    /// only if it was filled from the log rather than from a snapshot that predates it.
3568    #[test]
3569    fn a_composite_unique_index_declared_after_a_snapshot_still_holds_across_partitions() {
3570        let dir = tempdir().unwrap();
3571        let schema = parse_schema(&json!({
3572            "schemaVersion": 1,
3573            "table": "tickets",
3574            "primaryKey": "id",
3575            "partitionBy": "kind",
3576            "fields": {
3577                "id": { "type": "string" },
3578                "kind": { "type": "string", "required": true },
3579                "userId": { "type": "string", "required": true },
3580                "orderId": { "type": "int" },
3581            },
3582        }))
3583        .unwrap();
3584        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3585        let mut db = Database::open(dir.path()).unwrap();
3586        let tickets = db.table_mut("tickets").unwrap();
3587        tickets
3588            .put(
3589                &json!({ "id": "t1", "kind": "sale", "userId": "u1", "orderId": 1 }),
3590                false,
3591            )
3592            .unwrap();
3593        tickets
3594            .put(
3595                &json!({ "id": "t2", "kind": "sale", "userId": "u1", "orderId": 2 }),
3596                false,
3597            )
3598            .unwrap();
3599        tickets
3600            .put(
3601                &json!({ "id": "t3", "kind": "refund", "userId": "u1", "orderId": 3 }),
3602                false,
3603            )
3604            .unwrap();
3605        db.close().unwrap();
3606
3607        add_index_to_catalog_on_disk(
3608            dir.path(),
3609            "tickets",
3610            IndexEntry {
3611                name: "tickets_user_order".to_string(),
3612                fields: vec!["userId".to_string(), "orderId".to_string()],
3613                kind: CatalogIndexKind::Exact,
3614                unique: true,
3615                where_: None,
3616                expire_after_seconds: None,
3617            },
3618        );
3619
3620        let mut reopened = Database::open(dir.path()).unwrap();
3621        let tickets = reopened.table_mut("tickets").unwrap();
3622        // NOTE: the clash sits in the "sale" partition, the write lands in "refund" — one key, two files.
3623        let clash = tickets.put(
3624            &json!({ "id": "t9", "kind": "refund", "userId": "u1", "orderId": 2 }),
3625            false,
3626        );
3627        match clash {
3628            Err(TableError::Unique(violation)) => {
3629                assert_eq!(violation.field, "userId, orderId");
3630                assert_eq!(violation.holder, "t2");
3631            }
3632            other => panic!("a repeated composite key must be refused, got {other:?}"),
3633        }
3634        tickets
3635            .put(
3636                &json!({ "id": "t10", "kind": "refund", "userId": "u1", "orderId": 9 }),
3637                false,
3638            )
3639            .unwrap();
3640        reopened.close().unwrap();
3641    }
3642
3643    #[test]
3644    fn close_snapshots_dirty_tables_so_a_reopen_restores_instead_of_rebuilding() {
3645        let dir = tempdir().unwrap();
3646        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3647        let mut db = Database::open(dir.path()).unwrap();
3648
3649        // NOTE: well under DEFAULT_SNAPSHOT_EVERY — only close(), not the threshold, should snapshot.
3650        db.table_mut("users")
3651            .unwrap()
3652            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3653            .unwrap();
3654        db.table_mut("users")
3655            .unwrap()
3656            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3657            .unwrap();
3658        assert!(!index_path(dir.path(), "users").exists());
3659
3660        db.close().unwrap();
3661        assert!(index_path(dir.path(), "users").exists());
3662
3663        let reopened = Database::open(dir.path()).unwrap();
3664        assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
3665        assert_eq!(
3666            reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3667            Some(&json!("Ania"))
3668        );
3669        assert_eq!(
3670            reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
3671            Some(&json!("Piotr"))
3672        );
3673        reopened.close().unwrap();
3674    }
3675
3676    #[test]
3677    fn close_commits_a_staged_write_so_a_reopen_still_finds_it() {
3678        let dir = tempdir().unwrap();
3679        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3680        let mut db = Database::open(dir.path()).unwrap();
3681        db.table_mut("users")
3682            .unwrap()
3683            .stage_put(&json!({ "id": "u1", "name": "Ania" }), false)
3684            .unwrap();
3685        assert_eq!(
3686            db.table("users").unwrap().len(),
3687            0,
3688            "nothing is committed before the close"
3689        );
3690
3691        db.close().unwrap();
3692
3693        let reopened = Database::open(dir.path()).unwrap();
3694        assert_eq!(
3695            reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3696            Some(&json!("Ania"))
3697        );
3698        reopened.close().unwrap();
3699    }
3700
3701    /// Same bytes, new inode — what `git checkout` does to a tracked file. Only the inode can
3702    /// tell, so this needs a host that reports one.
3703    #[cfg(unix)]
3704    fn swap_file_in_place(path: &Path) {
3705        let swapped = path.with_extension("swap");
3706        std::fs::copy(path, &swapped).unwrap();
3707        std::fs::rename(&swapped, path).unwrap();
3708    }
3709
3710    fn log_replaced_of(error: &io::Error) -> Option<&crate::errors::HaysonLogReplacedError> {
3711        error.get_ref()?.downcast_ref::<crate::errors::HaysonLogReplacedError>()
3712    }
3713
3714    /// The lock file names a live pid, so no staleness rule ever reclaims it: a `close()` that
3715    /// keeps it after a refused flush locks the directory out of this very process until the lease
3716    /// runs out.
3717    #[cfg(unix)]
3718    #[test]
3719    fn a_close_refused_by_a_swapped_log_still_releases_the_directory_lock() {
3720        let dir = tempdir().unwrap();
3721        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3722        let mut db = Database::open(dir.path()).unwrap();
3723        db.table_mut("users")
3724            .unwrap()
3725            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3726            .unwrap();
3727        db.table_mut("users")
3728            .unwrap()
3729            .stage_put(&json!({ "id": "u2", "name": "Piotr" }), false)
3730            .unwrap();
3731        swap_file_in_place(&log_path(dir.path(), "users"));
3732
3733        let error = db.close().unwrap_err();
3734
3735        assert!(log_replaced_of(&error).is_some(), "{error}");
3736        assert!(!lock_path_of(dir.path()).exists(), "the lock file outlived the handle");
3737        let started = std::time::Instant::now();
3738        let reopened = Database::open(dir.path()).unwrap();
3739        assert!(
3740            started.elapsed() < Duration::from_secs(1),
3741            "the reopen waited {:?} on a lock nobody holds",
3742            started.elapsed()
3743        );
3744        assert_eq!(
3745            reopened.table("users").unwrap().get("u2").unwrap(),
3746            None,
3747            "the staged write never reached the log"
3748        );
3749        reopened.close().unwrap();
3750    }
3751
3752    /// Puts an older version of the log at `path` — its first line only, on a new inode — the way
3753    /// a `git checkout` of an earlier commit does.
3754    #[cfg(unix)]
3755    fn swap_older_version_in(path: &Path) {
3756        let current = std::fs::read_to_string(path).unwrap();
3757        let older = format!("{}\n", current.lines().next().unwrap());
3758        let swapped = path.with_extension("swap");
3759        std::fs::write(&swapped, older).unwrap();
3760        std::fs::rename(&swapped, path).unwrap();
3761    }
3762
3763    /// The snapshot `close()` writes pairs with the log through the descriptor it holds — after a
3764    /// swap that is an `.idx` describing the orphaned inode, and the next open trips over it
3765    /// instead of reading the file the checkout put there.
3766    #[cfg(unix)]
3767    #[test]
3768    fn a_close_after_a_swapped_log_leaves_no_snapshot_so_the_reopen_reads_the_new_file() {
3769        let dir = tempdir().unwrap();
3770        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3771        let mut db = Database::open(dir.path()).unwrap();
3772        db.table_mut("users")
3773            .unwrap()
3774            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3775            .unwrap();
3776        db.table_mut("users")
3777            .unwrap()
3778            .put(&json!({ "id": "u1", "name": "Anna" }), false)
3779            .unwrap();
3780        swap_older_version_in(&log_path(dir.path(), "users"));
3781
3782        let error = db.close().unwrap_err();
3783
3784        assert!(log_replaced_of(&error).is_some(), "{error}");
3785        assert!(
3786            !index_path(dir.path(), "users").exists(),
3787            "a snapshot describing the orphaned inode was written"
3788        );
3789        let reopened = Database::open(dir.path()).unwrap();
3790        assert_eq!(
3791            reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3792            Some(&json!("Ania")),
3793            "the reopen reads what the checkout put at the path"
3794        );
3795        reopened.close().unwrap();
3796    }
3797
3798    /// The alternative to closing and reopening the whole database after a swap: `refresh()`
3799    /// notices the inode moved, reopens the log at its path and rebuilds from what is there now.
3800    #[cfg(unix)]
3801    #[test]
3802    fn refresh_after_a_swapped_log_takes_the_new_file_and_writes_land_in_it_again() {
3803        let dir = tempdir().unwrap();
3804        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3805        let mut db = Database::open(dir.path()).unwrap();
3806        db.table_mut("users")
3807            .unwrap()
3808            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3809            .unwrap();
3810        db.table_mut("users")
3811            .unwrap()
3812            .put(&json!({ "id": "u1", "name": "Anna" }), false)
3813            .unwrap();
3814        let log = log_path(dir.path(), "users");
3815        swap_older_version_in(&log);
3816        assert!(db
3817            .table_mut("users")
3818            .unwrap()
3819            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3820            .is_err());
3821
3822        assert!(db.refresh().unwrap(), "the swap counts as a change");
3823
3824        let users = db.table_mut("users").unwrap();
3825        assert_eq!(users.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
3826        users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
3827        assert_eq!(std::fs::read_to_string(&log).unwrap().lines().count(), 2);
3828        db.close().unwrap();
3829
3830        let reopened = Database::open(dir.path()).unwrap();
3831        assert_eq!(
3832            reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
3833            Some(&json!("Piotr"))
3834        );
3835        reopened.close().unwrap();
3836    }
3837
3838    #[test]
3839    fn a_low_snapshot_every_writes_the_index_without_waiting_for_close() {
3840        let dir = tempdir().unwrap();
3841        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3842        let mut db = Database::open_with(
3843            dir.path(),
3844            OpenOptions {
3845                snapshot_every: 2,
3846                ..Default::default()
3847            },
3848        )
3849        .unwrap();
3850
3851        db.table_mut("users")
3852            .unwrap()
3853            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3854            .unwrap();
3855        assert!(!index_path(dir.path(), "users").exists());
3856        db.table_mut("users")
3857            .unwrap()
3858            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3859            .unwrap();
3860        assert!(index_path(dir.path(), "users").exists());
3861
3862        db.close().unwrap();
3863    }
3864
3865    #[test]
3866    fn a_snapshot_taken_mid_stream_does_not_disturb_writes_that_follow_it() {
3867        let dir = tempdir().unwrap();
3868        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3869        let mut db = Database::open_with(
3870            dir.path(),
3871            OpenOptions {
3872                snapshot_every: 2,
3873                ..Default::default()
3874            },
3875        )
3876        .unwrap();
3877
3878        db.table_mut("users")
3879            .unwrap()
3880            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3881            .unwrap();
3882        db.table_mut("users")
3883            .unwrap()
3884            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3885            .unwrap();
3886        db.table_mut("users")
3887            .unwrap()
3888            .put(&json!({ "id": "u3", "name": "Kasia" }), false)
3889            .unwrap();
3890        assert_eq!(
3891            db.table("users").unwrap().get("u3").unwrap().unwrap().get("name"),
3892            Some(&json!("Kasia"))
3893        );
3894        db.close().unwrap();
3895    }
3896
3897    fn users_insert_audits_into_orders() -> Value {
3898        json!({
3899            "name": "users_insert_audit",
3900            "table": "users",
3901            "event": "insert",
3902            "effect": {
3903                "table": "orders",
3904                "fields": [
3905                    { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
3906                    { "target": "userId", "source": { "kind": "new", "field": "id" } },
3907                ],
3908            },
3909        })
3910    }
3911
3912    #[test]
3913    fn a_table_declaring_an_image_dir_option_writes_bytes_under_it_instead_of_the_default() {
3914        let dir = tempdir().unwrap();
3915        let docs_schema = parse_schema(&json!({
3916            "schemaVersion": 1,
3917            "table": "docs",
3918            "primaryKey": "id",
3919            "fields": {
3920                "id": { "type": "string" },
3921                "cover": { "type": "image" },
3922            },
3923        }))
3924        .unwrap();
3925        let mut catalog = catalog_of(&[docs_schema]);
3926        catalog.tables.get_mut("docs").unwrap().options = Some(crate::catalog::TableOptions {
3927            image_dir: Some("shared-images".to_string()),
3928            ..Default::default()
3929        });
3930        write_catalog_to(dir.path(), &catalog);
3931
3932        let mut db = Database::open(dir.path()).unwrap();
3933        let record = db
3934            .table_mut("docs")
3935            .unwrap()
3936            .put(
3937                &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
3938                false,
3939            )
3940            .unwrap();
3941        let physical_path = record
3942            .get("cover")
3943            .unwrap()
3944            .get("physicalPath")
3945            .unwrap()
3946            .as_str()
3947            .unwrap();
3948
3949        assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
3950        assert!(dir.path().join("shared-images").join(physical_path).exists());
3951        db.close().unwrap();
3952    }
3953
3954    #[test]
3955    fn a_table_declaring_a_lazy_link_parsing_option_opens_with_it_instead_of_the_eager_default() {
3956        let dir = tempdir().unwrap();
3957        let pages_schema = parse_schema(&json!({
3958            "schemaVersion": 1,
3959            "table": "pages",
3960            "primaryKey": "id",
3961            "fields": {
3962                "id": { "type": "string" },
3963                "page": { "type": "markdown" },
3964            },
3965        }))
3966        .unwrap();
3967        let mut catalog = catalog_of(&[pages_schema]);
3968        catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
3969            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3970            ..Default::default()
3971        });
3972        write_catalog_to(dir.path(), &catalog);
3973
3974        let mut db = Database::open(dir.path()).unwrap();
3975        assert_eq!(
3976            db.table("pages").unwrap().link_parsing(),
3977            crate::catalog::LinkParsing::Lazy
3978        );
3979
3980        let outcome = db
3981            .table_mut("pages")
3982            .unwrap()
3983            .put_with_warnings(
3984                &json!({ "id": "a", "page": { "content": "[missing](/docs/x.md)", "virtualPath": "/docs/a.md" } }),
3985                false,
3986            )
3987            .unwrap();
3988        assert!(outcome.warnings.is_empty());
3989        db.close().unwrap();
3990    }
3991
3992    #[test]
3993    fn a_table_declaring_no_link_parsing_option_defaults_to_eager() {
3994        let dir = tempdir().unwrap();
3995        let pages_schema = parse_schema(&json!({
3996            "schemaVersion": 1,
3997            "table": "pages",
3998            "primaryKey": "id",
3999            "fields": {
4000                "id": { "type": "string" },
4001                "page": { "type": "markdown" },
4002            },
4003        }))
4004        .unwrap();
4005        write_catalog_to(dir.path(), &catalog_of(&[pages_schema]));
4006
4007        let db = Database::open(dir.path()).unwrap();
4008        assert_eq!(
4009            db.table("pages").unwrap().link_parsing(),
4010            crate::catalog::LinkParsing::Eager
4011        );
4012    }
4013
4014    #[test]
4015    fn declare_trigger_writes_the_catalog_and_refreshes_both_tables_live_trigger_set() {
4016        let dir = tempdir().unwrap();
4017        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4018        let mut db = Database::open(dir.path()).unwrap();
4019
4020        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4021
4022        assert_eq!(db.catalog.triggers.as_ref().unwrap().len(), 1);
4023        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4024        assert!(db.table("orders").unwrap().triggers().is_empty());
4025
4026        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
4027        assert_eq!(reread.triggers.as_ref().unwrap().len(), 1);
4028        assert_eq!(reread.triggers.as_ref().unwrap()[0].name, "users_insert_audit");
4029
4030        db.close().unwrap();
4031    }
4032
4033    #[test]
4034    fn a_declared_trigger_fires_its_effect_immediately_on_the_next_unit_of_work() {
4035        let dir = tempdir().unwrap();
4036        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4037        let mut db = Database::open(dir.path()).unwrap();
4038        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4039
4040        let mut unit = db.unit_of_work();
4041        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4042        db.commit(&mut unit).unwrap();
4043
4044        assert_eq!(
4045            db.table("orders")
4046                .unwrap()
4047                .get("o-audit")
4048                .unwrap()
4049                .unwrap()
4050                .get("userId"),
4051            Some(&json!("u1"))
4052        );
4053        db.close().unwrap();
4054    }
4055
4056    #[test]
4057    fn a_declared_trigger_still_fires_after_closing_and_reopening_the_database() {
4058        let dir = tempdir().unwrap();
4059        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4060        let mut db = Database::open(dir.path()).unwrap();
4061        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4062        db.close().unwrap();
4063
4064        let mut reopened = Database::open(dir.path()).unwrap();
4065        let mut unit = reopened.unit_of_work();
4066        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4067        reopened.commit(&mut unit).unwrap();
4068
4069        assert_eq!(
4070            reopened
4071                .table("orders")
4072                .unwrap()
4073                .get("o-audit")
4074                .unwrap()
4075                .unwrap()
4076                .get("userId"),
4077            Some(&json!("u1"))
4078        );
4079        reopened.close().unwrap();
4080    }
4081
4082    #[test]
4083    fn a_declared_trigger_still_fires_after_declare_index_reopened_its_source_table() {
4084        let dir = tempdir().unwrap();
4085        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4086        let mut db = Database::open(dir.path()).unwrap();
4087        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4088
4089        db.declare_index(
4090            "users",
4091            json!({ "name": "users_by_name", "fields": ["name"], "kind": "exact" }),
4092        )
4093        .unwrap();
4094
4095        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4096
4097        let mut unit = db.unit_of_work();
4098        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4099        db.commit(&mut unit).unwrap();
4100
4101        assert_eq!(
4102            db.table("orders")
4103                .unwrap()
4104                .get("o-audit")
4105                .unwrap()
4106                .unwrap()
4107                .get("userId"),
4108            Some(&json!("u1"))
4109        );
4110        db.close().unwrap();
4111    }
4112
4113    #[test]
4114    fn a_declared_trigger_still_fires_after_drop_field_reopened_its_source_table() {
4115        let dir = tempdir().unwrap();
4116        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4117        let mut db = Database::open(dir.path()).unwrap();
4118        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4119
4120        db.drop_field("users", "name", false).unwrap();
4121
4122        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4123
4124        let mut unit = db.unit_of_work();
4125        unit.put("users", json!({ "id": "u1" }));
4126        db.commit(&mut unit).unwrap();
4127
4128        assert_eq!(
4129            db.table("orders")
4130                .unwrap()
4131                .get("o-audit")
4132                .unwrap()
4133                .unwrap()
4134                .get("userId"),
4135            Some(&json!("u1"))
4136        );
4137        db.close().unwrap();
4138    }
4139
4140    #[test]
4141    fn a_declared_trigger_still_fires_after_create_table_migrated_its_source_table() {
4142        let dir = tempdir().unwrap();
4143        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4144        let mut db = Database::open(dir.path()).unwrap();
4145        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4146
4147        db.create_table(
4148            "users",
4149            json!({
4150                "schemaVersion": 1,
4151                "table": "users",
4152                "revision": 2,
4153                "primaryKey": "id",
4154                "fields": {
4155                    "id": { "type": "string" },
4156                    "name": { "type": "string", "required": true },
4157                    "city": { "type": "string" },
4158                },
4159            }),
4160        )
4161        .unwrap();
4162
4163        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4164
4165        let mut unit = db.unit_of_work();
4166        unit.put("users", json!({ "id": "u1", "name": "Ania", "city": "Kraków" }));
4167        db.commit(&mut unit).unwrap();
4168
4169        assert_eq!(
4170            db.table("orders")
4171                .unwrap()
4172                .get("o-audit")
4173                .unwrap()
4174                .unwrap()
4175                .get("userId"),
4176            Some(&json!("u1"))
4177        );
4178        db.close().unwrap();
4179    }
4180
4181    #[test]
4182    fn declare_trigger_refuses_an_unknown_effect_table_and_changes_nothing_on_disk() {
4183        let dir = tempdir().unwrap();
4184        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4185        let mut db = Database::open(dir.path()).unwrap();
4186
4187        let spec = json!({
4188            "name": "users_insert_audit",
4189            "table": "users",
4190            "event": "insert",
4191            "effect": {
4192                "table": "ghost",
4193                "fields": [
4194                    { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
4195                ],
4196            },
4197        });
4198        let error = db.declare_trigger(spec);
4199        assert!(error.is_err());
4200        assert!(db.catalog.triggers.is_none());
4201
4202        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
4203        assert!(reread.triggers.is_none());
4204        db.close().unwrap();
4205    }
4206
4207    mod automatic_compaction {
4208        use super::*;
4209        use crate::catalog::TableOptions;
4210
4211        fn enabled(dead_bytes_ratio: f64, min_dead_bytes: u64) -> CatalogCompaction {
4212            CatalogCompaction::Enabled {
4213                dead_bytes_ratio,
4214                min_dead_bytes,
4215            }
4216        }
4217
4218        fn opened(dir: &Path, compaction: Option<CatalogCompaction>) -> Database {
4219            Database::open_with(
4220                dir,
4221                OpenOptions {
4222                    compaction,
4223                    ..Default::default()
4224                },
4225            )
4226            .unwrap()
4227        }
4228
4229        fn lines_of(path: &Path) -> Vec<String> {
4230            fs::read_to_string(path).unwrap().lines().map(str::to_string).collect()
4231        }
4232
4233        /// One put and `overwrites` patches on the same key, so every patch but the last turns a
4234        /// whole line into garbage without changing how many records the table holds.
4235        fn churn(db: &mut Database, table: &str, key: &str, overwrites: usize) {
4236            let users = db.table_mut(table).unwrap();
4237            users.put(&json!({ "id": key, "name": "Ania" }), false).unwrap();
4238            for round in 0..overwrites {
4239                users
4240                    .patch(key, &json!({ "name": format!("Ania {round}") }), false)
4241                    .unwrap();
4242            }
4243        }
4244
4245        #[test]
4246        fn garbage_under_the_absolute_floor_leaves_the_log_alone_however_large_its_share() {
4247            let dir = tempdir().unwrap();
4248            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4249            let mut db = opened(dir.path(), Some(enabled(0.1, 1_000_000)));
4250
4251            churn(&mut db, "users", "u1", 12);
4252
4253            let stats = db.table("users").unwrap().stats();
4254            assert_eq!(stats.gen, 0, "minDeadBytes was never reached");
4255            assert!(stats.dead_bytes > 0, "the overwrites did make garbage");
4256            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
4257            db.close().unwrap();
4258        }
4259
4260        #[test]
4261        fn garbage_below_the_ratio_leaves_the_log_alone_however_many_bytes_it_is() {
4262            let dir = tempdir().unwrap();
4263            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4264            let mut db = opened(dir.path(), Some(enabled(0.99, 1)));
4265
4266            churn(&mut db, "users", "u1", 4);
4267
4268            let stats = db.table("users").unwrap().stats();
4269            assert_eq!(stats.gen, 0, "four of five lines is garbage, but not 99% of the bytes");
4270            assert!(stats.dead_bytes > 1);
4271            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 5);
4272            db.close().unwrap();
4273        }
4274
4275        #[test]
4276        fn crossing_both_thresholds_rewrites_the_log_and_moves_the_generation_on() {
4277            let dir = tempdir().unwrap();
4278            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4279            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4280
4281            churn(&mut db, "users", "u1", 12);
4282
4283            let stats = db.table("users").unwrap().stats();
4284            assert!(stats.gen > 0, "the write that crossed the thresholds compacted");
4285            assert_eq!(stats.dead_bytes, 0);
4286            assert_eq!(stats.records, 1);
4287            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
4288            assert_eq!(
4289                db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
4290                Some(&json!("Ania 11"))
4291            );
4292            db.close().unwrap();
4293        }
4294
4295        #[test]
4296        fn a_log_the_thresholds_rewrote_reopens_at_the_generation_and_the_records_it_left() {
4297            let dir = tempdir().unwrap();
4298            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4299            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4300            churn(&mut db, "users", "u1", 12);
4301            db.table_mut("users")
4302                .unwrap()
4303                .put(&json!({ "id": "u2", "name": "Piotr" }), false)
4304                .unwrap();
4305            let before = db.table("users").unwrap().stats();
4306            db.close().unwrap();
4307
4308            let reopened = opened(dir.path(), Some(enabled(0.5, 200)));
4309
4310            let after = reopened.table("users").unwrap().stats();
4311            assert_eq!(
4312                after.gen, before.gen,
4313                "the generation is carried by the snapshot, not reset"
4314            );
4315            assert_eq!(after.records, 2);
4316            assert_eq!(after.dead_bytes, 0);
4317            let mut keys: Vec<&String> = reopened.table("users").unwrap().keys().collect();
4318            keys.sort();
4319            assert_eq!(keys, vec!["u1", "u2"]);
4320            assert_eq!(
4321                reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
4322                Some(&json!("Ania 11"))
4323            );
4324            reopened.close().unwrap();
4325        }
4326
4327        #[test]
4328        fn a_partitioned_table_is_measured_and_rewritten_one_partition_at_a_time() {
4329            let dir = tempdir().unwrap();
4330            let schema = parse_schema(&json!({
4331                "schemaVersion": 1,
4332                "table": "tickets",
4333                "primaryKey": "id",
4334                "partitionBy": "kind",
4335                "fields": {
4336                    "id": { "type": "string" },
4337                    "kind": { "type": "string", "required": true },
4338                    "note": { "type": "string" },
4339                },
4340            }))
4341            .unwrap();
4342            write_catalog_to(dir.path(), &catalog_of(&[schema]));
4343            // NOTE: a floor this far above what 12 rewrites of one line make can only be crossed once.
4344            let mut db = opened(dir.path(), Some(enabled(0.5, 700)));
4345
4346            let tickets = db.table_mut("tickets").unwrap();
4347            tickets
4348                .put(&json!({ "id": "r1", "kind": "refund", "note": "n" }), false)
4349                .unwrap();
4350            let refund = dir.path().join("tickets").join("refund.jsonl");
4351            let quiet_size = fs::metadata(&refund).unwrap().len();
4352
4353            tickets
4354                .put(&json!({ "id": "s1", "kind": "sale", "note": "n" }), false)
4355                .unwrap();
4356            for round in 0..12 {
4357                tickets
4358                    .patch("s1", &json!({ "note": format!("n{round}") }), false)
4359                    .unwrap();
4360            }
4361
4362            let sale = dir.path().join("tickets").join("sale.jsonl");
4363            assert!(
4364                lines_of(&sale).len() < 13,
4365                "the partition over its thresholds was rewritten, so not all 13 lines are still there"
4366            );
4367            assert_eq!(
4368                fs::metadata(&refund).unwrap().len(),
4369                quiet_size,
4370                "the partition holding no garbage was left byte for byte"
4371            );
4372            assert_eq!(lines_of(&refund).len(), 1);
4373            let stats = db.table("tickets").unwrap().stats();
4374            assert_eq!(
4375                stats.gen, 1,
4376                "one pass moves the generation once, not once per partition"
4377            );
4378            assert_eq!(stats.records, 2);
4379            db.close().unwrap();
4380        }
4381
4382        #[test]
4383        fn a_catalog_declaring_compaction_false_keeps_the_garbage_an_open_call_named_nothing_about() {
4384            let dir = tempdir().unwrap();
4385            let mut catalog = catalog_of(&[users_schema()]);
4386            catalog.defaults = Some(TableOptions {
4387                compaction: Some(CatalogCompaction::Disabled),
4388                ..Default::default()
4389            });
4390            write_catalog_to(dir.path(), &catalog);
4391            let mut db = opened(dir.path(), None);
4392
4393            churn(&mut db, "users", "u1", 12);
4394
4395            assert_eq!(db.table("users").unwrap().stats().gen, 0);
4396            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
4397            db.close().unwrap();
4398        }
4399
4400        #[test]
4401        fn a_tables_own_entry_beats_the_catalog_defaults_and_an_open_call_beats_them_both() {
4402            let dir = tempdir().unwrap();
4403            let mut catalog = catalog_of(&[users_schema()]);
4404            catalog.defaults = Some(TableOptions {
4405                compaction: Some(CatalogCompaction::Disabled),
4406                ..Default::default()
4407            });
4408            catalog.tables.get_mut("users").unwrap().options = Some(TableOptions {
4409                compaction: Some(enabled(0.5, 200)),
4410                ..Default::default()
4411            });
4412            write_catalog_to(dir.path(), &catalog);
4413
4414            let db = opened(dir.path(), None);
4415            assert_eq!(db.table("users").unwrap().compaction(), enabled(0.5, 200));
4416            db.close().unwrap();
4417
4418            let db = opened(dir.path(), Some(CatalogCompaction::Disabled));
4419            assert_eq!(db.table("users").unwrap().compaction(), CatalogCompaction::Disabled);
4420            db.close().unwrap();
4421        }
4422
4423        #[test]
4424        fn a_table_created_under_a_handle_is_armed_on_the_same_terms_as_one_opened_with_it() {
4425            let dir = tempdir().unwrap();
4426            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4427            db.create_table(
4428                "users",
4429                json!({
4430                    "schemaVersion": 1,
4431                    "table": "users",
4432                    "primaryKey": "id",
4433                    "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
4434                }),
4435            )
4436            .unwrap();
4437
4438            churn(&mut db, "users", "u1", 12);
4439
4440            assert!(db.table("users").unwrap().stats().gen > 0);
4441            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
4442            db.close().unwrap();
4443        }
4444
4445        #[test]
4446        fn an_open_call_naming_nothing_falls_back_to_the_engine_default_thresholds() {
4447            let dir = tempdir().unwrap();
4448            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4449            let db = opened(dir.path(), None);
4450
4451            assert_eq!(db.table("users").unwrap().compaction(), DEFAULT_COMPACTION);
4452            db.close().unwrap();
4453        }
4454    }
4455
4456    /// `OpenOptions::cache` reaching the tables — the read cache itself is pinned in
4457    /// `table::cache` and `table::handle`.
4458    mod read_cache {
4459        use super::*;
4460
4461        fn schema() -> Value {
4462            json!({
4463                "schemaVersion": 1,
4464                "table": "users",
4465                "primaryKey": "id",
4466                "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
4467            })
4468        }
4469
4470        fn opened(dir: &Path, cache: CachePolicy) -> Database {
4471            Database::open_with(
4472                dir,
4473                OpenOptions {
4474                    cache,
4475                    ..Default::default()
4476                },
4477            )
4478            .unwrap()
4479        }
4480
4481        #[test]
4482        fn an_open_that_names_no_cache_leaves_every_table_caching_nothing() {
4483            let dir = tempdir().unwrap();
4484            let mut db = Database::open(dir.path()).unwrap();
4485            db.create_table("users", schema()).unwrap();
4486            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
4487            db.close().unwrap();
4488
4489            let db = Database::open(dir.path()).unwrap();
4490            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
4491            db.close().unwrap();
4492        }
4493
4494        #[test]
4495        fn the_policy_an_open_call_names_arms_the_tables_it_opens_and_the_ones_it_creates() {
4496            let dir = tempdir().unwrap();
4497            let mut db = opened(dir.path(), CachePolicy::with_entries(64));
4498            db.create_table("users", schema()).unwrap();
4499            assert_eq!(
4500                db.table("users").unwrap().cache_policy(),
4501                CachePolicy::with_entries(64),
4502                "a table created under the handle is armed on the same terms"
4503            );
4504            db.close().unwrap();
4505
4506            let db = opened(dir.path(), CachePolicy::with_entries(64));
4507            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::with_entries(64));
4508            db.close().unwrap();
4509        }
4510
4511        #[test]
4512        fn a_cached_table_read_through_the_database_answers_the_same_document_twice() {
4513            let dir = tempdir().unwrap();
4514            let mut db = opened(dir.path(), CachePolicy::with_entries(8));
4515            db.create_table("users", schema()).unwrap();
4516            let users = db.table_mut("users").unwrap();
4517            users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
4518
4519            let first = users.get("u1").unwrap();
4520            let second = users.get("u1").unwrap();
4521
4522            assert_eq!(first, second);
4523            assert_eq!(first.unwrap().get("name"), Some(&json!("Ania")));
4524            db.close().unwrap();
4525        }
4526    }
4527
4528    mod cypher {
4529        use super::*;
4530        use crate::crypto::{CipherOptions, MissingKeyPolicy};
4531        use std::collections::HashMap;
4532
4533        fn key(byte: u8) -> Vec<u8> {
4534            vec![byte; 32]
4535        }
4536
4537        fn secrets_schema_json() -> Value {
4538            json!({
4539                "schemaVersion": 1,
4540                "table": "secrets",
4541                "primaryKey": "id",
4542                "fields": {
4543                    "id": { "type": "string" },
4544                    "token": { "type": "string", "cypher": true },
4545                },
4546            })
4547        }
4548
4549        fn secrets_schema_json_with_note() -> Value {
4550            json!({
4551                "schemaVersion": 1,
4552                "table": "secrets",
4553                "primaryKey": "id",
4554                "fields": {
4555                    "id": { "type": "string" },
4556                    "token": { "type": "string", "cypher": true },
4557                    "note": { "type": "string" },
4558                },
4559            })
4560        }
4561
4562        fn open_with_keys(dir: &Path, keys: &[(&str, u8)], key_id: Option<&str>) -> Database {
4563            Database::open_with(
4564                dir,
4565                OpenOptions {
4566                    cypher: Some(CipherOptions {
4567                        keys: keys.iter().map(|(id, byte)| (id.to_string(), key(*byte))).collect(),
4568                        key_id: key_id.map(str::to_string),
4569                        on_missing_key: None,
4570                    }),
4571                    ..OpenOptions::default()
4572                },
4573            )
4574            .unwrap()
4575        }
4576
4577        #[test]
4578        fn a_cypher_field_round_trips_through_open_created_table_and_reopen() {
4579            let dir = tempdir().unwrap();
4580            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4581            db.create_table("secrets", secrets_schema_json()).unwrap();
4582            db.table_mut("secrets")
4583                .unwrap()
4584                .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4585                .unwrap();
4586            db.close().unwrap();
4587
4588            let reopened = open_with_keys(dir.path(), &[("k1", 1)], None);
4589            assert_eq!(
4590                reopened
4591                    .table("secrets")
4592                    .unwrap()
4593                    .get("s1")
4594                    .unwrap()
4595                    .unwrap()
4596                    .get("token"),
4597                Some(&json!("hunter2"))
4598            );
4599            reopened.close().unwrap();
4600        }
4601
4602        #[test]
4603        fn reads_with_an_old_key_and_writes_with_the_new_one_after_rotation() {
4604            let dir = tempdir().unwrap();
4605            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4606            db.create_table("secrets", secrets_schema_json()).unwrap();
4607            db.table_mut("secrets")
4608                .unwrap()
4609                .put(&json!({ "id": "s1", "token": "written-with-k1" }), false)
4610                .unwrap();
4611            db.close().unwrap();
4612
4613            let mut rotated = open_with_keys(dir.path(), &[("k1", 1), ("k2", 2)], Some("k2"));
4614            assert_eq!(
4615                rotated
4616                    .table("secrets")
4617                    .unwrap()
4618                    .get("s1")
4619                    .unwrap()
4620                    .unwrap()
4621                    .get("token"),
4622                Some(&json!("written-with-k1"))
4623            );
4624            rotated
4625                .table_mut("secrets")
4626                .unwrap()
4627                .put(&json!({ "id": "s2", "token": "written-with-k2" }), false)
4628                .unwrap();
4629            rotated.close().unwrap();
4630
4631            let log = std::fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
4632            assert!(log.contains("$cypher$v1$k1$"));
4633            assert!(log.contains("$cypher$v1$k2$"));
4634        }
4635
4636        #[test]
4637        fn keeps_the_keyring_across_drop_field_reopening_the_table() {
4638            let dir = tempdir().unwrap();
4639            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4640            db.create_table("secrets", secrets_schema_json_with_note()).unwrap();
4641            db.table_mut("secrets")
4642                .unwrap()
4643                .put(&json!({ "id": "s1", "token": "hunter2", "note": "x" }), false)
4644                .unwrap();
4645
4646            db.drop_field("secrets", "note", true).unwrap();
4647            assert_eq!(
4648                db.table("secrets").unwrap().get("s1").unwrap().unwrap().get("token"),
4649                Some(&json!("hunter2"))
4650            );
4651            db.close().unwrap();
4652        }
4653
4654        #[test]
4655        fn open_refuses_a_key_that_is_not_32_bytes() {
4656            let dir = tempdir().unwrap();
4657            let error = Database::open_with(
4658                dir.path(),
4659                OpenOptions {
4660                    cypher: Some(CipherOptions {
4661                        keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
4662                        key_id: None,
4663                        on_missing_key: None,
4664                    }),
4665                    ..OpenOptions::default()
4666                },
4667            )
4668            .map(|_| ())
4669            .unwrap_err();
4670            assert!(matches!(error, DatabaseError::Cipher(_)));
4671        }
4672
4673        #[test]
4674        fn missing_key_policy_raw_hands_back_the_envelope_through_the_database() {
4675            let dir = tempdir().unwrap();
4676            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4677            db.create_table("secrets", secrets_schema_json()).unwrap();
4678            db.table_mut("secrets")
4679                .unwrap()
4680                .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4681                .unwrap();
4682            db.close().unwrap();
4683
4684            let raw = Database::open_with(
4685                dir.path(),
4686                OpenOptions {
4687                    cypher: Some(CipherOptions {
4688                        keys: HashMap::from([("other".to_string(), key(2))]),
4689                        key_id: None,
4690                        on_missing_key: Some(MissingKeyPolicy::Raw),
4691                    }),
4692                    ..OpenOptions::default()
4693                },
4694            )
4695            .unwrap();
4696            let found = raw.table("secrets").unwrap().get("s1").unwrap().unwrap();
4697            assert!(found.get("token").unwrap().as_str().unwrap().contains("$cypher$v1$k1$"));
4698            raw.close().unwrap();
4699        }
4700    }
4701
4702    mod partitions {
4703        use super::*;
4704
4705        fn notes_schema_json(partitioned: bool) -> Value {
4706            let mut spec = json!({
4707                "schemaVersion": 1,
4708                "table": "notes",
4709                "primaryKey": "id",
4710                "fields": {
4711                    "id": { "type": "string" },
4712                    "kind": { "type": "string", "required": true },
4713                    "title": { "type": "string", "required": true },
4714                    "body": { "type": "string" },
4715                },
4716            });
4717            if partitioned {
4718                spec.as_object_mut()
4719                    .unwrap()
4720                    .insert("partitionBy".into(), json!("kind"));
4721            }
4722            spec
4723        }
4724
4725        fn lines_in(dir: &Path, part: &str) -> usize {
4726            fs::read_to_string(dir.join("notes").join(format!("{part}.jsonl")))
4727                .unwrap()
4728                .trim()
4729                .lines()
4730                .filter(|line| !line.is_empty())
4731                .count()
4732        }
4733
4734        #[test]
4735        fn a_partitioned_table_survives_the_catalog_round_trip() {
4736            let dir = tempdir().unwrap();
4737            {
4738                let mut db = Database::open(dir.path()).unwrap();
4739                db.create_table("notes", notes_schema_json(true)).unwrap();
4740                let notes = db.table_mut("notes").unwrap();
4741                notes
4742                    .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4743                    .unwrap();
4744                notes
4745                    .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4746                    .unwrap();
4747                db.close().unwrap();
4748            }
4749
4750            let db = Database::open(dir.path()).unwrap();
4751            let notes = db.table("notes").unwrap();
4752            assert_eq!(notes.schema().partition_by.as_deref(), Some("kind"));
4753            assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4754            assert_eq!(lines_in(dir.path(), "note"), 1);
4755            assert_eq!(lines_in(dir.path(), "adr"), 1);
4756            assert!(!dir.path().join("notes.jsonl").exists());
4757            db.close().unwrap();
4758        }
4759
4760        #[test]
4761        fn declaring_partitions_on_an_existing_table_moves_its_records_into_the_new_layout() {
4762            let dir = tempdir().unwrap();
4763            {
4764                let mut db = Database::open(dir.path()).unwrap();
4765                db.create_table("notes", notes_schema_json(false)).unwrap();
4766                let notes = db.table_mut("notes").unwrap();
4767                notes
4768                    .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4769                    .unwrap();
4770                notes
4771                    .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4772                    .unwrap();
4773                notes.delete("n1", false).unwrap();
4774                db.close().unwrap();
4775            }
4776
4777            let mut db = Database::open(dir.path()).unwrap();
4778            let mut partitioned = notes_schema_json(true);
4779            partitioned.as_object_mut().unwrap().insert("revision".into(), json!(2));
4780            db.create_table("notes", partitioned).unwrap();
4781
4782            let notes = db.table("notes").unwrap();
4783            assert_eq!(notes.get("n1").unwrap(), None);
4784            assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4785            assert!(!dir.path().join("notes.jsonl").exists());
4786            assert_eq!(lines_in(dir.path(), "adr"), 1);
4787            db.close().unwrap();
4788        }
4789
4790        #[test]
4791        fn dropping_the_field_the_table_partitions_by_is_refused() {
4792            let dir = tempdir().unwrap();
4793            let mut db = Database::open(dir.path()).unwrap();
4794            db.create_table("notes", notes_schema_json(true)).unwrap();
4795            db.table_mut("notes")
4796                .unwrap()
4797                .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4798                .unwrap();
4799
4800            let error = db.drop_field("notes", "kind", true).unwrap_err();
4801            assert!(error.to_string().contains("partitioned by"));
4802            assert_eq!(
4803                db.table("notes").unwrap().get("n1").unwrap().unwrap().get("kind"),
4804                Some(&json!("note"))
4805            );
4806            db.close().unwrap();
4807        }
4808
4809        #[test]
4810        fn dropping_an_unrelated_field_keeps_every_partition_readable() {
4811            let dir = tempdir().unwrap();
4812            let mut db = Database::open(dir.path()).unwrap();
4813            db.create_table("notes", notes_schema_json(true)).unwrap();
4814            {
4815                let notes = db.table_mut("notes").unwrap();
4816                notes
4817                    .put(
4818                        &json!({ "id": "n1", "kind": "note", "title": "first", "body": "gone soon" }),
4819                        false,
4820                    )
4821                    .unwrap();
4822                notes
4823                    .put(&json!({ "id": "a1", "kind": "adr", "title": "kept" }), false)
4824                    .unwrap();
4825            }
4826
4827            db.drop_field("notes", "body", true).unwrap();
4828
4829            let notes = db.table("notes").unwrap();
4830            assert!(!notes.schema().fields.contains_key("body"));
4831            assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("first")));
4832            assert_eq!(notes.get("a1").unwrap().unwrap().get("kind"), Some(&json!("adr")));
4833            assert_eq!(lines_in(dir.path(), "note"), 1);
4834            assert_eq!(lines_in(dir.path(), "adr"), 1);
4835            db.close().unwrap();
4836        }
4837
4838        #[test]
4839        fn a_unit_of_work_lands_in_two_partitions_of_one_table() {
4840            let dir = tempdir().unwrap();
4841            let mut db = Database::open(dir.path()).unwrap();
4842            db.create_table("notes", notes_schema_json(true)).unwrap();
4843
4844            let mut unit = db.unit_of_work();
4845            unit.put("notes", json!({ "id": "n1", "kind": "note", "title": "from unit" }));
4846            unit.put("notes", json!({ "id": "a1", "kind": "adr", "title": "from unit too" }));
4847            let result = db.commit(&mut unit).unwrap();
4848            assert_eq!(result.operations, 2);
4849
4850            let notes = db.table("notes").unwrap();
4851            assert_eq!(
4852                notes.get("n1").unwrap().unwrap().get("title"),
4853                Some(&json!("from unit"))
4854            );
4855            assert_eq!(
4856                notes.get("a1").unwrap().unwrap().get("title"),
4857                Some(&json!("from unit too"))
4858            );
4859            assert_eq!(lines_in(dir.path(), "note"), 1);
4860            assert_eq!(lines_in(dir.path(), "adr"), 1);
4861            db.close().unwrap();
4862        }
4863    }
4864    mod following_the_log {
4865        use super::*;
4866
4867        fn readonly(follow: FollowOption) -> OpenOptions {
4868            OpenOptions {
4869                readonly: true,
4870                follow,
4871                ..Default::default()
4872            }
4873        }
4874
4875        /// A writer and a reader on the same directory, as two separate handles — the shape of the
4876        /// TS `multiprocess.test.ts`, which runs the writer in a child process for the lock's sake.
4877        /// One process is enough here: the lock is only taken by the writable handle.
4878        fn writer_and_reader(dir: &Path, follow: FollowOption) -> (Database, Database) {
4879            let mut writer = Database::open(dir).unwrap();
4880            writer.create_table("users", users_schema_spec()).unwrap();
4881            writer
4882                .table_mut("users")
4883                .unwrap()
4884                .put(&json!({ "id": "u1", "name": "Ania" }), false)
4885                .unwrap();
4886            let reader = Database::open_with(dir, readonly(follow)).unwrap();
4887            (writer, reader)
4888        }
4889
4890        fn users_schema_spec() -> Value {
4891            json!({
4892                "schemaVersion": 1,
4893                "table": "users",
4894                "primaryKey": "id",
4895                "fields": {
4896                    "id": { "type": "string" },
4897                    "name": { "type": "string", "required": true },
4898                },
4899            })
4900        }
4901
4902        fn put(writer: &mut Database, key: &str, name: &str) {
4903            writer
4904                .table_mut("users")
4905                .unwrap()
4906                .put(&json!({ "id": key, "name": name }), false)
4907                .unwrap();
4908        }
4909
4910        #[test]
4911        fn a_readonly_open_follows_by_default_and_a_writable_one_does_not() {
4912            let dir = tempdir().unwrap();
4913            let mut writer = Database::open(dir.path()).unwrap();
4914            writer.create_table("users", users_schema_spec()).unwrap();
4915
4916            assert_eq!(writer.follow(), None);
4917            assert_eq!(writer.table("users").unwrap().follow(), None);
4918
4919            let reader = Database::open_with(dir.path(), readonly(FollowOption::Default)).unwrap();
4920            assert_eq!(
4921                reader.follow(),
4922                Some(FollowPolicy {
4923                    interval: DEFAULT_FOLLOW_INTERVAL
4924                })
4925            );
4926            assert_eq!(
4927                reader.table("users").unwrap().follow(),
4928                Some(FollowPolicy {
4929                    interval: DEFAULT_FOLLOW_INTERVAL
4930                })
4931            );
4932        }
4933
4934        #[test]
4935        fn a_plain_get_on_a_following_reader_sees_what_the_writer_committed_after_it_opened() {
4936            let dir = tempdir().unwrap();
4937            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4938
4939            for index in 2..=5 {
4940                put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
4941                let found = reader
4942                    .table_mut("users")
4943                    .unwrap()
4944                    .get_following(&format!("u{index}"))
4945                    .unwrap();
4946                assert_eq!(found.unwrap().get("name"), Some(&json!(format!("Osoba {index}"))));
4947            }
4948        }
4949
4950        #[test]
4951        fn a_plain_query_on_a_following_reader_counts_what_the_writer_committed() {
4952            let dir = tempdir().unwrap();
4953            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4954
4955            let spec = QueryBuilder::new().spec();
4956            assert_eq!(
4957                reader
4958                    .table_mut("users")
4959                    .unwrap()
4960                    .query_following(&spec)
4961                    .unwrap()
4962                    .matched,
4963                1
4964            );
4965
4966            put(&mut writer, "u2", "Osoba 2");
4967
4968            assert_eq!(
4969                reader
4970                    .table_mut("users")
4971                    .unwrap()
4972                    .query_following(&spec)
4973                    .unwrap()
4974                    .matched,
4975                2
4976            );
4977        }
4978
4979        #[test]
4980        fn a_reader_with_follow_off_stands_still_until_refresh_is_called() {
4981            let dir = tempdir().unwrap();
4982            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Off);
4983            put(&mut writer, "u2", "Osoba 2");
4984
4985            assert!(reader
4986                .table_mut("users")
4987                .unwrap()
4988                .get_following("u2")
4989                .unwrap()
4990                .is_none());
4991            assert!(reader.table_mut("users").unwrap().refresh().unwrap().changed());
4992            assert!(reader
4993                .table_mut("users")
4994                .unwrap()
4995                .get_following("u2")
4996                .unwrap()
4997                .is_some());
4998        }
4999
5000        #[test]
5001        fn a_custom_interval_coalesces_the_freshness_check_and_lets_it_through_once_it_elapses() {
5002            let dir = tempdir().unwrap();
5003            let (mut writer, mut reader) =
5004                writer_and_reader(dir.path(), FollowOption::Every(Duration::from_millis(150)));
5005
5006            // The first read is always due — nothing has been followed yet — and starts the clock.
5007            assert!(reader
5008                .table_mut("users")
5009                .unwrap()
5010                .get_following("u1")
5011                .unwrap()
5012                .is_some());
5013            put(&mut writer, "u2", "Osoba 2");
5014            assert!(reader
5015                .table_mut("users")
5016                .unwrap()
5017                .get_following("u2")
5018                .unwrap()
5019                .is_none());
5020
5021            std::thread::sleep(Duration::from_millis(200));
5022
5023            assert!(reader
5024                .table_mut("users")
5025                .unwrap()
5026                .get_following("u2")
5027                .unwrap()
5028                .is_some());
5029        }
5030
5031        #[test]
5032        fn get_many_and_the_aggregates_follow_the_log_the_same_way_a_single_get_does() {
5033            let dir = tempdir().unwrap();
5034            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
5035            put(&mut writer, "u2", "Osoba 2");
5036
5037            let keys = vec!["u1".to_string(), "u2".to_string()];
5038            let rows = reader.table_mut("users").unwrap().get_many_following(&keys).unwrap();
5039            assert!(rows.iter().all(Option::is_some), "{rows:?}");
5040
5041            put(&mut writer, "u3", "Osoba 3");
5042            let spec = QueryBuilder::new().spec();
5043            let aggregate = crate::query::aggregate::AggregateSpec {
5044                count: true,
5045                ..Default::default()
5046            };
5047            let counted = reader
5048                .table_mut("users")
5049                .unwrap()
5050                .aggregate_following(&spec, &aggregate)
5051                .unwrap();
5052            assert_eq!(counted.count, 3);
5053        }
5054
5055        #[test]
5056        fn a_following_reader_reloads_instead_of_reading_the_offsets_a_foreign_compaction_moved() {
5057            let dir = tempdir().unwrap();
5058            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
5059            for index in 2..=6 {
5060                put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
5061            }
5062            writer.table_mut("users").unwrap().delete("u5", false).unwrap();
5063            assert!(reader
5064                .table_mut("users")
5065                .unwrap()
5066                .get_following("u6")
5067                .unwrap()
5068                .is_some());
5069
5070            writer.table_mut("users").unwrap().compact().unwrap();
5071
5072            let users = reader.table_mut("users").unwrap();
5073            assert_eq!(
5074                users.get_following("u3").unwrap().unwrap().get("name"),
5075                Some(&json!("Osoba 3"))
5076            );
5077            assert!(users.get_following("u5").unwrap().is_none());
5078            assert_eq!(users.query_following(&QueryBuilder::new().spec()).unwrap().matched, 5);
5079        }
5080    }
5081
5082    mod where_the_files_land {
5083        use super::*;
5084        use crate::database::LayoutKind;
5085
5086        fn notes_schema_spec() -> Value {
5087            json!({
5088                "schemaVersion": 1,
5089                "table": "notes",
5090                "primaryKey": "id",
5091                "partitionBy": "kind",
5092                "fields": {
5093                    "id": { "type": "string" },
5094                    "kind": { "type": "string", "required": true, "index": "exact" },
5095                    "title": { "type": "string", "required": true },
5096                },
5097            })
5098        }
5099
5100        fn nested(metadata_file: Option<&str>) -> OpenOptions {
5101            OpenOptions {
5102                layout: LayoutOptions {
5103                    kind: Some(LayoutKind::Nested),
5104                    metadata_file: metadata_file.map(str::to_string),
5105                    ..Default::default()
5106                },
5107                ..Default::default()
5108            }
5109        }
5110
5111        #[test]
5112        fn the_default_open_writes_exactly_what_it_wrote_before_layouts_existed() {
5113            let dir = tempdir().unwrap();
5114            let mut db = Database::open(dir.path()).unwrap();
5115            db.create_table(
5116                "users",
5117                json!({
5118                    "schemaVersion": 1,
5119                    "table": "users",
5120                    "primaryKey": "id",
5121                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5122                }),
5123            )
5124            .unwrap();
5125            db.table_mut("users")
5126                .unwrap()
5127                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5128                .unwrap();
5129            db.close().unwrap();
5130
5131            assert!(dir.path().join("hayson.json").is_file());
5132            assert!(dir.path().join("users.jsonl").is_file());
5133            assert!(!dir.path().join("tables").exists());
5134            assert!(!dir.path().join("runtime").exists());
5135
5136            let db = Database::open(dir.path()).unwrap();
5137            assert_eq!(db.layout().kind(), LayoutKind::Flat);
5138            assert_eq!(db.layout().metadata_file(), "hayson.json");
5139            assert_eq!(db.layout().metadata_path(), dir.path().join("hayson.json"));
5140        }
5141
5142        #[test]
5143        fn a_custom_metadata_file_holds_the_catalog_and_nothing_is_written_under_the_default_name() {
5144            let dir = tempdir().unwrap();
5145            let options = OpenOptions {
5146                layout: LayoutOptions {
5147                    metadata_file: Some("cerebro.json".into()),
5148                    ..Default::default()
5149                },
5150                ..Default::default()
5151            };
5152
5153            let mut db = Database::open_with(dir.path(), options.clone()).unwrap();
5154            db.create_table(
5155                "users",
5156                json!({
5157                    "schemaVersion": 1,
5158                    "table": "users",
5159                    "primaryKey": "id",
5160                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5161                }),
5162            )
5163            .unwrap();
5164            db.table_mut("users")
5165                .unwrap()
5166                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5167                .unwrap();
5168            db.close().unwrap();
5169
5170            assert!(dir.path().join("cerebro.json").is_file());
5171            assert!(!dir.path().join("hayson.json").exists());
5172
5173            let db = Database::open_with(dir.path(), options).unwrap();
5174            assert_eq!(db.layout().metadata_file(), "cerebro.json");
5175            assert_eq!(db.table_names(), vec!["users"]);
5176            assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5177        }
5178
5179        #[test]
5180        fn the_deprecated_manifest_name_opens_what_the_new_one_wrote_and_the_other_way_round() {
5181            let dir = tempdir().unwrap();
5182            let legacy = OpenOptions {
5183                layout: LayoutOptions {
5184                    manifest: Some("cerebro.json".into()),
5185                    ..Default::default()
5186                },
5187                ..Default::default()
5188            };
5189            let mut db = Database::open_with(dir.path(), legacy).unwrap();
5190            db.create_table(
5191                "users",
5192                json!({
5193                    "schemaVersion": 1,
5194                    "table": "users",
5195                    "primaryKey": "id",
5196                    "fields": { "id": { "type": "string" } },
5197                }),
5198            )
5199            .unwrap();
5200            db.close().unwrap();
5201
5202            let current = OpenOptions {
5203                layout: LayoutOptions {
5204                    metadata_file: Some("cerebro.json".into()),
5205                    ..Default::default()
5206                },
5207                ..Default::default()
5208            };
5209            let db = Database::open_with(dir.path(), current).unwrap();
5210            assert_eq!(db.table_names(), vec!["users"]);
5211            assert_eq!(db.layout().metadata_path(), dir.path().join("cerebro.json"));
5212        }
5213
5214        #[test]
5215        fn the_two_names_for_the_metadata_document_are_refused_when_they_disagree_touching_nothing() {
5216            let dir = tempdir().unwrap();
5217            let options = OpenOptions {
5218                layout: LayoutOptions {
5219                    metadata_file: Some("cerebro.json".into()),
5220                    manifest: Some("hayson.json".into()),
5221                    ..Default::default()
5222                },
5223                ..Default::default()
5224            };
5225
5226            let Err(error) = Database::open_with(dir.path(), options) else {
5227                panic!("two disagreeing names for the metadata document were accepted")
5228            };
5229
5230            assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
5231            let message = error.to_string();
5232            assert!(
5233                message.contains("metadataFile") && message.contains("manifest"),
5234                "{message}"
5235            );
5236            assert!(!dir.path().join("hayson.json").exists());
5237            assert!(!dir.path().join("cerebro.json").exists());
5238            assert!(!dir.path().join(crate::lock::LOCK_FILE).exists());
5239        }
5240
5241        #[test]
5242        fn the_nested_layout_sorts_the_logs_the_snapshots_and_the_lock_into_their_subdirectories() {
5243            let dir = tempdir().unwrap();
5244            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5245            db.create_table(
5246                "users",
5247                json!({
5248                    "schemaVersion": 1,
5249                    "table": "users",
5250                    "primaryKey": "id",
5251                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5252                }),
5253            )
5254            .unwrap();
5255            db.table_mut("users")
5256                .unwrap()
5257                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5258                .unwrap();
5259
5260            assert!(dir.path().join("runtime").join(crate::lock::LOCK_FILE).is_file());
5261            db.close().unwrap();
5262
5263            assert!(dir.path().join("hayson.json").is_file());
5264            assert!(dir.path().join("tables").join("users.jsonl").is_file());
5265            assert!(dir.path().join("index").join("users.idx").is_file());
5266            assert!(!dir.path().join("users.jsonl").exists());
5267        }
5268
5269        #[test]
5270        fn a_partitioned_table_under_the_nested_layout_lands_under_tables_slash_its_own_name() {
5271            let dir = tempdir().unwrap();
5272            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5273            db.create_table("notes", notes_schema_spec()).unwrap();
5274            db.table_mut("notes")
5275                .unwrap()
5276                .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
5277                .unwrap();
5278            db.close().unwrap();
5279
5280            assert!(dir.path().join("tables").join("notes").join("note.jsonl").is_file());
5281        }
5282
5283        #[test]
5284        fn a_nested_database_reopens_as_nested_without_being_told_and_is_refused_as_flat() {
5285            let dir = tempdir().unwrap();
5286            let mut db = Database::open_with(dir.path(), nested(Some("cerebro.json"))).unwrap();
5287            db.create_table(
5288                "users",
5289                json!({
5290                    "schemaVersion": 1,
5291                    "table": "users",
5292                    "primaryKey": "id",
5293                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5294                }),
5295            )
5296            .unwrap();
5297            db.table_mut("users")
5298                .unwrap()
5299                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5300                .unwrap();
5301            db.close().unwrap();
5302
5303            let named_only = OpenOptions {
5304                layout: LayoutOptions {
5305                    metadata_file: Some("cerebro.json".into()),
5306                    ..Default::default()
5307                },
5308                ..Default::default()
5309            };
5310            let db = Database::open_with(dir.path(), named_only).unwrap();
5311            assert_eq!(db.layout().kind(), LayoutKind::Nested);
5312            assert_eq!(
5313                db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
5314                Some(&json!("Ania"))
5315            );
5316            db.close().unwrap();
5317
5318            let as_flat = OpenOptions {
5319                layout: LayoutOptions {
5320                    kind: Some(LayoutKind::Flat),
5321                    metadata_file: Some("cerebro.json".into()),
5322                    ..Default::default()
5323                },
5324                ..Default::default()
5325            };
5326            let Err(error) = Database::open_with(dir.path(), as_flat) else {
5327                panic!("a nested database was opened as flat")
5328            };
5329            assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
5330            assert!(error.to_string().contains("nested"), "{error}");
5331        }
5332
5333        #[test]
5334        fn a_unit_of_work_under_the_nested_layout_journals_into_runtime_and_reopens_whole() {
5335            let dir = tempdir().unwrap();
5336            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5337            db.create_table(
5338                "users",
5339                json!({
5340                    "schemaVersion": 1,
5341                    "table": "users",
5342                    "primaryKey": "id",
5343                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5344                }),
5345            )
5346            .unwrap();
5347
5348            let mut unit = db.unit_of_work();
5349            unit.put("users", json!({ "id": "u1", "name": "Ania" }));
5350            unit.put("users", json!({ "id": "u2", "name": "Bartek" }));
5351            db.commit(&mut unit).unwrap();
5352            db.close().unwrap();
5353
5354            assert!(!dir.path().join("runtime").join(".hayson.txn").exists());
5355            let db = Database::open_with(dir.path(), nested(None)).unwrap();
5356            assert_eq!(db.table("users").unwrap().len(), 2);
5357        }
5358    }
5359
5360    /// F8: a catalog naming a table whose log is not on disk. A writable handle makes the log it
5361    /// misses, so the two open modes owe the caller opposite answers here.
5362    mod a_declared_table_whose_log_is_not_on_disk {
5363        use super::*;
5364
5365        fn notes_spec() -> Value {
5366            json!({
5367                "schemaVersion": 1,
5368                "table": "notes",
5369                "primaryKey": "id",
5370                "partitionBy": "kind",
5371                "fields": {
5372                    "id": { "type": "string" },
5373                    "kind": { "type": "string", "required": true },
5374                    "title": { "type": "string", "required": true },
5375                },
5376            })
5377        }
5378
5379        fn readonly() -> OpenOptions {
5380            OpenOptions {
5381                readonly: true,
5382                ..Default::default()
5383            }
5384        }
5385
5386        fn users_with_a_record(dir: &Path) {
5387            let mut db = Database::open(dir).unwrap();
5388            db.create_table(
5389                "users",
5390                json!({
5391                    "schemaVersion": 1,
5392                    "table": "users",
5393                    "primaryKey": "id",
5394                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5395                }),
5396            )
5397            .unwrap();
5398            db.table_mut("users")
5399                .unwrap()
5400                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5401                .unwrap();
5402            db.close().unwrap();
5403        }
5404
5405        fn notes_with_a_record(dir: &Path) {
5406            let mut db = Database::open(dir).unwrap();
5407            db.create_table("notes", notes_spec()).unwrap();
5408            db.table_mut("notes")
5409                .unwrap()
5410                .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
5411                .unwrap();
5412            db.close().unwrap();
5413        }
5414
5415        fn refusal(dir: &Path) -> DatabaseError {
5416            match Database::open_with(dir, readonly()) {
5417                Err(error) => error,
5418                Ok(_) => panic!("expected \"{}\" to be refused", dir.display()),
5419            }
5420        }
5421
5422        fn schema_refusal(error: DatabaseError) -> String {
5423            match error {
5424                DatabaseError::Schema(HaysonSchemaError(message)) => message,
5425                other => panic!("expected a schema refusal, got {other:?}"),
5426            }
5427        }
5428
5429        #[test]
5430        fn a_readonly_open_refuses_naming_the_table_and_the_log_without_advising_create_table() {
5431            let dir = tempdir().unwrap();
5432            users_with_a_record(dir.path());
5433            let log = log_path(dir.path(), "users");
5434            fs::remove_file(&log).unwrap();
5435
5436            let message = schema_refusal(refusal(dir.path()));
5437
5438            assert!(message.contains("\"users\""), "{message}");
5439            assert!(message.contains(&log.display().to_string()), "{message}");
5440            assert!(message.contains("readonly"), "{message}");
5441            assert!(!message.contains("createTable"), "{message}");
5442            assert!(!log.exists(), "the refused open must not have created the log");
5443        }
5444
5445        #[test]
5446        fn a_writable_open_of_the_same_directory_creates_the_log_and_serves_an_empty_table() {
5447            let dir = tempdir().unwrap();
5448            users_with_a_record(dir.path());
5449            fs::remove_file(log_path(dir.path(), "users")).unwrap();
5450
5451            let mut db = Database::open(dir.path()).unwrap();
5452
5453            assert_eq!(db.table_names(), vec!["users"]);
5454            assert!(log_path(dir.path(), "users").is_file());
5455            assert_eq!(db.table("users").unwrap().len(), 0);
5456            db.table_mut("users")
5457                .unwrap()
5458                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5459                .unwrap();
5460            assert_eq!(db.table("users").unwrap().len(), 1);
5461            db.close().unwrap();
5462        }
5463
5464        // NOTE: an empty log is a table nothing was written to; a missing one is a table gone.
5465        #[test]
5466        fn a_readonly_open_accepts_a_log_that_is_present_and_empty() {
5467            let dir = tempdir().unwrap();
5468            users_with_a_record(dir.path());
5469            fs::write(log_path(dir.path(), "users"), "").unwrap();
5470
5471            let db = Database::open_with(dir.path(), readonly()).unwrap();
5472
5473            assert_eq!(db.table_names(), vec!["users"]);
5474            assert_eq!(db.table("users").unwrap().len(), 0);
5475        }
5476
5477        #[test]
5478        fn a_readonly_open_refuses_a_partitioned_table_whose_recorded_partition_is_gone() {
5479            let dir = tempdir().unwrap();
5480            notes_with_a_record(dir.path());
5481            let partition = dir.path().join("notes").join("note.jsonl");
5482            assert!(
5483                index_path(dir.path(), "notes").is_file(),
5484                "the refusal reads the snapshot"
5485            );
5486            fs::remove_dir_all(dir.path().join("notes")).unwrap();
5487
5488            let message = schema_refusal(refusal(dir.path()));
5489
5490            assert!(message.contains("\"notes\""), "{message}");
5491            assert!(message.contains(&partition.display().to_string()), "{message}");
5492            assert!(message.contains("readonly"), "{message}");
5493            assert!(!message.contains("createTable"), "{message}");
5494        }
5495
5496        #[test]
5497        fn a_readonly_open_accepts_a_partitioned_table_that_was_never_written_to() {
5498            let dir = tempdir().unwrap();
5499            let mut db = Database::open(dir.path()).unwrap();
5500            db.create_table("notes", notes_spec()).unwrap();
5501            db.close().unwrap();
5502
5503            let db = Database::open_with(dir.path(), readonly()).unwrap();
5504
5505            assert_eq!(db.table_names(), vec!["notes"]);
5506            assert_eq!(db.table("notes").unwrap().len(), 0);
5507        }
5508
5509        /// A partition the listing shows but the filesystem cannot deliver is corruption, not a
5510        /// table that opens empty — and the log set would otherwise create a fresh file over it.
5511        #[cfg(unix)]
5512        #[test]
5513        fn a_readonly_open_reports_a_partition_that_cannot_be_opened_against_its_own_path() {
5514            let dir = tempdir().unwrap();
5515            notes_with_a_record(dir.path());
5516            let partition = dir.path().join("notes").join("note.jsonl");
5517            fs::remove_file(&partition).unwrap();
5518            std::os::unix::fs::symlink(dir.path().join("notes").join("gone.jsonl"), &partition).unwrap();
5519
5520            let error = refusal(dir.path());
5521
5522            match error {
5523                DatabaseError::Log(path, inner) => {
5524                    assert_eq!(path, partition);
5525                    assert_eq!(inner.kind(), io::ErrorKind::NotFound);
5526                }
5527                other => panic!("expected a log refusal naming the partition, got {other:?}"),
5528            }
5529            assert!(!dir.path().join("notes").join("gone.jsonl").exists());
5530        }
5531    }
5532
5533    mod cascading_delete {
5534        use super::*;
5535
5536        fn orders_schema_optional_user() -> TableSchema {
5537            parse_schema(&json!({
5538                "schemaVersion": 1,
5539                "table": "orders",
5540                "primaryKey": "id",
5541                "fields": {
5542                    "id": { "type": "string" },
5543                    "userId": { "type": "string" },
5544                },
5545            }))
5546            .unwrap()
5547        }
5548
5549        fn order_items_schema() -> TableSchema {
5550            parse_schema(&json!({
5551                "schemaVersion": 1,
5552                "table": "order_items",
5553                "primaryKey": "id",
5554                "fields": {
5555                    "id": { "type": "string" },
5556                    "orderId": { "type": "string", "required": true },
5557                },
5558            }))
5559            .unwrap()
5560        }
5561
5562        fn orders_user_relation_on_delete(action: &str) -> Value {
5563            json!({
5564                "name": "orders_user",
5565                "from": { "table": "orders", "field": "userId" },
5566                "to": { "table": "users", "field": "id" },
5567                "cardinality": "many-to-one",
5568                "onDelete": action,
5569            })
5570        }
5571
5572        fn order_items_order_relation_cascade() -> Value {
5573            json!({
5574                "name": "order_items_order",
5575                "from": { "table": "order_items", "field": "orderId" },
5576                "to": { "table": "orders", "field": "id" },
5577                "cardinality": "many-to-one",
5578                "onDelete": "cascade",
5579            })
5580        }
5581
5582        #[test]
5583        fn a_table_no_relation_names_as_to_goes_straight_through_plain_delete() {
5584            let dir = tempdir().unwrap();
5585            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
5586            let mut db = Database::open(dir.path()).unwrap();
5587            db.table_mut("users")
5588                .unwrap()
5589                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5590                .unwrap();
5591
5592            let deleted = db.delete_checked("users", "u1", false).unwrap();
5593
5594            assert!(deleted);
5595            assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5596            db.close().unwrap();
5597        }
5598
5599        #[test]
5600        fn a_missing_key_answers_false_without_staging_anything() {
5601            let dir = tempdir().unwrap();
5602            write_catalog_to(
5603                dir.path(),
5604                &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5605            );
5606            let mut db = Database::open(dir.path()).unwrap();
5607            db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5608                .unwrap();
5609
5610            let deleted = db.delete_checked("users", "ghost", false).unwrap();
5611
5612            assert!(!deleted);
5613            db.close().unwrap();
5614        }
5615
5616        #[test]
5617        fn restrict_refuses_when_a_referencing_record_exists_and_leaves_everything_untouched() {
5618            let dir = tempdir().unwrap();
5619            write_catalog_to(
5620                dir.path(),
5621                &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5622            );
5623            let mut db = Database::open(dir.path()).unwrap();
5624            db.declare_relation(orders_user_relation_on_delete("restrict"), false)
5625                .unwrap();
5626            db.table_mut("users")
5627                .unwrap()
5628                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5629                .unwrap();
5630            db.table_mut("orders")
5631                .unwrap()
5632                .put(&json!({ "id": "o1", "userId": "u1" }), false)
5633                .unwrap();
5634
5635            let error = db.delete_checked("users", "u1", false).unwrap_err();
5636
5637            match error {
5638                DatabaseError::Table(table, TableError::RelationRestrict(inner)) => {
5639                    assert_eq!(table, "users");
5640                    assert_eq!(inner.relation, "orders_user");
5641                    assert_eq!(inner.referencing_table, "orders");
5642                    assert_eq!(inner.referencing, 1);
5643                }
5644                other => panic!("expected a RelationRestrict refusal, got {other:?}"),
5645            }
5646            assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5647            assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
5648            db.close().unwrap();
5649        }
5650
5651        #[test]
5652        fn restrict_allows_the_delete_once_nothing_references_it() {
5653            let dir = tempdir().unwrap();
5654            write_catalog_to(
5655                dir.path(),
5656                &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5657            );
5658            let mut db = Database::open(dir.path()).unwrap();
5659            db.declare_relation(orders_user_relation_on_delete("restrict"), false)
5660                .unwrap();
5661            db.table_mut("users")
5662                .unwrap()
5663                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5664                .unwrap();
5665
5666            let deleted = db.delete_checked("users", "u1", false).unwrap();
5667
5668            assert!(deleted);
5669            db.close().unwrap();
5670        }
5671
5672        #[test]
5673        fn cascade_deletes_every_referencing_record() {
5674            let dir = tempdir().unwrap();
5675            write_catalog_to(
5676                dir.path(),
5677                &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5678            );
5679            let mut db = Database::open(dir.path()).unwrap();
5680            db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5681                .unwrap();
5682            db.table_mut("users")
5683                .unwrap()
5684                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5685                .unwrap();
5686            db.table_mut("orders")
5687                .unwrap()
5688                .put(&json!({ "id": "o1", "userId": "u1" }), false)
5689                .unwrap();
5690            db.table_mut("orders")
5691                .unwrap()
5692                .put(&json!({ "id": "o2", "userId": "u1" }), false)
5693                .unwrap();
5694            db.table_mut("orders")
5695                .unwrap()
5696                .put(&json!({ "id": "o3", "userId": "other" }), false)
5697                .unwrap();
5698
5699            let deleted = db.delete_checked("users", "u1", false).unwrap();
5700
5701            assert!(deleted);
5702            assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5703            assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
5704            assert!(db.table("orders").unwrap().get("o2").unwrap().is_none());
5705            assert!(db.table("orders").unwrap().get("o3").unwrap().is_some());
5706            db.close().unwrap();
5707        }
5708
5709        #[test]
5710        fn cascade_recurses_through_a_chain_of_relations() {
5711            let dir = tempdir().unwrap();
5712            write_catalog_to(
5713                dir.path(),
5714                &catalog_of(&[users_schema(), orders_schema_optional_user(), order_items_schema()]),
5715            );
5716            let mut db = Database::open(dir.path()).unwrap();
5717            db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5718                .unwrap();
5719            db.declare_relation(order_items_order_relation_cascade(), false)
5720                .unwrap();
5721            db.table_mut("users")
5722                .unwrap()
5723                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5724                .unwrap();
5725            db.table_mut("orders")
5726                .unwrap()
5727                .put(&json!({ "id": "o1", "userId": "u1" }), false)
5728                .unwrap();
5729            db.table_mut("order_items")
5730                .unwrap()
5731                .put(&json!({ "id": "i1", "orderId": "o1" }), false)
5732                .unwrap();
5733
5734            let deleted = db.delete_checked("users", "u1", false).unwrap();
5735
5736            assert!(deleted);
5737            assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
5738            assert!(db.table("order_items").unwrap().get("i1").unwrap().is_none());
5739            db.close().unwrap();
5740        }
5741
5742        #[test]
5743        fn set_null_clears_the_referencing_field_instead_of_deleting_the_row() {
5744            let dir = tempdir().unwrap();
5745            write_catalog_to(
5746                dir.path(),
5747                &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5748            );
5749            let mut db = Database::open(dir.path()).unwrap();
5750            db.declare_relation(orders_user_relation_on_delete("set-null"), false)
5751                .unwrap();
5752            db.table_mut("users")
5753                .unwrap()
5754                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5755                .unwrap();
5756            db.table_mut("orders")
5757                .unwrap()
5758                .put(&json!({ "id": "o1", "userId": "u1" }), false)
5759                .unwrap();
5760
5761            let deleted = db.delete_checked("users", "u1", false).unwrap();
5762
5763            assert!(deleted);
5764            assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5765            let order = db.table("orders").unwrap().get("o1").unwrap().unwrap();
5766            assert_eq!(order.get("userId"), Some(&Value::Null));
5767            db.close().unwrap();
5768        }
5769
5770        #[test]
5771        fn no_action_and_undeclared_leave_referencing_records_untouched() {
5772            let dir = tempdir().unwrap();
5773            write_catalog_to(
5774                dir.path(),
5775                &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5776            );
5777            let mut db = Database::open(dir.path()).unwrap();
5778            db.declare_relation(orders_user_relation_on_delete("no-action"), false)
5779                .unwrap();
5780            db.table_mut("users")
5781                .unwrap()
5782                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5783                .unwrap();
5784            db.table_mut("orders")
5785                .unwrap()
5786                .put(&json!({ "id": "o1", "userId": "u1" }), false)
5787                .unwrap();
5788
5789            let deleted = db.delete_checked("users", "u1", false).unwrap();
5790
5791            assert!(deleted);
5792            let order = db.table("orders").unwrap().get("o1").unwrap().unwrap();
5793            assert_eq!(order.get("userId"), Some(&Value::String("u1".to_string())));
5794            db.close().unwrap();
5795        }
5796
5797        #[test]
5798        fn a_restrict_deep_in_the_cascade_aborts_the_whole_delete_atomically() {
5799            let dir = tempdir().unwrap();
5800            write_catalog_to(
5801                dir.path(),
5802                &catalog_of(&[users_schema(), orders_schema_optional_user(), order_items_schema()]),
5803            );
5804            let mut db = Database::open(dir.path()).unwrap();
5805            db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5806                .unwrap();
5807            db.declare_relation(
5808                json!({
5809                    "name": "order_items_order",
5810                    "from": { "table": "order_items", "field": "orderId" },
5811                    "to": { "table": "orders", "field": "id" },
5812                    "cardinality": "many-to-one",
5813                    "onDelete": "restrict",
5814                }),
5815                false,
5816            )
5817            .unwrap();
5818            db.table_mut("users")
5819                .unwrap()
5820                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5821                .unwrap();
5822            db.table_mut("orders")
5823                .unwrap()
5824                .put(&json!({ "id": "o1", "userId": "u1" }), false)
5825                .unwrap();
5826            db.table_mut("order_items")
5827                .unwrap()
5828                .put(&json!({ "id": "i1", "orderId": "o1" }), false)
5829                .unwrap();
5830
5831            let error = db.delete_checked("users", "u1", false).unwrap_err();
5832
5833            assert!(matches!(
5834                error,
5835                DatabaseError::Table(_, TableError::RelationRestrict(_))
5836            ));
5837            assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5838            assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
5839            assert!(db.table("order_items").unwrap().get("i1").unwrap().is_some());
5840            db.close().unwrap();
5841        }
5842
5843        #[test]
5844        fn a_cycle_of_cascading_relations_does_not_recurse_forever() {
5845            let dir = tempdir().unwrap();
5846            let a_schema = parse_schema(&json!({
5847                "schemaVersion": 1,
5848                "table": "a",
5849                "primaryKey": "id",
5850                "fields": { "id": { "type": "string" }, "bId": { "type": "string" } },
5851            }))
5852            .unwrap();
5853            let b_schema = parse_schema(&json!({
5854                "schemaVersion": 1,
5855                "table": "b",
5856                "primaryKey": "id",
5857                "fields": { "id": { "type": "string" }, "aId": { "type": "string" } },
5858            }))
5859            .unwrap();
5860            write_catalog_to(dir.path(), &catalog_of(&[a_schema, b_schema]));
5861            let mut db = Database::open(dir.path()).unwrap();
5862            db.declare_relation(
5863                json!({
5864                    "name": "a_b",
5865                    "from": { "table": "a", "field": "bId" },
5866                    "to": { "table": "b", "field": "id" },
5867                    "cardinality": "many-to-one",
5868                    "onDelete": "cascade",
5869                }),
5870                false,
5871            )
5872            .unwrap();
5873            db.declare_relation(
5874                json!({
5875                    "name": "b_a",
5876                    "from": { "table": "b", "field": "aId" },
5877                    "to": { "table": "a", "field": "id" },
5878                    "cardinality": "many-to-one",
5879                    "onDelete": "cascade",
5880                }),
5881                false,
5882            )
5883            .unwrap();
5884            db.table_mut("a")
5885                .unwrap()
5886                .put(&json!({ "id": "a1", "bId": "b1" }), false)
5887                .unwrap();
5888            db.table_mut("b")
5889                .unwrap()
5890                .put(&json!({ "id": "b1", "aId": "a1" }), false)
5891                .unwrap();
5892
5893            let deleted = db.delete_checked("a", "a1", false).unwrap();
5894
5895            assert!(deleted);
5896            assert!(db.table("a").unwrap().get("a1").unwrap().is_none());
5897            assert!(db.table("b").unwrap().get("b1").unwrap().is_none());
5898            db.close().unwrap();
5899        }
5900
5901        #[test]
5902        fn delete_checked_refuses_on_a_readonly_handle() {
5903            let dir = tempdir().unwrap();
5904            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
5905            {
5906                let mut db = Database::open(dir.path()).unwrap();
5907                db.table_mut("users")
5908                    .unwrap()
5909                    .put(&json!({ "id": "u1", "name": "Ania" }), false)
5910                    .unwrap();
5911                db.close().unwrap();
5912            }
5913            let mut db = Database::open_with(
5914                dir.path(),
5915                OpenOptions {
5916                    readonly: true,
5917                    ..OpenOptions::default()
5918                },
5919            )
5920            .unwrap();
5921
5922            let error = db.delete_checked("users", "u1", false).unwrap_err();
5923
5924            assert!(matches!(error, DatabaseError::ReadOnly { .. }));
5925        }
5926    }
5927
5928    mod describing_one_table {
5929        use super::*;
5930
5931        fn users_with_city() -> TableSchema {
5932            parse_schema(&json!({
5933                "schemaVersion": 1,
5934                "table": "users",
5935                "revision": 2,
5936                "primaryKey": "id",
5937                "fields": {
5938                    "id": { "type": "string" },
5939                    "name": { "type": "string", "required": true },
5940                    "city": { "type": "string", "index": "exact" },
5941                },
5942            }))
5943            .unwrap()
5944        }
5945
5946        /// A handle over both tables, with the catalog on disk moved on under it for `users`
5947        /// alone — the drift every gate here has to tell apart from a healthy `orders`.
5948        fn users_drifted_under(dir: &Path) -> Database {
5949            write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
5950            let mut db = Database::open(dir).unwrap();
5951            write_catalog_to(dir, &catalog_of(&[users_with_city(), orders_schema()]));
5952            assert!(db.refresh().unwrap());
5953            db
5954        }
5955
5956        fn refusal_of(answered: Result<Value, DatabaseError>) -> String {
5957            match answered {
5958                Err(DatabaseError::Schema(error)) => error.to_string(),
5959                other => panic!("expected a schema refusal, got {other:?}"),
5960            }
5961        }
5962
5963        #[test]
5964        fn describes_a_table_of_its_own_while_another_one_has_drifted() {
5965            let dir = tempdir().unwrap();
5966            let db = users_drifted_under(dir.path());
5967
5968            let document = db.describe_table("orders").unwrap();
5969
5970            assert_eq!(document["table"], json!("orders"));
5971            assert_eq!(document["revision"], json!(1));
5972            assert!(db.catalog().is_err());
5973        }
5974
5975        #[test]
5976        fn refuses_the_table_that_drifted_itself() {
5977            let dir = tempdir().unwrap();
5978            let db = users_drifted_under(dir.path());
5979
5980            let message = refusal_of(db.describe_table("users"));
5981
5982            assert!(message.contains("\"users\""), "{message}");
5983            assert!(message.contains("revision 2"), "{message}");
5984            // NOTE: refusing replaces the reload, so the stale handle stays stale.
5985            assert_eq!(db.table("users").unwrap().schema().revision, 1);
5986        }
5987
5988        #[test]
5989        fn describes_the_index_set_the_handle_serves_and_leaves_the_named_ones_out() {
5990            let dir = tempdir().unwrap();
5991            let mut catalog = catalog_of(&[users_schema()]);
5992            let entry = catalog.tables.get_mut("users").unwrap();
5993            entry.indexes = Some(vec![IndexEntry {
5994                name: "by_name".to_string(),
5995                fields: vec!["name".to_string()],
5996                kind: crate::catalog::CatalogIndexKind::Exact,
5997                unique: false,
5998                where_: None,
5999                expire_after_seconds: None,
6000            }]);
6001            write_catalog_to(dir.path(), &catalog);
6002            let db = Database::open(dir.path()).unwrap();
6003
6004            let document = db.describe_table("users").unwrap();
6005
6006            assert_eq!(
6007                document["fields"]["name"],
6008                json!({ "type": "string", "required": true })
6009            );
6010            assert!(document.as_object().unwrap().get("indexes").is_none());
6011            assert!(db.catalog().is_ok());
6012        }
6013
6014        #[test]
6015        fn refuses_a_table_this_handle_holds_nowhere() {
6016            let dir = tempdir().unwrap();
6017            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
6018            let db = Database::open(dir.path()).unwrap();
6019
6020            let message = refusal_of(db.describe_table("ghosts"));
6021
6022            assert!(message.contains("declares no table \"ghosts\""), "{message}");
6023        }
6024    }
6025}