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