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