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, TableEntry, TableOptions, TriggerEntry,
15};
16use crate::checkpoint::{read_snapshot, RestoreOutcome};
17use crate::crypto::{CipherOptions, FieldCipher};
18use crate::errors::{HaysonCipherError, 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    /// Renews the writer lease from outside — [`DirLock::tick`] reached through the lock this
1046    /// handle owns privately. A host without background threads (`wasm32-wasip1`) has no heartbeat
1047    /// of its own, so this is what a timer on its side drives; call it at least twice per
1048    /// `LockOptions::heartbeat_ms`. Does nothing on a readonly handle, which holds no lock.
1049    pub fn tick(&self) -> Result<(), DatabaseError> {
1050        self.renew_lock()
1051    }
1052
1053    fn renew_lock(&self) -> Result<(), DatabaseError> {
1054        match &self.lock {
1055            Some(lock) => Ok(lock.tick()?),
1056            None => Ok(()),
1057        }
1058    }
1059
1060    /// How the table named `name` was brought up at open — `Restored` from `<table>.idx` or
1061    /// `Rebuilt` by a full scan of its log. `None` when the database holds no such table.
1062    pub fn restore_outcome(&self, name: &str) -> Option<RestoreOutcome> {
1063        self.restore_outcomes.get(name).copied()
1064    }
1065
1066    /// The metadata catalog as this handle last read it, without the drift gate — every table it
1067    /// describes may be one this handle does not serve. `catalog` is what a caller outside the
1068    /// engine should ask for.
1069    pub fn catalog_unchecked(&self) -> &Catalog {
1070        &self.catalog
1071    }
1072
1073    /// The metadata catalog as this handle last read it — the parsed form of `hayson.json`, which
1074    /// `catalog::catalog_document` turns back into the document a caller can serialize.
1075    ///
1076    /// Refuses when any table open here has drifted from what the catalog now declares, because
1077    /// handing the document over would describe tables this handle does not serve. Refusing is all
1078    /// that happens: the handle is never reloaded behind the caller's back, so the schema and the
1079    /// index set every open table was opened with stay exactly as they were.
1080    pub fn catalog(&self) -> Result<&Catalog, DatabaseError> {
1081        let drifted = self.drifted_tables();
1082        if drifted.is_empty() {
1083            return Ok(&self.catalog);
1084        }
1085        Err(DatabaseError::Schema(self.drift_error(&drifted)))
1086    }
1087
1088    /// The schema of one table as a standalone document, the shape `create_table` accepts — what
1089    /// the TS `describeTable` answers with.
1090    ///
1091    /// Read off the live handle (`table::describe_table_document`), not off the catalog entry, so
1092    /// it describes the schema and the index set the table is actually being served with — every
1093    /// table the catalog declares is open here, both after `open` and after
1094    /// [`Database::refresh`], so there is no catalog-only table to fall back to.
1095    ///
1096    /// The drift gate here is the one table named, unlike [`Database::catalog`], whose document
1097    /// covers every table and so has to answer for every table: a table of its own that moved on
1098    /// says nothing about this one.
1099    pub fn describe_table(&self, name: &str) -> Result<Value, DatabaseError> {
1100        if let Some(drift) = self.drift_of(name) {
1101            let drifted = [DriftedTable {
1102                name: name.to_string(),
1103                drift,
1104            }];
1105            return Err(DatabaseError::Schema(self.drift_error(&drifted)));
1106        }
1107        let table = self.tables.get(name).ok_or_else(|| {
1108            DatabaseError::Schema(HaysonSchemaError(format!("the catalog declares no table \"{name}\"")))
1109        })?;
1110        Ok(describe_table_document(table))
1111    }
1112
1113    fn drift_of(&self, name: &str) -> Option<TableDrift> {
1114        let table = self.tables.get(name)?;
1115        let Some(entry) = self.catalog.tables.get(name) else {
1116            return Some(TableDrift::Absent);
1117        };
1118        if shape_of(&table_schema_of(name, entry)) != shape_of(table.schema()) {
1119            return Some(TableDrift::Schema {
1120                revision: entry.revision,
1121            });
1122        }
1123        let declared = expand_indexes(name, entry);
1124        let served = table.declared_indexes();
1125        let unserved = index_names_missing_from(&declared, served);
1126        let stale = index_names_missing_from(served, &declared);
1127        if unserved.is_empty() && stale.is_empty() {
1128            return None;
1129        }
1130        Some(TableDrift::Indexes { unserved, stale })
1131    }
1132
1133    fn drifted_tables(&self) -> Vec<DriftedTable> {
1134        let mut names: Vec<&String> = self.tables.keys().collect();
1135        names.sort();
1136        names
1137            .into_iter()
1138            .filter_map(|name| {
1139                self.drift_of(name).map(|drift| DriftedTable {
1140                    name: name.clone(),
1141                    drift,
1142                })
1143            })
1144            .collect()
1145    }
1146
1147    fn drift_error(&self, drifted: &[DriftedTable]) -> HaysonSchemaError {
1148        let detail = drifted
1149            .iter()
1150            .map(|DriftedTable { name, drift }| {
1151                let revision = self.tables.get(name).map(|table| table.schema().revision);
1152                let open = revision.map_or_else(|| "not open".to_string(), |at| format!("revision {at}"));
1153                format!(
1154                    "\"{name}\" is open here at {open} while the catalog {}",
1155                    stored_side_of(drift)
1156                )
1157            })
1158            .collect::<Vec<_>>()
1159            .join("; ");
1160        HaysonSchemaError(format!(
1161            "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",
1162            self.layout.root().display()
1163        ))
1164    }
1165
1166    /// The relations the catalog declares.
1167    ///
1168    /// @param table When given, keeps only the relations touching this table, on either end.
1169    pub fn relations(&self, table: Option<&str>) -> Vec<Relation> {
1170        relations_touching(&self.catalog, table)
1171    }
1172
1173    /// Walks every declared relation and reports the records whose reference has no counterpart —
1174    /// a diagnosis, not a gate; nothing is repaired, removed or written.
1175    pub fn check_integrity(&self) -> io::Result<IntegrityReport> {
1176        let refs: HashMap<String, &Table> = self.tables.iter().map(|(name, table)| (name.clone(), table)).collect();
1177        check_integrity(&self.catalog, &refs)
1178    }
1179
1180    /// Catches up with everything other processes did to this directory since this handle last
1181    /// looked — the port of the TS `Database.refresh()`, in the same two steps and the same order:
1182    ///
1183    /// 1. `hayson.json` is read again, and every table it now declares that this handle does not
1184    ///    hold is opened, through the very code path `open` uses per table;
1185    /// 2. every table already open here is caught up through [`Table::refresh`].
1186    ///
1187    /// Answers whether anything at all changed. A catalog that lost a table on disk does not close
1188    /// the one already open here: a live table is not dropped out from under a caller holding it.
1189    pub fn refresh(&mut self) -> Result<bool, DatabaseError> {
1190        let mut changed = false;
1191        if let Some(catalog) = read_catalog(self.layout.metadata_path())? {
1192            if catalog != self.catalog {
1193                self.catalog = catalog;
1194                changed = true;
1195            }
1196        }
1197
1198        let missing: Vec<String> = self
1199            .catalog
1200            .tables
1201            .keys()
1202            .filter(|name| !self.tables.contains_key(*name))
1203            .cloned()
1204            .collect();
1205        for name in missing {
1206            let Some(entry) = self.catalog.tables.get(&name) else {
1207                continue;
1208            };
1209            let (table, outcome) = open_catalog_table(
1210                &self.layout,
1211                &name,
1212                &self.catalog,
1213                entry,
1214                self.snapshot_every,
1215                self.cipher.as_ref(),
1216                self.follow,
1217                self.readonly,
1218                compaction_of(self.compaction, &self.catalog, Some(entry)),
1219                self.cache,
1220            )?;
1221            self.tables.insert(name.clone(), table);
1222            self.restore_outcomes.insert(name, outcome);
1223            changed = true;
1224        }
1225
1226        for (name, table) in self.tables.iter_mut() {
1227            let outcome = table
1228                .refresh()
1229                .map_err(|error| DatabaseError::Table(name.clone(), TableError::Io(error)))?;
1230            changed |= outcome.changed();
1231        }
1232
1233        // NOTE: triggers are resynced after the catalog re-read — a table opened above has none yet.
1234        sync_table_triggers(&self.catalog, &mut self.tables);
1235        Ok(changed)
1236    }
1237
1238    /// Starts a unit of work: stage `put`/`patch`/`delete` across several tables, then pass it to
1239    /// [`Database::commit`], which lands it whole or not at all.
1240    pub fn unit_of_work(&mut self) -> UnitOfWork {
1241        UnitOfWork::new()
1242    }
1243
1244    /// Runs every operation staged in `unit`, across the tables it names, through the six-step
1245    /// commit `txn::UnitOfWork::commit` documents (resolve, barrier, prepare, journal, write,
1246    /// cleanup) — refused when the database is readonly.
1247    ///
1248    /// The unit is borrowed, not consumed: a commit refused before anything was journalled leaves
1249    /// it staged and retriable, exactly as the TS engine leaves it.
1250    pub fn commit(&mut self, unit: &mut UnitOfWork) -> Result<UnitOfWorkResult, DatabaseError> {
1251        let prepared = self.prepare_commit(unit)?;
1252        unit.journal(&prepared, &self.layout.journal_path())?;
1253        self.apply_commit(prepared)
1254    }
1255
1256    /// The first phase of [`Database::commit`] — the readonly refusal, the lock renewal and
1257    /// [`UnitOfWork::prepare`]. Nothing is appended and no file is written, so a refusal from here
1258    /// leaves `unit` staged and retriable exactly as a refused `commit` does.
1259    ///
1260    /// Split out for a caller that keeps the database behind a lock and wants to release it across
1261    /// the journal write, which touches no table: prepare here, [`UnitOfWork::journal`] with the
1262    /// database let go, then [`Database::apply_commit`] once it is held again. The caller must stay
1263    /// the only writer across all three, or the burned `_seq` numbers would land out of order.
1264    pub fn prepare_commit(&mut self, unit: &mut UnitOfWork) -> Result<PreparedUnit, DatabaseError> {
1265        if self.readonly {
1266            return Err(DatabaseError::ReadOnly {
1267                dir: self.layout.root().to_path_buf(),
1268                operation: "commit".to_string(),
1269            });
1270        }
1271        self.renew_lock()?;
1272        Ok(unit.prepare(&mut self.tables)?)
1273    }
1274
1275    /// The last phase of [`Database::commit`]: every prepared entry appended and the journal
1276    /// removed. See [`Database::prepare_commit`] for the contract a caller running the phases apart
1277    /// has to keep.
1278    pub fn apply_commit(&mut self, prepared: PreparedUnit) -> Result<UnitOfWorkResult, DatabaseError> {
1279        let journal_path = self.layout.journal_path();
1280        Ok(apply_prepared(prepared, &mut self.tables, &journal_path)?)
1281    }
1282
1283    /// Refuses the operation when this handle is not the writer, and renews the lease while it is
1284    /// — so a lock stolen out from under a live handle stops the next write instead of letting it
1285    /// through into a directory somebody else now owns.
1286    fn assert_writable(&self, operation: &str) -> Result<(), DatabaseError> {
1287        if self.readonly {
1288            return Err(DatabaseError::ReadOnly {
1289                dir: self.layout.root().to_path_buf(),
1290                operation: operation.to_string(),
1291            });
1292        }
1293        self.renew_lock()
1294    }
1295
1296    fn write_catalog_to_disk(&self, catalog: &Catalog) -> Result<(), DatabaseError> {
1297        write_catalog(self.layout.metadata_path(), catalog).map_err(catalog_write_error)
1298    }
1299
1300    /// Creates a table, or brings an existing one up to a newer (additive) schema — `catalog::build`
1301    /// does not check revision compatibility yet (see its own doc comment), so a genuinely breaking
1302    /// change is not refused here either; that check is still on the TODO list this method inherits.
1303    ///
1304    /// Disk first: the catalog is written before `self.catalog`/`self.tables` change, so a failure
1305    /// past that point is the only way this call can leave the two out of step with each other.
1306    ///
1307    /// The handle comes from `open_catalog_table` — the same helper `open()`/`refresh()` use — so a
1308    /// table migrated here carries exactly what it would carry after a restart, triggers included.
1309    ///
1310    /// `name` is the key the table lands under in the catalog, and the schema's own `table` is
1311    /// carried into every log line and index snapshot written for it. The two disagreeing is
1312    /// refused rather than resolved: the catalog key wins wherever they are read back, so accepting
1313    /// the spec would silently drop the name its author wrote.
1314    pub fn create_table(&mut self, name: &str, schema_spec: Value) -> Result<(), DatabaseError> {
1315        self.assert_writable("createTable")?;
1316        let schema = parse_schema(&schema_spec)?;
1317        if schema.table != name {
1318            return Err(DatabaseError::Schema(HaysonSchemaError(format!(
1319                "schema declares table \"{}\" but it is being created as \"{name}\" — name them the same",
1320                schema.table
1321            ))));
1322        }
1323        let now = now_ms();
1324        let next_catalog = crate::catalog::build::create_table(&self.catalog, name, &schema, now);
1325        self.write_catalog_to_disk(&next_catalog)?;
1326
1327        let next_entry = &next_catalog.tables[name];
1328        let (table, outcome) = open_catalog_table(
1329            &self.layout,
1330            name,
1331            &next_catalog,
1332            next_entry,
1333            self.snapshot_every,
1334            self.cipher.as_ref(),
1335            self.follow,
1336            self.readonly,
1337            compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1338            self.cache,
1339        )?;
1340
1341        self.catalog = next_catalog;
1342        self.tables.insert(name.to_string(), table);
1343        self.restore_outcomes.insert(name.to_string(), outcome);
1344        Ok(())
1345    }
1346
1347    /// Takes a field off the schema of a table without touching its log — records written before
1348    /// the drop keep carrying the field on disk, reads just stop handing it back.
1349    ///
1350    /// Refuses (leaving disk and `self` untouched) when the field is the primary key, is what the
1351    /// table partitions by, or an index/check/relation of the catalog still depends on it
1352    /// (`catalog::build::drop_field`, unconditional on `force`) — and, unless `force` is `true`,
1353    /// when a live record still holds a value for it (checked here, against the open table, per the
1354    /// TODO `catalog::build::drop_field` leaves for this layer).
1355    ///
1356    /// The declaration check runs first on purpose: with both refusals due the caller is told about
1357    /// the one `force` cannot get them past, and the error class matches what the field's
1358    /// dependents are — see `packages/hayson/test/drop-pola-kolejnosc-odmowy.test.ts`.
1359    pub fn drop_field(&mut self, table: &str, field: &str, force: bool) -> Result<(), DatabaseError> {
1360        self.assert_writable("dropField")?;
1361        self.flush_table(table)?;
1362        let existing = self.tables.get(table).ok_or_else(|| {
1363            DatabaseError::Catalog(format!(
1364                "table \"{table}\" does not exist in \"{}\"",
1365                self.layout.root().display()
1366            ))
1367        })?;
1368
1369        let now = now_ms();
1370        let (next_catalog, next_entry) = crate::catalog::build::drop_field(&self.catalog, table, field, now)?;
1371
1372        if !force {
1373            assert_no_residual_values(table, existing, field)?;
1374        }
1375
1376        self.write_catalog_to_disk(&next_catalog)?;
1377
1378        let (mut table_handle, _) = open_catalog_table(
1379            &self.layout,
1380            table,
1381            &next_catalog,
1382            &next_entry,
1383            self.snapshot_every,
1384            self.cipher.as_ref(),
1385            self.follow,
1386            self.readonly,
1387            compaction_of(self.compaction, &next_catalog, Some(&next_entry)),
1388            self.cache,
1389        )?;
1390        // NOTE: the .idx on disk still projects the dropped field — rewrite it before anyone restores from it.
1391        table_handle
1392            .snapshot()
1393            .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1394
1395        self.catalog = next_catalog;
1396        self.tables.insert(table.to_string(), table_handle);
1397        Ok(())
1398    }
1399
1400    /// Adds a named index to a table of the catalog — a unique index is refused whole if a live
1401    /// record already shares a value on it (`catalog::declare_index`, checked against the open
1402    /// table). The reopened table's registry is built from the catalog's full expanded index list
1403    /// (`catalog::expand_indexes`), so a composite or `where`-scoped index is enforced/served right
1404    /// away, same as the single-field unconditional case `catalog::expand_indexes` derives as sugar.
1405    ///
1406    /// Reopened through `open_catalog_table` — the same helper `open()`/`refresh()` use — so the
1407    /// handle keeps its snapshot pairing, and the stale `.idx` is rewritten before this returns.
1408    pub fn declare_index(&mut self, table: &str, index_spec: Value) -> Result<(), DatabaseError> {
1409        self.assert_writable("declareIndex")?;
1410        self.flush_table(table)?;
1411        let existing = self.tables.get(table).ok_or_else(|| {
1412            DatabaseError::Catalog(format!(
1413                "table \"{table}\" does not exist in \"{}\"",
1414                self.layout.root().display()
1415            ))
1416        })?;
1417
1418        let now = now_ms();
1419        let next_catalog = crate::catalog::declare_index(&self.catalog, table, &index_spec, existing, now)?;
1420        self.write_catalog_to_disk(&next_catalog)?;
1421
1422        let next_entry = &next_catalog.tables[table];
1423        let (mut table_handle, _) = open_catalog_table(
1424            &self.layout,
1425            table,
1426            &next_catalog,
1427            next_entry,
1428            self.snapshot_every,
1429            self.cipher.as_ref(),
1430            self.follow,
1431            self.readonly,
1432            compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1433            self.cache,
1434        )?;
1435        // NOTE: the .idx on disk projects nothing of the new index — rewrite it before anyone restores from it.
1436        table_handle
1437            .snapshot()
1438            .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1439
1440        self.catalog = next_catalog;
1441        self.tables.insert(table.to_string(), table_handle);
1442        Ok(())
1443    }
1444
1445    /// Adds a named relation to the catalog (`catalog::declare_relation`). With `enforce: true`,
1446    /// refuses whole if a live record already breaks it — unless `force` is `true`, which declares
1447    /// it anyway and lets enforcement start from here on, existing orphans left as they are.
1448    ///
1449    /// Unlike `declare_index`, nothing here reopens a table: a relation is not read by a table's own
1450    /// registry the way an index is, so no live handle changes shape. `from`/`to` are read straight
1451    /// out of `relation_spec` first, ahead of the real parse inside `catalog::declare_relation`, only
1452    /// to know which two tables to flush and hand it — a malformed spec still gets the full,
1453    /// authoritative error from the parser, this is not validation.
1454    pub fn declare_relation(&mut self, relation_spec: Value, force: bool) -> Result<(), DatabaseError> {
1455        self.assert_writable("declareRelation")?;
1456
1457        let from_table = relation_spec
1458            .get("from")
1459            .and_then(|end| end.get("table"))
1460            .and_then(Value::as_str)
1461            .ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"from.table\"".to_string()))?
1462            .to_string();
1463        let to_table = relation_spec
1464            .get("to")
1465            .and_then(|end| end.get("table"))
1466            .and_then(Value::as_str)
1467            .ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"to.table\"".to_string()))?
1468            .to_string();
1469
1470        self.flush_table(&from_table)?;
1471        self.flush_table(&to_table)?;
1472        let from_handle = self.tables.get(&from_table).ok_or_else(|| {
1473            DatabaseError::Catalog(format!(
1474                "table \"{from_table}\" does not exist in \"{}\"",
1475                self.layout.root().display()
1476            ))
1477        })?;
1478        let to_handle = self.tables.get(&to_table).ok_or_else(|| {
1479            DatabaseError::Catalog(format!(
1480                "table \"{to_table}\" does not exist in \"{}\"",
1481                self.layout.root().display()
1482            ))
1483        })?;
1484
1485        let now = now_ms();
1486        let next_catalog = declare_relation_in(&self.catalog, &relation_spec, from_handle, to_handle, now, force)?;
1487        self.write_catalog_to_disk(&next_catalog)?;
1488        self.catalog = next_catalog;
1489        Ok(())
1490    }
1491
1492    /// Declares an AFTER trigger (`catalog::build::declare_trigger`, which refuses a spec naming a
1493    /// table the catalog does not declare) and refreshes every open table's live trigger set from
1494    /// the result — both the source table `trigger_spec.table` names and the effect table
1495    /// `trigger_spec.effect.table` names, since either could be a live table this call did not
1496    /// itself just open. Disk first, same discipline as `create_table`/`declare_index`.
1497    pub fn declare_trigger(&mut self, trigger_spec: Value) -> Result<(), DatabaseError> {
1498        self.assert_writable("declareTrigger")?;
1499
1500        let now = now_ms();
1501        let next_catalog = crate::catalog::declare_trigger(&self.catalog, &trigger_spec, now)?;
1502        self.write_catalog_to_disk(&next_catalog)?;
1503
1504        self.catalog = next_catalog;
1505        sync_table_triggers(&self.catalog, &mut self.tables);
1506        Ok(())
1507    }
1508
1509    /// Appends every write staged against any table of this database and not yet committed — the
1510    /// TS `Database.flush()`. A no-op for a table with nothing in flight, which is every table of a
1511    /// database whose writes all went through `put`/`patch`/`delete`.
1512    pub fn flush(&mut self) -> Result<(), DatabaseError> {
1513        for (name, table) in &mut self.tables {
1514            table
1515                .flush()
1516                .map_err(|error| DatabaseError::Table(name.clone(), error))?;
1517        }
1518        Ok(())
1519    }
1520
1521    /// Commits what is staged against one table. Every path that reopens a table over its log, or
1522    /// rewrites that log, has to run this first: the handle it replaces takes its staging window
1523    /// with it.
1524    fn flush_table(&mut self, table: &str) -> Result<(), DatabaseError> {
1525        let Some(handle) = self.tables.get_mut(table) else {
1526            return Ok(());
1527        };
1528        handle
1529            .flush()
1530            .map_err(|error| DatabaseError::Table(table.to_string(), error))
1531    }
1532
1533    /// Appends whatever every table still holds staged, snapshots the ones with writes since their
1534    /// last snapshot (see [`Table::needs_snapshot`]), then releases the writer lock, if this handle
1535    /// holds one — mirrors the TS `Database.close()`. The catalog needs no flush of its own: it is
1536    /// already durable (`fsync`) by the time the call that wrote it returns.
1537    pub fn close(mut self) -> io::Result<()> {
1538        if !self.readonly {
1539            for table in self.tables.values_mut() {
1540                // NOTE: flush before the snapshot — a staged write is in neither the log nor it.
1541                table.flush()?;
1542                if table.needs_snapshot() {
1543                    table.snapshot()?;
1544                }
1545            }
1546        }
1547        if let Some(mut lock) = self.lock.take() {
1548            lock.release();
1549        }
1550        Ok(())
1551    }
1552}
1553
1554#[cfg(test)]
1555mod tests {
1556    use serde_json::json;
1557    use tempfile::tempdir;
1558
1559    use super::*;
1560    use crate::catalog::create_table;
1561    use crate::catalog::types::{CatalogIndexKind, IndexEntry};
1562    use crate::checkpoint::write_snapshot;
1563    use crate::registry::table_registry::TableRegistry;
1564    use crate::schema::parse_schema;
1565    use crate::storage::{serialize_entry, LogMeta, LogOp};
1566    use crate::txn::journal::{write_journal, Journal, JournalEntry, JournalHeader, JOURNAL_VERSION};
1567    use crate::txn::JOURNAL_FILE;
1568
1569    const MANIFEST_FILE: &str = DEFAULT_METADATA_FILE;
1570
1571    fn flat(dir: &Path) -> Layout {
1572        Layout::new(dir, crate::database::LayoutKind::Flat, MANIFEST_FILE)
1573    }
1574
1575    fn log_path(dir: &Path, table: &str) -> std::path::PathBuf {
1576        flat(dir).log_path(table)
1577    }
1578
1579    /// The binding above this crate keeps the database behind an `RwLock` shared across the libuv
1580    /// pool, which needs `Sync` as well as `Send`. Nothing here runs: the call fails to compile the
1581    /// moment an interior-mutability cell that is not thread-safe reappears anywhere below.
1582    #[test]
1583    fn a_database_and_its_tables_can_be_shared_across_threads() {
1584        fn shareable<T: Send + Sync>() {}
1585        shareable::<Database>();
1586        shareable::<Table>();
1587        shareable::<crate::registry::fulltext_index::FulltextIndex>();
1588    }
1589
1590    fn index_path(dir: &Path, table: &str) -> std::path::PathBuf {
1591        flat(dir).index_path(table)
1592    }
1593
1594    fn users_schema() -> TableSchema {
1595        parse_schema(&json!({
1596            "schemaVersion": 1,
1597            "table": "users",
1598            "primaryKey": "id",
1599            "fields": {
1600                "id": { "type": "string" },
1601                "name": { "type": "string", "required": true },
1602            },
1603        }))
1604        .unwrap()
1605    }
1606
1607    fn orders_schema() -> TableSchema {
1608        parse_schema(&json!({
1609            "schemaVersion": 1,
1610            "table": "orders",
1611            "primaryKey": "id",
1612            "fields": {
1613                "id": { "type": "string" },
1614                "userId": { "type": "string", "required": true },
1615            },
1616        }))
1617        .unwrap()
1618    }
1619
1620    fn catalog_of(schemas: &[TableSchema]) -> Catalog {
1621        let mut catalog = empty_catalog(CatalogLayout::Flat, 1_000);
1622        for schema in schemas {
1623            catalog = create_table(&catalog, &schema.table, schema, 1_000);
1624        }
1625        catalog
1626    }
1627
1628    fn write_catalog_to(dir: &Path, catalog: &Catalog) {
1629        write_catalog(&dir.join(MANIFEST_FILE), catalog).unwrap();
1630    }
1631
1632    #[test]
1633    fn open_builds_a_database_with_every_declared_table_available() {
1634        let dir = tempdir().unwrap();
1635        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1636
1637        let mut db = Database::open(dir.path()).unwrap();
1638        assert_eq!(db.table_names(), vec!["orders", "users"]);
1639
1640        db.table_mut("users")
1641            .unwrap()
1642            .put(&json!({ "id": "u1", "name": "Ania" }), false)
1643            .unwrap();
1644        assert_eq!(
1645            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1646            Some(&json!("Ania"))
1647        );
1648        db.close().unwrap();
1649    }
1650
1651    #[test]
1652    fn open_restores_the_registry_from_a_paired_snapshot_instead_of_scanning_the_log() {
1653        let dir = tempdir().unwrap();
1654        let catalog = catalog_of(&[users_schema()]);
1655        write_catalog_to(dir.path(), &catalog);
1656
1657        let schema = table_schema_of("users", &catalog.tables["users"]);
1658        let hash = schema_fingerprint(&schema, &expand_indexes("users", &catalog.tables["users"]));
1659        let log_path = log_path(dir.path(), "users");
1660        let mut log = crate::storage::LogFile::open(&log_path, false).unwrap();
1661        log.append(&[serialize_entry(
1662            &LogMeta {
1663                key: "u1".into(),
1664                op: LogOp::Put,
1665                seq: 1,
1666                ts: 1,
1667            },
1668            Some(&{
1669                let mut record = crate::schema::HaysonRecord::new();
1670                record.insert("id".into(), json!("u1"));
1671                record.insert("name".into(), json!("Ania"));
1672                record
1673            }),
1674        )])
1675        .unwrap();
1676        let registry = TableRegistry::build_from_log(schema.clone(), &log).unwrap();
1677        write_snapshot(&index_path(dir.path(), "users"), &registry, "users", &hash, 0, &log).unwrap();
1678
1679        let db = Database::open(dir.path()).unwrap();
1680        assert_eq!(db.restore_outcome("users"), Some(RestoreOutcome::Restored));
1681        assert_eq!(
1682            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1683            Some(&json!("Ania"))
1684        );
1685        db.close().unwrap();
1686    }
1687
1688    #[test]
1689    fn open_finishes_an_unfinished_journal_before_returning() {
1690        let dir = tempdir().unwrap();
1691        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1692
1693        {
1694            let mut users = Table::open(users_schema(), log_path(dir.path(), "users")).unwrap();
1695            let mut orders = Table::open(orders_schema(), log_path(dir.path(), "orders")).unwrap();
1696            let user_entries = users
1697                .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
1698                .unwrap();
1699            let order_entries = orders
1700                .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "o1", "userId": "u1" }))])
1701                .unwrap();
1702
1703            let mut entries = Vec::new();
1704            for entry in &user_entries {
1705                entries.push(JournalEntry {
1706                    t: "users".into(),
1707                    s: entry.seq(),
1708                    l: entry.line().to_string(),
1709                });
1710            }
1711            for entry in &order_entries {
1712                entries.push(JournalEntry {
1713                    t: "orders".into(),
1714                    s: entry.seq(),
1715                    l: entry.line().to_string(),
1716                });
1717            }
1718            write_journal(
1719                &dir.path().join(JOURNAL_FILE),
1720                &Journal {
1721                    header: JournalHeader {
1722                        journal_version: JOURNAL_VERSION,
1723                        kind: "unit".into(),
1724                        id: "crash".into(),
1725                        started_at: 0,
1726                        tables: vec!["users".into(), "orders".into()],
1727                    },
1728                    entries,
1729                },
1730            )
1731            .unwrap();
1732            // process dies here: prepared and journalled, never applied to either log.
1733        }
1734
1735        assert!(dir.path().join(JOURNAL_FILE).exists());
1736        let db = Database::open(dir.path()).unwrap();
1737        assert!(!dir.path().join(JOURNAL_FILE).exists());
1738        assert_eq!(
1739            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1740            Some(&json!("Ania"))
1741        );
1742        assert_eq!(
1743            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
1744            Some(&json!("u1"))
1745        );
1746        db.close().unwrap();
1747    }
1748
1749    #[test]
1750    fn a_second_open_of_the_same_directory_is_denied_while_the_first_still_holds_the_lock() {
1751        let dir = tempdir().unwrap();
1752        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1753
1754        let _first = Database::open(dir.path()).unwrap();
1755        let second = Database::open_with(
1756            dir.path(),
1757            OpenOptions {
1758                readonly: false,
1759                lock: LockMode::Enabled(LockOptions {
1760                    timeout_ms: Some(0),
1761                    ..Default::default()
1762                }),
1763                ..Default::default()
1764            },
1765        );
1766        assert!(matches!(second, Err(DatabaseError::Lock(_))));
1767    }
1768
1769    fn lock_path_of(dir: &Path) -> std::path::PathBuf {
1770        dir.join(crate::lock::LOCK_FILE)
1771    }
1772
1773    fn open_refusing(dir: &Path) -> DatabaseError {
1774        Database::open_with(
1775            dir,
1776            OpenOptions {
1777                lock: LockMode::Enabled(LockOptions {
1778                    timeout_ms: Some(0),
1779                    ..Default::default()
1780                }),
1781                ..Default::default()
1782            },
1783        )
1784        .map(|_| ())
1785        .expect_err("this directory must not open")
1786    }
1787
1788    #[test]
1789    fn a_directory_with_logs_but_no_catalog_is_refused_instead_of_opening_empty() {
1790        let dir = tempdir().unwrap();
1791        let mut db = Database::open(dir.path()).unwrap();
1792        db.create_table(
1793            "users",
1794            json!({
1795                "schemaVersion": 1,
1796                "table": "users",
1797                "primaryKey": "id",
1798                "fields": {
1799                    "id": { "type": "string" },
1800                    "name": { "type": "string", "required": true },
1801                },
1802            }),
1803        )
1804        .unwrap();
1805        db.table_mut("users")
1806            .unwrap()
1807            .put(&json!({ "id": "u1", "name": "Ania" }), false)
1808            .unwrap();
1809        db.close().unwrap();
1810
1811        let log = fs::read(dir.path().join("users.jsonl")).unwrap();
1812        fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
1813
1814        let refusal = open_refusing(dir.path()).to_string();
1815        assert!(
1816            refusal.contains("users.jsonl"),
1817            "the refusal has to name the log it found: {refusal}"
1818        );
1819        assert!(
1820            refusal.contains(MANIFEST_FILE),
1821            "the refusal has to name the catalog: {refusal}"
1822        );
1823
1824        // The refusal is worth nothing if it already wrote a fresh catalog over the data it refused.
1825        assert!(
1826            !dir.path().join(MANIFEST_FILE).exists(),
1827            "a refused open must not leave a catalog"
1828        );
1829        assert_eq!(
1830            fs::read(dir.path().join("users.jsonl")).unwrap(),
1831            log,
1832            "the log must be untouched"
1833        );
1834    }
1835
1836    #[test]
1837    fn an_empty_directory_and_one_holding_no_logs_still_open_as_a_new_database() {
1838        let empty = tempdir().unwrap();
1839        assert!(
1840            Database::open(empty.path()).is_ok(),
1841            "an empty directory is still a new database"
1842        );
1843
1844        let littered = tempdir().unwrap();
1845        fs::write(littered.path().join("README.md"), "# not a table\n").unwrap();
1846        fs::create_dir(littered.path().join("users-backup")).unwrap();
1847        assert!(Database::open(littered.path()).is_ok(), "nothing here is a table log");
1848    }
1849
1850    #[test]
1851    fn the_logs_of_a_partitioned_table_are_found_one_level_down_too() {
1852        let dir = tempdir().unwrap();
1853        fs::create_dir_all(dir.path().join("notes")).unwrap();
1854        fs::write(dir.path().join("notes").join("note.jsonl"), "").unwrap();
1855
1856        let refusal = open_refusing(dir.path()).to_string();
1857        assert!(
1858            refusal.contains("note.jsonl"),
1859            "a partitioned log still counts as data: {refusal}"
1860        );
1861    }
1862
1863    /// Every way the catalog read under the lock can refuse. The lock is taken before any of them
1864    /// runs (`stored_kind_of` is deliberately lenient), so each one is a chance to strand the lock
1865    /// file on a process that is still alive — which no staleness rule would ever reclaim.
1866    #[test]
1867    fn a_directory_refused_at_open_is_left_unlocked_whichever_way_the_catalog_was_broken() {
1868        for (case, document) in [
1869            ("not JSON at all", "{ this is not json".to_string()),
1870            (
1871                "no formatVersion",
1872                json!({ "catalogVersion": 2, "tables": {} }).to_string(),
1873            ),
1874            (
1875                "the format-1 marker",
1876                json!({ "hayson": 1, "formatVersion": 2, "tables": {} }).to_string(),
1877            ),
1878            (
1879                "a formatVersion from the future",
1880                json!({ "catalogVersion": 2, "formatVersion": 3, "tables": {} }).to_string(),
1881            ),
1882        ] {
1883            let dir = tempdir().unwrap();
1884            std::fs::write(dir.path().join(MANIFEST_FILE), &document).unwrap();
1885
1886            open_refusing(dir.path());
1887
1888            assert!(
1889                !lock_path_of(dir.path()).exists(),
1890                "{case}: the refused open left its lock file behind"
1891            );
1892        }
1893    }
1894
1895    /// The point of releasing on refusal: the directory has to stay openable. Nothing else reclaims
1896    /// such a lock — its owner is the very process that got the refusal, and it is still running.
1897    #[test]
1898    fn a_directory_opens_normally_after_an_open_that_was_refused_a_broken_catalog() {
1899        let dir = tempdir().unwrap();
1900        std::fs::write(dir.path().join(MANIFEST_FILE), "{ this is not json").unwrap();
1901
1902        open_refusing(dir.path());
1903
1904        std::fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
1905        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1906
1907        let mut db = Database::open(dir.path()).unwrap();
1908        db.table_mut("users")
1909            .unwrap()
1910            .put(&json!({ "id": "u1", "name": "Ania" }), false)
1911            .unwrap();
1912        db.close().unwrap();
1913    }
1914
1915    /// A refusal raised past the catalog — the cipher is built after it parsed — has to unwind the
1916    /// same way, which is what putting the lock behind a guard instead of one `release` call buys.
1917    #[test]
1918    fn a_directory_refused_for_a_bad_cipher_key_is_left_unlocked_too() {
1919        let dir = tempdir().unwrap();
1920        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1921
1922        let error = Database::open_with(
1923            dir.path(),
1924            OpenOptions {
1925                cypher: Some(CipherOptions {
1926                    keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
1927                    key_id: None,
1928                    on_missing_key: None,
1929                }),
1930                ..OpenOptions::default()
1931            },
1932        )
1933        .map(|_| ())
1934        .unwrap_err();
1935
1936        assert!(matches!(error, DatabaseError::Cipher(_)));
1937        assert!(!lock_path_of(dir.path()).exists());
1938        Database::open(dir.path()).unwrap().close().unwrap();
1939    }
1940
1941    /// The other half of the guard: a successful open must still be holding the lock, not have
1942    /// released it on the way out of `open_locked`.
1943    #[test]
1944    fn an_open_that_succeeded_still_holds_the_lock_until_the_handle_closes() {
1945        let dir = tempdir().unwrap();
1946        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1947
1948        let held = Database::open(dir.path()).unwrap();
1949        assert!(lock_path_of(dir.path()).exists());
1950        assert!(matches!(open_refusing(dir.path()), DatabaseError::Lock(_)));
1951
1952        held.close().unwrap();
1953        assert!(!lock_path_of(dir.path()).exists());
1954    }
1955
1956    /// A second writer of a directory somebody else already holds — the port of the TS
1957    /// `open(dir, { lock: false })`, deliberately dangerous and only ever reached on request.
1958    fn open_skipping_the_lock(dir: &Path) -> Database {
1959        Database::open_with(
1960            dir,
1961            OpenOptions {
1962                lock: LockMode::Skip,
1963                ..Default::default()
1964            },
1965        )
1966        .unwrap()
1967    }
1968
1969    fn declared_on_disk(dir: &Path, table: &str) -> Vec<String> {
1970        let catalog = read_catalog(&dir.join(MANIFEST_FILE))
1971            .unwrap()
1972            .expect("a catalog on disk");
1973        let mut names: Vec<String> = catalog.tables[table]
1974            .indexes
1975            .iter()
1976            .flatten()
1977            .map(|index| index.name.clone())
1978            .collect();
1979        names.sort();
1980        names
1981    }
1982
1983    /// The fixture the three lock-skipping tests share: a directory with two tables, two records,
1984    /// and no live handle — the `usersWithTwoRecords` of the TS
1985    /// `deklarowanie-indeksu-wspolbieznie.test.ts`.
1986    fn two_tables_with_records(dir: &Path) {
1987        write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
1988        let mut db = Database::open(dir).unwrap();
1989        db.table_mut("users")
1990            .unwrap()
1991            .put(&json!({ "id": "u1", "name": "Ania" }), false)
1992            .unwrap();
1993        db.table_mut("users")
1994            .unwrap()
1995            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
1996            .unwrap();
1997        db.close().unwrap();
1998    }
1999
2000    #[test]
2001    fn a_second_writer_that_skips_the_lock_opens_and_writes_beside_the_one_holding_it() {
2002        let dir = tempdir().unwrap();
2003        two_tables_with_records(dir.path());
2004
2005        let mut locked = Database::open(dir.path()).unwrap();
2006        let mut unlocked = open_skipping_the_lock(dir.path());
2007
2008        // Both are writers, on their own table each — the lock never came into it.
2009        locked
2010            .table_mut("users")
2011            .unwrap()
2012            .put(&json!({ "id": "u3", "name": "Kasia" }), false)
2013            .unwrap();
2014        unlocked
2015            .table_mut("orders")
2016            .unwrap()
2017            .put(&json!({ "id": "o1", "userId": "u1" }), false)
2018            .unwrap();
2019
2020        // While a third handle asking for the lock is still refused, so nothing was released.
2021        let third = Database::open_with(
2022            dir.path(),
2023            OpenOptions {
2024                lock: LockMode::Enabled(LockOptions {
2025                    timeout_ms: Some(0),
2026                    ..Default::default()
2027                }),
2028                ..Default::default()
2029            },
2030        );
2031        assert!(matches!(third, Err(DatabaseError::Lock(_))));
2032
2033        unlocked.close().unwrap();
2034        locked.close().unwrap();
2035    }
2036
2037    /// The window `declare_index` documents: it writes the catalog it holds in memory, with no
2038    /// check against the disk, and the writer lock is the only guard. A handle that skipped the
2039    /// lock therefore overwrites what the other one declared. This is the TS behaviour, asserted
2040    /// so a change to it is a decision and not an accident.
2041    #[test]
2042    fn a_lock_skipping_handle_that_never_refreshed_loses_the_declaration_of_the_locked_one() {
2043        let dir = tempdir().unwrap();
2044        two_tables_with_records(dir.path());
2045
2046        let mut locked = Database::open(dir.path()).unwrap();
2047        let mut unlocked = open_skipping_the_lock(dir.path());
2048
2049        locked
2050            .declare_index(
2051                "users",
2052                json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
2053            )
2054            .unwrap();
2055        unlocked
2056            .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
2057            .unwrap();
2058
2059        unlocked.close().unwrap();
2060        locked.close().unwrap();
2061
2062        assert_eq!(declared_on_disk(dir.path(), "users"), vec!["by_id".to_string()]);
2063    }
2064
2065    #[test]
2066    fn a_lock_skipping_handle_that_refreshes_first_keeps_both_declarations() {
2067        let dir = tempdir().unwrap();
2068        two_tables_with_records(dir.path());
2069
2070        let mut locked = Database::open(dir.path()).unwrap();
2071        let mut unlocked = open_skipping_the_lock(dir.path());
2072
2073        locked
2074            .declare_index(
2075                "users",
2076                json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
2077            )
2078            .unwrap();
2079        assert!(unlocked.refresh().unwrap(), "the catalog changed under it");
2080        unlocked
2081            .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
2082            .unwrap();
2083
2084        unlocked.close().unwrap();
2085        locked.close().unwrap();
2086
2087        assert_eq!(
2088            declared_on_disk(dir.path(), "users"),
2089            vec!["by_id".to_string(), "by_name".to_string()]
2090        );
2091    }
2092
2093    /// A reader of the same directory as the live writer: readonly, so it takes no lock and the
2094    /// writer keeps its own — the shape a follower process has.
2095    fn open_follower(dir: &Path) -> Database {
2096        Database::open_with(
2097            dir,
2098            OpenOptions {
2099                readonly: true,
2100                lock: LockMode::default(),
2101                ..Default::default()
2102            },
2103        )
2104        .unwrap()
2105    }
2106
2107    #[test]
2108    fn refresh_with_nothing_new_on_disk_reports_no_change() {
2109        let dir = tempdir().unwrap();
2110        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2111        let mut writer = Database::open(dir.path()).unwrap();
2112        writer
2113            .table_mut("users")
2114            .unwrap()
2115            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2116            .unwrap();
2117        let mut follower = open_follower(dir.path());
2118        follower.refresh().unwrap();
2119
2120        assert!(!follower.refresh().unwrap());
2121        writer.close().unwrap();
2122    }
2123
2124    #[test]
2125    fn refresh_picks_up_both_a_table_and_the_records_another_process_wrote() {
2126        let dir = tempdir().unwrap();
2127        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2128        let mut writer = Database::open(dir.path()).unwrap();
2129        let mut follower = open_follower(dir.path());
2130        assert_eq!(follower.table_names(), vec!["users"]);
2131
2132        writer
2133            .table_mut("users")
2134            .unwrap()
2135            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2136            .unwrap();
2137        writer
2138            .create_table(
2139                "orders",
2140                json!({
2141                    "schemaVersion": 1,
2142                    "table": "orders",
2143                    "primaryKey": "id",
2144                    "fields": { "id": { "type": "string" }, "userId": { "type": "string", "required": true } },
2145                }),
2146            )
2147            .unwrap();
2148        writer
2149            .table_mut("orders")
2150            .unwrap()
2151            .put(&json!({ "id": "o1", "userId": "u1" }), false)
2152            .unwrap();
2153
2154        assert!(follower.refresh().unwrap());
2155
2156        assert_eq!(follower.table_names(), vec!["orders", "users"]);
2157        assert_eq!(
2158            follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2159            Some(&json!("Ania"))
2160        );
2161        assert_eq!(
2162            follower
2163                .table("orders")
2164                .unwrap()
2165                .get("o1")
2166                .unwrap()
2167                .unwrap()
2168                .get("userId"),
2169            Some(&json!("u1"))
2170        );
2171        writer.close().unwrap();
2172    }
2173
2174    #[test]
2175    fn refresh_survives_a_compaction_run_by_the_writer_process() {
2176        let dir = tempdir().unwrap();
2177        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2178        let mut writer = Database::open(dir.path()).unwrap();
2179        let users = writer.table_mut("users").unwrap();
2180        users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
2181        users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
2182        users.patch("u1", &json!({ "name": "Anna" }), false).unwrap();
2183        let mut follower = open_follower(dir.path());
2184        assert_eq!(follower.table("users").unwrap().len(), 2);
2185
2186        writer.table_mut("users").unwrap().compact().unwrap();
2187        assert!(follower.refresh().unwrap());
2188
2189        assert_eq!(follower.table("users").unwrap().len(), 2);
2190        assert_eq!(
2191            follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2192            Some(&json!("Anna"))
2193        );
2194        assert_eq!(follower.table("users").unwrap().stats().gen, 1);
2195        writer.close().unwrap();
2196    }
2197
2198    #[test]
2199    fn unit_of_work_commits_across_two_tables_and_the_database_sees_both() {
2200        let dir = tempdir().unwrap();
2201        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2202        let mut db = Database::open(dir.path()).unwrap();
2203
2204        let mut unit = db.unit_of_work();
2205        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
2206        unit.put("orders", json!({ "id": "o1", "userId": "u1" }));
2207        let result = db.commit(&mut unit).unwrap();
2208        assert_eq!(result.operations, 2);
2209
2210        assert_eq!(
2211            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2212            Some(&json!("Ania"))
2213        );
2214        assert_eq!(
2215            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2216            Some(&json!("u1"))
2217        );
2218        db.close().unwrap();
2219    }
2220
2221    #[test]
2222    fn create_table_on_an_open_database_adds_a_new_working_table_immediately() {
2223        let dir = tempdir().unwrap();
2224        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2225        let mut db = Database::open(dir.path()).unwrap();
2226
2227        db.create_table(
2228            "orders",
2229            json!({
2230                "schemaVersion": 1,
2231                "table": "orders",
2232                "primaryKey": "id",
2233                "fields": {
2234                    "id": { "type": "string" },
2235                    "userId": { "type": "string", "required": true },
2236                },
2237            }),
2238        )
2239        .unwrap();
2240
2241        assert_eq!(db.table_names(), vec!["orders", "users"]);
2242        assert!(db.catalog.tables.contains_key("orders"));
2243
2244        db.table_mut("orders")
2245            .unwrap()
2246            .put(&json!({ "id": "o1", "userId": "u1" }), false)
2247            .unwrap();
2248        assert_eq!(
2249            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2250            Some(&json!("u1"))
2251        );
2252
2253        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2254        assert!(reread.tables.contains_key("orders"));
2255        db.close().unwrap();
2256    }
2257
2258    #[test]
2259    fn create_table_refuses_a_schema_naming_a_table_other_than_the_one_being_created() {
2260        let dir = tempdir().unwrap();
2261        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2262        let mut db = Database::open(dir.path()).unwrap();
2263        let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2264
2265        let error = db
2266            .create_table(
2267                "orders",
2268                json!({
2269                    "schemaVersion": 1,
2270                    "table": "invoices",
2271                    "primaryKey": "id",
2272                    "fields": { "id": { "type": "string" } },
2273                }),
2274            )
2275            .expect_err("a schema naming another table is refused");
2276
2277        assert!(matches!(error, DatabaseError::Schema(_)), "got {error:?}");
2278        let message = error.to_string();
2279        assert!(
2280            message.contains("invoices"),
2281            "the refusal names the schema's table: {message}"
2282        );
2283        assert!(
2284            message.contains("orders"),
2285            "the refusal names the key it was created under: {message}"
2286        );
2287
2288        assert_eq!(db.table_names(), vec!["users"]);
2289        assert!(!db.catalog.tables.contains_key("orders"));
2290        assert!(!db.catalog.tables.contains_key("invoices"));
2291        assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2292        db.close().unwrap();
2293    }
2294
2295    #[test]
2296    fn create_table_still_takes_a_schema_naming_exactly_the_table_it_creates() {
2297        let dir = tempdir().unwrap();
2298        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2299        let mut db = Database::open(dir.path()).unwrap();
2300
2301        db.create_table(
2302            "orders",
2303            json!({
2304                "schemaVersion": 1,
2305                "table": "orders",
2306                "primaryKey": "id",
2307                "fields": { "id": { "type": "string" } },
2308            }),
2309        )
2310        .unwrap();
2311
2312        assert_eq!(db.table("orders").unwrap().schema().table, "orders");
2313        db.close().unwrap();
2314    }
2315
2316    #[test]
2317    fn create_table_refuses_on_a_readonly_database_and_changes_nothing() {
2318        let dir = tempdir().unwrap();
2319        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2320        // NOTE: a readonly open refuses a catalog whose logs are not on disk yet (F8) — make them.
2321        Database::open(dir.path()).unwrap().close().unwrap();
2322        let mut db = Database::open_with(
2323            dir.path(),
2324            OpenOptions {
2325                readonly: true,
2326                lock: LockMode::default(),
2327                ..Default::default()
2328            },
2329        )
2330        .unwrap();
2331
2332        let error = db
2333            .create_table(
2334                "orders",
2335                json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2336            )
2337            .unwrap_err();
2338        assert!(
2339            matches!(&error, DatabaseError::ReadOnly { dir: refused, operation } if refused == dir.path() && operation == "createTable"),
2340            "expected a typed readonly refusal, got {error:?}"
2341        );
2342        assert_eq!(db.table_names(), vec!["users"]);
2343    }
2344
2345    #[test]
2346    fn create_table_refused_by_a_directory_on_the_catalog_temporary_path_keeps_the_typed_refusal() {
2347        let dir = tempdir().unwrap();
2348        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2349        let mut db = Database::open(dir.path()).unwrap();
2350        let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2351        // NOTE: a directory on the .tmp path fails the write with EISDIR whatever the uid is.
2352        let temporary = dir.path().join(format!("{MANIFEST_FILE}.tmp"));
2353        fs::create_dir(&temporary).unwrap();
2354
2355        let error = db
2356            .create_table(
2357                "orders",
2358                json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2359            )
2360            .unwrap_err();
2361
2362        let refusal = match &error {
2363            DatabaseError::Io(inner) => AtomicTemporaryFileError::downcast(inner)
2364                .expect("a .tmp path that could not be claimed stays a temporary-file refusal"),
2365            other => panic!("expected an io error keeping the typed refusal, got {other:?}"),
2366        };
2367        assert_eq!(refusal.path, temporary);
2368        assert!(refusal.reason.contains("could not be cleared"), "{}", refusal.reason);
2369        assert_eq!(db.table_names(), vec!["users"]);
2370        assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2371    }
2372
2373    #[test]
2374    fn a_catalog_write_that_fails_for_any_other_reason_is_still_a_catalog_refusal() {
2375        let unreadable = catalog_write_error(Box::new(HaysonSchemaError("nope".into())));
2376
2377        assert!(matches!(unreadable, DatabaseError::Catalog(_)), "got {unreadable:?}");
2378        assert!(matches!(
2379            catalog_write_error(Box::new(io::Error::from(io::ErrorKind::PermissionDenied))),
2380            DatabaseError::Catalog(_)
2381        ));
2382    }
2383
2384    #[test]
2385    fn drop_field_without_dependents_removes_it_from_disk_and_from_memory() {
2386        let dir = tempdir().unwrap();
2387        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2388        let mut db = Database::open(dir.path()).unwrap();
2389        db.table_mut("users")
2390            .unwrap()
2391            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2392            .unwrap();
2393        db.table_mut("users").unwrap().delete("u1", false).unwrap();
2394
2395        db.drop_field("users", "name", false).unwrap();
2396
2397        assert!(!db.table("users").unwrap().schema().fields.contains_key("name"));
2398        assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2399
2400        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2401        assert!(!reread.tables["users"].fields.contains_key("name"));
2402        db.close().unwrap();
2403    }
2404
2405    #[test]
2406    fn drop_field_refuses_a_live_residual_field_without_force() {
2407        let dir = tempdir().unwrap();
2408        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2409        let mut db = Database::open(dir.path()).unwrap();
2410        db.table_mut("users")
2411            .unwrap()
2412            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2413            .unwrap();
2414
2415        let error = db.drop_field("users", "name", false);
2416        assert!(error.is_err());
2417        assert!(db.catalog.tables["users"].fields.contains_key("name"));
2418        assert!(db.table("users").unwrap().schema().fields.contains_key("name"));
2419
2420        db.drop_field("users", "name", true).unwrap();
2421        assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2422        db.close().unwrap();
2423    }
2424
2425    fn users_catalog_indexing_name() -> Catalog {
2426        let mut catalog = catalog_of(&[users_schema()]);
2427        catalog.tables.get_mut("users").unwrap().indexes = Some(vec![IndexEntry {
2428            name: "by_name".to_string(),
2429            fields: vec!["name".to_string()],
2430            kind: CatalogIndexKind::Exact,
2431            unique: false,
2432            where_: None,
2433            expire_after_seconds: None,
2434        }]);
2435        catalog
2436    }
2437
2438    #[test]
2439    fn drop_field_names_the_declaration_when_residual_records_also_block_it() {
2440        let dir = tempdir().unwrap();
2441        write_catalog_to(dir.path(), &users_catalog_indexing_name());
2442        let mut db = Database::open(dir.path()).unwrap();
2443        db.table_mut("users")
2444            .unwrap()
2445            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2446            .unwrap();
2447
2448        let error = db.drop_field("users", "name", false).unwrap_err();
2449
2450        assert!(
2451            matches!(error, DatabaseError::Schema(_)),
2452            "expected a schema refusal, got {error:?}"
2453        );
2454        let message = error.to_string();
2455        assert!(message.contains("by_name"), "{message}");
2456        assert!(!message.contains("live record(s)"), "{message}");
2457        assert!(db.catalog.tables["users"].fields.contains_key("name"));
2458        db.close().unwrap();
2459    }
2460
2461    #[test]
2462    fn drop_field_names_the_declaration_when_it_is_the_only_thing_blocking_it() {
2463        let dir = tempdir().unwrap();
2464        write_catalog_to(dir.path(), &users_catalog_indexing_name());
2465        let mut db = Database::open(dir.path()).unwrap();
2466
2467        let error = db.drop_field("users", "name", false).unwrap_err();
2468
2469        assert!(
2470            matches!(error, DatabaseError::Schema(_)),
2471            "expected a schema refusal, got {error:?}"
2472        );
2473        assert!(error.to_string().contains("by_name"));
2474        assert!(db.catalog.tables["users"].fields.contains_key("name"));
2475        db.close().unwrap();
2476    }
2477
2478    #[test]
2479    fn drop_field_names_the_residual_records_when_no_declaration_blocks_it() {
2480        let dir = tempdir().unwrap();
2481        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2482        let mut db = Database::open(dir.path()).unwrap();
2483        db.table_mut("users")
2484            .unwrap()
2485            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2486            .unwrap();
2487
2488        let error = db.drop_field("users", "name", false).unwrap_err();
2489
2490        assert!(
2491            matches!(
2492                &error,
2493                DatabaseError::FieldInUse { table, field, records }
2494                    if table == "users" && field == "name" && *records == 1
2495            ),
2496            "expected a typed field-in-use refusal, got {error:?}"
2497        );
2498        assert!(db.catalog.tables["users"].fields.contains_key("name"));
2499        db.close().unwrap();
2500    }
2501
2502    #[test]
2503    fn drop_field_keeps_refusing_a_declared_dependency_even_with_force() {
2504        let dir = tempdir().unwrap();
2505        write_catalog_to(dir.path(), &users_catalog_indexing_name());
2506        let mut db = Database::open(dir.path()).unwrap();
2507        db.table_mut("users")
2508            .unwrap()
2509            .put(&json!({ "id": "u1", "name": "Ania" }), false)
2510            .unwrap();
2511
2512        let error = db.drop_field("users", "name", true).unwrap_err();
2513
2514        assert!(
2515            matches!(error, DatabaseError::Schema(_)),
2516            "expected a schema refusal, got {error:?}"
2517        );
2518        assert!(error.to_string().contains("by_name"));
2519        db.close().unwrap();
2520    }
2521
2522    #[test]
2523    fn declare_index_with_a_conflicting_unique_value_refuses_and_changes_nothing() {
2524        let dir = tempdir().unwrap();
2525        let schema = parse_schema(&json!({
2526            "schemaVersion": 1,
2527            "table": "users",
2528            "primaryKey": "id",
2529            "fields": {
2530                "id": { "type": "string" },
2531                "email": { "type": "string" },
2532            },
2533        }))
2534        .unwrap();
2535        write_catalog_to(dir.path(), &catalog_of(&[schema]));
2536        let mut db = Database::open(dir.path()).unwrap();
2537        db.table_mut("users")
2538            .unwrap()
2539            .put(&json!({ "id": "u1", "email": "same@example.com" }), false)
2540            .unwrap();
2541        db.table_mut("users")
2542            .unwrap()
2543            .put(&json!({ "id": "u2", "email": "same@example.com" }), false)
2544            .unwrap();
2545
2546        let error = db.declare_index(
2547            "users",
2548            json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2549        );
2550        assert!(error.is_err());
2551        assert!(db.catalog.tables["users"].indexes.is_none());
2552
2553        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2554        assert!(reread.tables["users"].indexes.is_none());
2555        db.close().unwrap();
2556    }
2557
2558    fn orders_user_relation(enforce: bool) -> Value {
2559        json!({
2560            "name": "orders_user",
2561            "from": { "table": "orders", "field": "userId" },
2562            "to": { "table": "users", "field": "id" },
2563            "cardinality": "many-to-one",
2564            "enforce": enforce,
2565        })
2566    }
2567
2568    #[test]
2569    fn declare_relation_adds_a_relation_to_the_catalog_and_persists_it() {
2570        let dir = tempdir().unwrap();
2571        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2572        let mut db = Database::open(dir.path()).unwrap();
2573
2574        db.declare_relation(orders_user_relation(false), false).unwrap();
2575
2576        assert_eq!(db.catalog.relations.as_ref().unwrap()[0].name, "orders_user");
2577        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2578        assert_eq!(reread.relations.unwrap()[0].name, "orders_user");
2579        db.close().unwrap();
2580    }
2581
2582    #[test]
2583    fn declare_relation_with_enforce_true_refuses_over_a_live_orphan_and_force_overrides() {
2584        let dir = tempdir().unwrap();
2585        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2586        let mut db = Database::open(dir.path()).unwrap();
2587        db.table_mut("orders")
2588            .unwrap()
2589            .put(&json!({ "id": "o1", "userId": "ghost" }), false)
2590            .unwrap();
2591
2592        let error = db.declare_relation(orders_user_relation(true), false).unwrap_err();
2593        assert!(error.to_string().contains("orders_user"));
2594        assert!(db.catalog.relations.is_none());
2595
2596        db.declare_relation(orders_user_relation(true), true).unwrap();
2597        assert_eq!(db.catalog.relations.as_ref().unwrap()[0].enforce, Some(true));
2598        db.close().unwrap();
2599    }
2600
2601    #[test]
2602    fn put_checked_refuses_a_value_with_no_counterpart_on_an_enforced_relation() {
2603        let dir = tempdir().unwrap();
2604        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2605        let mut db = Database::open(dir.path()).unwrap();
2606        db.declare_relation(orders_user_relation(true), false).unwrap();
2607
2608        let error = db
2609            .put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
2610            .unwrap_err();
2611
2612        assert!(error.to_string().contains("orders_user"));
2613        assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
2614        db.close().unwrap();
2615    }
2616
2617    #[test]
2618    fn put_checked_allows_a_value_with_a_live_counterpart() {
2619        let dir = tempdir().unwrap();
2620        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2621        let mut db = Database::open(dir.path()).unwrap();
2622        db.declare_relation(orders_user_relation(true), false).unwrap();
2623        db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2624            .unwrap();
2625
2626        db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2627            .unwrap();
2628
2629        assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
2630        db.close().unwrap();
2631    }
2632
2633    #[test]
2634    fn put_checked_ignores_a_relation_that_does_not_declare_enforce() {
2635        let dir = tempdir().unwrap();
2636        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2637        let mut db = Database::open(dir.path()).unwrap();
2638        db.declare_relation(orders_user_relation(false), false).unwrap();
2639
2640        db.put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
2641            .unwrap();
2642
2643        assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
2644        db.close().unwrap();
2645    }
2646
2647    #[test]
2648    fn patch_checked_refuses_changing_the_field_to_a_value_with_no_counterpart() {
2649        let dir = tempdir().unwrap();
2650        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2651        let mut db = Database::open(dir.path()).unwrap();
2652        db.declare_relation(orders_user_relation(true), false).unwrap();
2653        db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2654            .unwrap();
2655        db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2656            .unwrap();
2657
2658        let error = db
2659            .patch_checked("orders", "o1", &json!({ "userId": "ghost" }), false)
2660            .unwrap_err();
2661
2662        assert!(error.to_string().contains("orders_user"));
2663        assert_eq!(
2664            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2665            Some(&json!("u1"))
2666        );
2667        db.close().unwrap();
2668    }
2669
2670    #[test]
2671    fn patch_checked_leaves_the_field_alone_when_changes_does_not_touch_it() {
2672        let dir = tempdir().unwrap();
2673        let orders_named = parse_schema(&json!({
2674            "schemaVersion": 1,
2675            "table": "orders",
2676            "primaryKey": "id",
2677            "fields": {
2678                "id": { "type": "string" },
2679                "userId": { "type": "string", "required": true },
2680                "note": { "type": "string" },
2681            },
2682        }))
2683        .unwrap();
2684        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_named]));
2685        let mut db = Database::open(dir.path()).unwrap();
2686        db.declare_relation(orders_user_relation(true), false).unwrap();
2687        db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2688            .unwrap();
2689        db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2690            .unwrap();
2691
2692        db.patch_checked("orders", "o1", &json!({ "note": "gift-wrapped" }), false)
2693            .unwrap();
2694
2695        assert_eq!(
2696            db.table("orders").unwrap().get("o1").unwrap().unwrap().get("note"),
2697            Some(&json!("gift-wrapped"))
2698        );
2699        db.close().unwrap();
2700    }
2701
2702    #[test]
2703    fn declare_index_without_a_conflict_covers_records_written_before_the_declaration() {
2704        let dir = tempdir().unwrap();
2705        let schema = parse_schema(&json!({
2706            "schemaVersion": 1,
2707            "table": "users",
2708            "primaryKey": "id",
2709            "fields": {
2710                "id": { "type": "string" },
2711                "email": { "type": "string" },
2712            },
2713        }))
2714        .unwrap();
2715        write_catalog_to(dir.path(), &catalog_of(&[schema]));
2716        let mut db = Database::open(dir.path()).unwrap();
2717        db.table_mut("users")
2718            .unwrap()
2719            .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
2720            .unwrap();
2721        db.table_mut("users")
2722            .unwrap()
2723            .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
2724            .unwrap();
2725        db.table_mut("users")
2726            .unwrap()
2727            .put(&json!({ "id": "u3", "email": "kasia@example.com" }), false)
2728            .unwrap();
2729
2730        db.declare_index(
2731            "users",
2732            json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2733        )
2734        .unwrap();
2735
2736        assert_eq!(db.catalog.tables["users"].indexes.as_ref().unwrap().len(), 1);
2737
2738        let spec = QueryBuilder::new()
2739            .r#where(
2740                "email",
2741                crate::query::predicates::ComparisonOp::Eq,
2742                json!("piotr@example.com"),
2743            )
2744            .spec();
2745        let found = db.table("users").unwrap().query(&spec).unwrap();
2746        assert_eq!(found.keys, vec!["u2".to_string()]);
2747
2748        let conflicting = db
2749            .table_mut("users")
2750            .unwrap()
2751            .put(&json!({ "id": "u4", "email": "piotr@example.com" }), false);
2752        assert!(conflicting.is_err());
2753        db.close().unwrap();
2754    }
2755
2756    #[test]
2757    fn declare_index_snapshots_the_reopened_table_and_keeps_snapshotting_it_afterwards() {
2758        let dir = tempdir().unwrap();
2759        let schema = parse_schema(&json!({
2760            "schemaVersion": 1,
2761            "table": "users",
2762            "primaryKey": "id",
2763            "fields": {
2764                "id": { "type": "string" },
2765                "email": { "type": "string" },
2766            },
2767        }))
2768        .unwrap();
2769        write_catalog_to(dir.path(), &catalog_of(&[schema]));
2770        let mut db = Database::open(dir.path()).unwrap();
2771        db.table_mut("users")
2772            .unwrap()
2773            .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
2774            .unwrap();
2775
2776        db.declare_index(
2777            "users",
2778            json!({ "name": "users_email_exact", "fields": ["email"], "kind": "exact" }),
2779        )
2780        .unwrap();
2781
2782        assert!(
2783            index_path(dir.path(), "users").exists(),
2784            "declare_index owes the table a snapshot carrying the new index's projection"
2785        );
2786
2787        db.table_mut("users")
2788            .unwrap()
2789            .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
2790            .unwrap();
2791        assert!(
2792            db.table("users").unwrap().needs_snapshot(),
2793            "the handle declare_index left behind must still be able to snapshot itself"
2794        );
2795        db.close().unwrap();
2796
2797        let reopened = Database::open(dir.path()).unwrap();
2798        assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
2799        let spec = QueryBuilder::new()
2800            .r#where(
2801                "email",
2802                crate::query::predicates::ComparisonOp::Eq,
2803                json!("piotr@example.com"),
2804            )
2805            .spec();
2806        assert_eq!(
2807            reopened.table("users").unwrap().query(&spec).unwrap().keys,
2808            vec!["u2".to_string()]
2809        );
2810        assert_eq!(
2811            declared_on_disk(dir.path(), "users"),
2812            vec!["users_email_exact".to_string()]
2813        );
2814        reopened.close().unwrap();
2815    }
2816
2817    #[test]
2818    fn a_database_wide_sync_default_reaches_a_table_declaring_none_of_its_own() {
2819        let dir = tempdir().unwrap();
2820        let mut catalog = catalog_of(&[users_schema()]);
2821        catalog.defaults = Some(crate::catalog::TableOptions {
2822            sync: Some(crate::catalog::CatalogSyncMode::Never),
2823            ..Default::default()
2824        });
2825        write_catalog_to(dir.path(), &catalog);
2826
2827        let db = Database::open(dir.path()).unwrap();
2828        assert_eq!(
2829            db.table("users").unwrap().sync_mode(),
2830            crate::catalog::CatalogSyncMode::Never
2831        );
2832        db.close().unwrap();
2833    }
2834
2835    #[test]
2836    fn a_table_declaring_its_own_sync_mode_beats_the_database_wide_default() {
2837        let dir = tempdir().unwrap();
2838        let mut catalog = catalog_of(&[users_schema(), orders_schema()]);
2839        catalog.defaults = Some(crate::catalog::TableOptions {
2840            sync: Some(crate::catalog::CatalogSyncMode::Never),
2841            ..Default::default()
2842        });
2843        catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
2844            sync: Some(crate::catalog::CatalogSyncMode::Commit),
2845            ..Default::default()
2846        });
2847        write_catalog_to(dir.path(), &catalog);
2848
2849        let db = Database::open(dir.path()).unwrap();
2850        assert_eq!(
2851            db.table("users").unwrap().sync_mode(),
2852            crate::catalog::CatalogSyncMode::Commit
2853        );
2854        assert_eq!(
2855            db.table("orders").unwrap().sync_mode(),
2856            crate::catalog::CatalogSyncMode::Never
2857        );
2858        db.close().unwrap();
2859    }
2860
2861    #[test]
2862    fn a_table_declaring_validation_warn_takes_a_write_with_a_stray_field_and_stores_it_without_that_field() {
2863        let dir = tempdir().unwrap();
2864        let mut catalog = catalog_of(&[users_schema()]);
2865        catalog.tables.get_mut("users").unwrap().validation = Some(crate::catalog::CatalogValidation::Warn);
2866        write_catalog_to(dir.path(), &catalog);
2867
2868        let mut db = Database::open(dir.path()).unwrap();
2869        let document = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });
2870        let table = db.table_mut("users").unwrap();
2871        let record = table.put(&document, false).unwrap();
2872
2873        assert_eq!(record.get("nickname"), None);
2874        assert_eq!(
2875            table.dropped_field_warnings(&document),
2876            vec!["field \"nickname\": is not declared in the schema, dropped from the write".to_string()]
2877        );
2878        db.close().unwrap();
2879
2880        let reopened = Database::open(dir.path()).unwrap();
2881        let stored = reopened.table("users").unwrap().get("u1").unwrap().unwrap();
2882        assert_eq!(stored.get("name"), Some(&json!("Ania")));
2883        assert_eq!(stored.get("nickname"), None);
2884        reopened.close().unwrap();
2885    }
2886
2887    #[test]
2888    fn a_table_declaring_no_validation_refuses_a_write_with_a_stray_field() {
2889        let dir = tempdir().unwrap();
2890        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2891
2892        let mut db = Database::open(dir.path()).unwrap();
2893        assert_eq!(
2894            db.table("users").unwrap().validation(),
2895            crate::catalog::CatalogValidation::Strict
2896        );
2897        let error = db
2898            .table_mut("users")
2899            .unwrap()
2900            .put(&json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }), false)
2901            .unwrap_err();
2902
2903        assert!(error.to_string().contains("is not declared in the schema"), "{error}");
2904        db.close().unwrap();
2905    }
2906
2907    fn deprecating_age(dir: &Path, validation: Option<crate::catalog::CatalogValidation>, note: &str) {
2908        let mut catalog = catalog_of(&[users_with_age_schema()]);
2909        let entry = catalog.tables.get_mut("users").unwrap();
2910        entry.validation = validation;
2911        entry.fields.get_mut("age").unwrap().deprecated = Some(note.to_string());
2912        write_catalog_to(dir, &catalog);
2913    }
2914
2915    fn users_with_age_schema() -> TableSchema {
2916        parse_schema(&json!({
2917            "schemaVersion": 1,
2918            "table": "users",
2919            "primaryKey": "id",
2920            "fields": {
2921                "id": { "type": "string" },
2922                "name": { "type": "string", "required": true },
2923                "age": { "type": "int" },
2924            },
2925        }))
2926        .unwrap()
2927    }
2928
2929    #[test]
2930    fn a_write_touching_a_field_the_catalog_marks_deprecated_warns_with_the_field_and_the_note() {
2931        let dir = tempdir().unwrap();
2932        deprecating_age(dir.path(), None, "use \"bornAt\" instead");
2933
2934        let mut db = Database::open(dir.path()).unwrap();
2935        let table = db.table_mut("users").unwrap();
2936        let touching = json!({ "id": "u1", "name": "Ania", "age": 30 });
2937        let record = table.put(&touching, false).unwrap();
2938
2939        assert_eq!(record.get("age"), Some(&json!(30)));
2940        assert_eq!(
2941            table.deprecated_field_warnings(&touching),
2942            vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()]
2943        );
2944        assert!(table
2945            .deprecated_field_warnings(&json!({ "id": "u2", "name": "Ola" }))
2946            .is_empty());
2947        db.close().unwrap();
2948    }
2949
2950    #[test]
2951    fn the_deprecated_mark_reaches_a_table_in_either_validation_mode_and_on_every_write() {
2952        let expected = vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()];
2953        let document = json!({ "id": "u1", "name": "Ania", "age": 30 });
2954
2955        for validation in [None, Some(crate::catalog::CatalogValidation::Warn)] {
2956            let dir = tempdir().unwrap();
2957            deprecating_age(dir.path(), validation, "use \"bornAt\" instead");
2958
2959            let mut db = Database::open(dir.path()).unwrap();
2960            let table = db.table_mut("users").unwrap();
2961            table.put(&document, false).unwrap();
2962            assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
2963            table.put(&document, false).unwrap();
2964            assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
2965            db.close().unwrap();
2966
2967            let reopened = Database::open(dir.path()).unwrap();
2968            assert_eq!(
2969                reopened.table("users").unwrap().deprecated_field_warnings(&document),
2970                expected
2971            );
2972            reopened.close().unwrap();
2973        }
2974    }
2975
2976    #[test]
2977    fn a_catalog_marking_no_field_deprecated_leaves_every_table_without_the_warning() {
2978        let dir = tempdir().unwrap();
2979        write_catalog_to(dir.path(), &catalog_of(&[users_with_age_schema()]));
2980
2981        let db = Database::open(dir.path()).unwrap();
2982        let table = db.table("users").unwrap();
2983
2984        assert!(table.deprecated_fields().is_empty());
2985        assert!(table
2986            .deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania", "age": 30 }))
2987            .is_empty());
2988        db.close().unwrap();
2989    }
2990
2991    #[test]
2992    fn a_catalog_naming_no_sync_anywhere_leaves_every_table_committing() {
2993        let dir = tempdir().unwrap();
2994        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2995
2996        let db = Database::open(dir.path()).unwrap();
2997        assert_eq!(
2998            db.table("users").unwrap().sync_mode(),
2999            crate::catalog::CatalogSyncMode::Commit
3000        );
3001        db.close().unwrap();
3002    }
3003
3004    #[test]
3005    fn a_sync_never_table_survives_a_close_and_reopen_with_its_writes_and_its_mode() {
3006        let dir = tempdir().unwrap();
3007        let mut catalog = catalog_of(&[users_schema()]);
3008        catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
3009            sync: Some(crate::catalog::CatalogSyncMode::Never),
3010            ..Default::default()
3011        });
3012        write_catalog_to(dir.path(), &catalog);
3013
3014        {
3015            let mut db = Database::open(dir.path()).unwrap();
3016            db.table_mut("users")
3017                .unwrap()
3018                .put(&json!({ "id": "u1", "name": "Ania" }), false)
3019                .unwrap();
3020            db.close().unwrap();
3021        }
3022
3023        let db = Database::open(dir.path()).unwrap();
3024        assert_eq!(
3025            db.table("users").unwrap().sync_mode(),
3026            crate::catalog::CatalogSyncMode::Never
3027        );
3028        assert_eq!(
3029            db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3030            Some(&json!("Ania"))
3031        );
3032        db.close().unwrap();
3033    }
3034
3035    #[test]
3036    fn a_database_wide_default_names_the_link_parsing_mode_a_table_declaring_none_opens_with() {
3037        let dir = tempdir().unwrap();
3038        let pages_schema = parse_schema(&json!({
3039            "schemaVersion": 1,
3040            "table": "pages",
3041            "primaryKey": "id",
3042            "fields": {
3043                "id": { "type": "string" },
3044                "page": { "type": "markdown" },
3045            },
3046        }))
3047        .unwrap();
3048        let mut catalog = catalog_of(&[pages_schema]);
3049        catalog.defaults = Some(crate::catalog::TableOptions {
3050            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3051            ..Default::default()
3052        });
3053        write_catalog_to(dir.path(), &catalog);
3054
3055        let db = Database::open(dir.path()).unwrap();
3056        assert_eq!(
3057            db.table("pages").unwrap().link_parsing(),
3058            crate::catalog::LinkParsing::Lazy
3059        );
3060        db.close().unwrap();
3061    }
3062
3063    #[test]
3064    fn a_table_declaring_its_own_link_parsing_mode_beats_the_database_wide_default() {
3065        let dir = tempdir().unwrap();
3066        let pages_schema = parse_schema(&json!({
3067            "schemaVersion": 1,
3068            "table": "pages",
3069            "primaryKey": "id",
3070            "fields": {
3071                "id": { "type": "string" },
3072                "page": { "type": "markdown" },
3073            },
3074        }))
3075        .unwrap();
3076        let mut catalog = catalog_of(&[pages_schema]);
3077        catalog.defaults = Some(crate::catalog::TableOptions {
3078            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3079            ..Default::default()
3080        });
3081        catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
3082            link_parsing: Some(crate::catalog::LinkParsing::Eager),
3083            ..Default::default()
3084        });
3085        write_catalog_to(dir.path(), &catalog);
3086
3087        let db = Database::open(dir.path()).unwrap();
3088        assert_eq!(
3089            db.table("pages").unwrap().link_parsing(),
3090            crate::catalog::LinkParsing::Eager
3091        );
3092        db.close().unwrap();
3093    }
3094
3095    #[test]
3096    fn a_database_wide_default_names_the_image_dir_a_table_declaring_none_writes_under() {
3097        let dir = tempdir().unwrap();
3098        let docs_schema = parse_schema(&json!({
3099            "schemaVersion": 1,
3100            "table": "docs",
3101            "primaryKey": "id",
3102            "fields": {
3103                "id": { "type": "string" },
3104                "cover": { "type": "image" },
3105            },
3106        }))
3107        .unwrap();
3108        let mut catalog = catalog_of(&[docs_schema]);
3109        catalog.defaults = Some(crate::catalog::TableOptions {
3110            image_dir: Some("shared-images".to_string()),
3111            ..Default::default()
3112        });
3113        write_catalog_to(dir.path(), &catalog);
3114
3115        let mut db = Database::open(dir.path()).unwrap();
3116        let record = db
3117            .table_mut("docs")
3118            .unwrap()
3119            .put(
3120                &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
3121                false,
3122            )
3123            .unwrap();
3124        let physical_path = record
3125            .get("cover")
3126            .unwrap()
3127            .get("physicalPath")
3128            .unwrap()
3129            .as_str()
3130            .unwrap();
3131
3132        assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
3133        assert!(dir.path().join("shared-images").join(physical_path).exists());
3134        db.close().unwrap();
3135    }
3136
3137    #[test]
3138    fn declare_index_of_a_composite_field_pair_serves_a_query_matching_both_fields() {
3139        let dir = tempdir().unwrap();
3140        let schema = parse_schema(&json!({
3141            "schemaVersion": 1,
3142            "table": "orders",
3143            "primaryKey": "id",
3144            "fields": {
3145                "id": { "type": "string" },
3146                "userId": { "type": "string" },
3147                "status": { "type": "string" },
3148            },
3149        }))
3150        .unwrap();
3151        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3152        let mut db = Database::open(dir.path()).unwrap();
3153        db.table_mut("orders")
3154            .unwrap()
3155            .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
3156            .unwrap();
3157        db.table_mut("orders")
3158            .unwrap()
3159            .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
3160            .unwrap();
3161        db.table_mut("orders")
3162            .unwrap()
3163            .put(&json!({ "id": "o3", "userId": "u2", "status": "open" }), false)
3164            .unwrap();
3165
3166        db.declare_index(
3167            "orders",
3168            json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
3169        )
3170        .unwrap();
3171
3172        let spec = QueryBuilder::new()
3173            .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
3174            .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("open"))
3175            .spec();
3176        let found = db.table("orders").unwrap().query(&spec).unwrap();
3177        assert_eq!(found.keys, vec!["o1".to_string()]);
3178        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3179        assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
3180        assert_eq!(found.plan.candidates, 1);
3181        db.close().unwrap();
3182    }
3183
3184    #[test]
3185    fn a_where_scoped_index_answers_only_a_query_that_repeats_its_predicate() {
3186        let dir = tempdir().unwrap();
3187        let schema = parse_schema(&json!({
3188            "schemaVersion": 1,
3189            "table": "users",
3190            "primaryKey": "id",
3191            "fields": {
3192                "id": { "type": "string" },
3193                "role": { "type": "string" },
3194                "status": { "type": "string" },
3195            },
3196        }))
3197        .unwrap();
3198        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3199        let mut db = Database::open(dir.path()).unwrap();
3200        db.table_mut("users")
3201            .unwrap()
3202            .put(&json!({ "id": "u1", "role": "admin", "status": "active" }), false)
3203            .unwrap();
3204        db.table_mut("users")
3205            .unwrap()
3206            .put(&json!({ "id": "u2", "role": "admin", "status": "inactive" }), false)
3207            .unwrap();
3208
3209        db.declare_index(
3210            "users",
3211            json!({
3212                "name": "users_role_active",
3213                "fields": ["role"],
3214                "kind": "exact",
3215                "where": [{ "field": "status", "op": "=", "value": "active" }],
3216            }),
3217        )
3218        .unwrap();
3219
3220        let eq = crate::query::predicates::ComparisonOp::Eq;
3221        let omitted = QueryBuilder::new().r#where("role", eq, json!("admin")).spec();
3222        let found = db.table("users").unwrap().query(&omitted).unwrap();
3223        assert_eq!(
3224            found.keys,
3225            vec!["u1".to_string(), "u2".to_string()],
3226            "the index holds u1 alone, but the query never asked for active records only"
3227        );
3228        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Scan);
3229
3230        let repeated = QueryBuilder::new()
3231            .r#where("role", eq, json!("admin"))
3232            .r#where("status", eq, json!("active"))
3233            .spec();
3234        let narrowed = db.table("users").unwrap().query(&repeated).unwrap();
3235        assert_eq!(narrowed.keys, vec!["u1".to_string()]);
3236        assert_eq!(narrowed.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3237        assert_eq!(narrowed.plan.index, Some("users_role_active".to_string()));
3238        assert_eq!(narrowed.plan.candidates, 1);
3239        db.close().unwrap();
3240    }
3241
3242    #[test]
3243    fn a_declared_composite_index_still_serves_queries_after_closing_and_reopening_the_database() {
3244        let dir = tempdir().unwrap();
3245        let schema = parse_schema(&json!({
3246            "schemaVersion": 1,
3247            "table": "orders",
3248            "primaryKey": "id",
3249            "fields": {
3250                "id": { "type": "string" },
3251                "userId": { "type": "string" },
3252                "status": { "type": "string" },
3253            },
3254        }))
3255        .unwrap();
3256        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3257        let mut db = Database::open(dir.path()).unwrap();
3258        db.table_mut("orders")
3259            .unwrap()
3260            .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
3261            .unwrap();
3262        db.table_mut("orders")
3263            .unwrap()
3264            .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
3265            .unwrap();
3266        db.declare_index(
3267            "orders",
3268            json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
3269        )
3270        .unwrap();
3271        db.close().unwrap();
3272
3273        let reopened = Database::open(dir.path()).unwrap();
3274        let spec = QueryBuilder::new()
3275            .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
3276            .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("closed"))
3277            .spec();
3278        let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3279        assert_eq!(found.keys, vec!["o2".to_string()]);
3280        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3281        assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
3282        reopened.close().unwrap();
3283    }
3284
3285    fn add_index_to_catalog_on_disk(dir: &Path, table: &str, index: IndexEntry) {
3286        let mut catalog = read_catalog(&dir.join(MANIFEST_FILE)).unwrap().unwrap();
3287        let entry = catalog.tables.get_mut(table).unwrap();
3288        entry.indexes = Some(vec![index]);
3289        write_catalog_to(dir, &catalog);
3290    }
3291
3292    fn sorted_index_entry(name: &str, fields: &[&str]) -> IndexEntry {
3293        IndexEntry {
3294            name: name.to_string(),
3295            fields: fields.iter().map(|field| field.to_string()).collect(),
3296            kind: CatalogIndexKind::Sorted,
3297            unique: false,
3298            where_: None,
3299            expire_after_seconds: None,
3300        }
3301    }
3302
3303    /// A snapshot entry carries the projection the registry held when it was written, so a snapshot
3304    /// taken before an index was declared knows none of that index's fields. Restoring from it
3305    /// would leave the new index empty and answer every query against it with silence — the
3306    /// fingerprint has to count the index set, not the schema alone, so this pairing is stale.
3307    #[test]
3308    fn a_snapshot_written_before_an_index_was_declared_is_rebuilt_rather_than_restored_into_it() {
3309        let dir = tempdir().unwrap();
3310        let schema = parse_schema(&json!({
3311            "schemaVersion": 1,
3312            "table": "orders",
3313            "primaryKey": "id",
3314            "fields": {
3315                "id": { "type": "string" },
3316                "userId": { "type": "string" },
3317                "total": { "type": "int" },
3318            },
3319        }))
3320        .unwrap();
3321        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3322        let mut db = Database::open(dir.path()).unwrap();
3323        let orders = db.table_mut("orders").unwrap();
3324        orders
3325            .put(&json!({ "id": "a", "userId": "u1", "total": 10 }), false)
3326            .unwrap();
3327        orders.put(&json!({ "id": "b", "userId": "u1" }), false).unwrap();
3328        orders
3329            .put(&json!({ "id": "c", "userId": "u0", "total": 5 }), false)
3330            .unwrap();
3331        db.close().unwrap();
3332        assert!(
3333            index_path(dir.path(), "orders").exists(),
3334            "the fixture needs a snapshot to restore from"
3335        );
3336
3337        add_index_to_catalog_on_disk(
3338            dir.path(),
3339            "orders",
3340            sorted_index_entry("orders_user_total", &["userId", "total"]),
3341        );
3342
3343        let reopened = Database::open(dir.path()).unwrap();
3344        assert_eq!(reopened.restore_outcome("orders"), Some(RestoreOutcome::Rebuilt));
3345
3346        let spec = QueryBuilder::new()
3347            .r#where("userId", crate::query::predicates::ComparisonOp::Gte, json!("u1"))
3348            .spec();
3349        let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3350        assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Sorted);
3351        assert_eq!(found.plan.index, Some("orders_user_total".to_string()));
3352        let mut keys = found.keys.clone();
3353        keys.sort();
3354        assert_eq!(
3355            keys,
3356            vec!["a".to_string(), "b".to_string()],
3357            "b has no total, but it still has a userId"
3358        );
3359        reopened.close().unwrap();
3360    }
3361
3362    /// The registry spans every partition of a table, so a composite unique index does too — but
3363    /// only if it was filled from the log rather than from a snapshot that predates it.
3364    #[test]
3365    fn a_composite_unique_index_declared_after_a_snapshot_still_holds_across_partitions() {
3366        let dir = tempdir().unwrap();
3367        let schema = parse_schema(&json!({
3368            "schemaVersion": 1,
3369            "table": "tickets",
3370            "primaryKey": "id",
3371            "partitionBy": "kind",
3372            "fields": {
3373                "id": { "type": "string" },
3374                "kind": { "type": "string", "required": true },
3375                "userId": { "type": "string", "required": true },
3376                "orderId": { "type": "int" },
3377            },
3378        }))
3379        .unwrap();
3380        write_catalog_to(dir.path(), &catalog_of(&[schema]));
3381        let mut db = Database::open(dir.path()).unwrap();
3382        let tickets = db.table_mut("tickets").unwrap();
3383        tickets
3384            .put(
3385                &json!({ "id": "t1", "kind": "sale", "userId": "u1", "orderId": 1 }),
3386                false,
3387            )
3388            .unwrap();
3389        tickets
3390            .put(
3391                &json!({ "id": "t2", "kind": "sale", "userId": "u1", "orderId": 2 }),
3392                false,
3393            )
3394            .unwrap();
3395        tickets
3396            .put(
3397                &json!({ "id": "t3", "kind": "refund", "userId": "u1", "orderId": 3 }),
3398                false,
3399            )
3400            .unwrap();
3401        db.close().unwrap();
3402
3403        add_index_to_catalog_on_disk(
3404            dir.path(),
3405            "tickets",
3406            IndexEntry {
3407                name: "tickets_user_order".to_string(),
3408                fields: vec!["userId".to_string(), "orderId".to_string()],
3409                kind: CatalogIndexKind::Exact,
3410                unique: true,
3411                where_: None,
3412                expire_after_seconds: None,
3413            },
3414        );
3415
3416        let mut reopened = Database::open(dir.path()).unwrap();
3417        let tickets = reopened.table_mut("tickets").unwrap();
3418        // NOTE: the clash sits in the "sale" partition, the write lands in "refund" — one key, two files.
3419        let clash = tickets.put(
3420            &json!({ "id": "t9", "kind": "refund", "userId": "u1", "orderId": 2 }),
3421            false,
3422        );
3423        match clash {
3424            Err(TableError::Unique(violation)) => {
3425                assert_eq!(violation.field, "userId, orderId");
3426                assert_eq!(violation.holder, "t2");
3427            }
3428            other => panic!("a repeated composite key must be refused, got {other:?}"),
3429        }
3430        tickets
3431            .put(
3432                &json!({ "id": "t10", "kind": "refund", "userId": "u1", "orderId": 9 }),
3433                false,
3434            )
3435            .unwrap();
3436        reopened.close().unwrap();
3437    }
3438
3439    #[test]
3440    fn close_snapshots_dirty_tables_so_a_reopen_restores_instead_of_rebuilding() {
3441        let dir = tempdir().unwrap();
3442        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3443        let mut db = Database::open(dir.path()).unwrap();
3444
3445        // NOTE: well under DEFAULT_SNAPSHOT_EVERY — only close(), not the threshold, should snapshot.
3446        db.table_mut("users")
3447            .unwrap()
3448            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3449            .unwrap();
3450        db.table_mut("users")
3451            .unwrap()
3452            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3453            .unwrap();
3454        assert!(!index_path(dir.path(), "users").exists());
3455
3456        db.close().unwrap();
3457        assert!(index_path(dir.path(), "users").exists());
3458
3459        let reopened = Database::open(dir.path()).unwrap();
3460        assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
3461        assert_eq!(
3462            reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3463            Some(&json!("Ania"))
3464        );
3465        assert_eq!(
3466            reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
3467            Some(&json!("Piotr"))
3468        );
3469        reopened.close().unwrap();
3470    }
3471
3472    #[test]
3473    fn close_commits_a_staged_write_so_a_reopen_still_finds_it() {
3474        let dir = tempdir().unwrap();
3475        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3476        let mut db = Database::open(dir.path()).unwrap();
3477        db.table_mut("users")
3478            .unwrap()
3479            .stage_put(&json!({ "id": "u1", "name": "Ania" }), false)
3480            .unwrap();
3481        assert_eq!(
3482            db.table("users").unwrap().len(),
3483            0,
3484            "nothing is committed before the close"
3485        );
3486
3487        db.close().unwrap();
3488
3489        let reopened = Database::open(dir.path()).unwrap();
3490        assert_eq!(
3491            reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3492            Some(&json!("Ania"))
3493        );
3494        reopened.close().unwrap();
3495    }
3496
3497    #[test]
3498    fn a_low_snapshot_every_writes_the_index_without_waiting_for_close() {
3499        let dir = tempdir().unwrap();
3500        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3501        let mut db = Database::open_with(
3502            dir.path(),
3503            OpenOptions {
3504                snapshot_every: 2,
3505                ..Default::default()
3506            },
3507        )
3508        .unwrap();
3509
3510        db.table_mut("users")
3511            .unwrap()
3512            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3513            .unwrap();
3514        assert!(!index_path(dir.path(), "users").exists());
3515        db.table_mut("users")
3516            .unwrap()
3517            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3518            .unwrap();
3519        assert!(index_path(dir.path(), "users").exists());
3520
3521        db.close().unwrap();
3522    }
3523
3524    #[test]
3525    fn a_snapshot_taken_mid_stream_does_not_disturb_writes_that_follow_it() {
3526        let dir = tempdir().unwrap();
3527        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3528        let mut db = Database::open_with(
3529            dir.path(),
3530            OpenOptions {
3531                snapshot_every: 2,
3532                ..Default::default()
3533            },
3534        )
3535        .unwrap();
3536
3537        db.table_mut("users")
3538            .unwrap()
3539            .put(&json!({ "id": "u1", "name": "Ania" }), false)
3540            .unwrap();
3541        db.table_mut("users")
3542            .unwrap()
3543            .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3544            .unwrap();
3545        db.table_mut("users")
3546            .unwrap()
3547            .put(&json!({ "id": "u3", "name": "Kasia" }), false)
3548            .unwrap();
3549        assert_eq!(
3550            db.table("users").unwrap().get("u3").unwrap().unwrap().get("name"),
3551            Some(&json!("Kasia"))
3552        );
3553        db.close().unwrap();
3554    }
3555
3556    fn users_insert_audits_into_orders() -> Value {
3557        json!({
3558            "name": "users_insert_audit",
3559            "table": "users",
3560            "event": "insert",
3561            "effect": {
3562                "table": "orders",
3563                "fields": [
3564                    { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
3565                    { "target": "userId", "source": { "kind": "new", "field": "id" } },
3566                ],
3567            },
3568        })
3569    }
3570
3571    #[test]
3572    fn a_table_declaring_an_image_dir_option_writes_bytes_under_it_instead_of_the_default() {
3573        let dir = tempdir().unwrap();
3574        let docs_schema = parse_schema(&json!({
3575            "schemaVersion": 1,
3576            "table": "docs",
3577            "primaryKey": "id",
3578            "fields": {
3579                "id": { "type": "string" },
3580                "cover": { "type": "image" },
3581            },
3582        }))
3583        .unwrap();
3584        let mut catalog = catalog_of(&[docs_schema]);
3585        catalog.tables.get_mut("docs").unwrap().options = Some(crate::catalog::TableOptions {
3586            image_dir: Some("shared-images".to_string()),
3587            ..Default::default()
3588        });
3589        write_catalog_to(dir.path(), &catalog);
3590
3591        let mut db = Database::open(dir.path()).unwrap();
3592        let record = db
3593            .table_mut("docs")
3594            .unwrap()
3595            .put(
3596                &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
3597                false,
3598            )
3599            .unwrap();
3600        let physical_path = record
3601            .get("cover")
3602            .unwrap()
3603            .get("physicalPath")
3604            .unwrap()
3605            .as_str()
3606            .unwrap();
3607
3608        assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
3609        assert!(dir.path().join("shared-images").join(physical_path).exists());
3610        db.close().unwrap();
3611    }
3612
3613    #[test]
3614    fn a_table_declaring_a_lazy_link_parsing_option_opens_with_it_instead_of_the_eager_default() {
3615        let dir = tempdir().unwrap();
3616        let pages_schema = parse_schema(&json!({
3617            "schemaVersion": 1,
3618            "table": "pages",
3619            "primaryKey": "id",
3620            "fields": {
3621                "id": { "type": "string" },
3622                "page": { "type": "markdown" },
3623            },
3624        }))
3625        .unwrap();
3626        let mut catalog = catalog_of(&[pages_schema]);
3627        catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
3628            link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3629            ..Default::default()
3630        });
3631        write_catalog_to(dir.path(), &catalog);
3632
3633        let mut db = Database::open(dir.path()).unwrap();
3634        assert_eq!(
3635            db.table("pages").unwrap().link_parsing(),
3636            crate::catalog::LinkParsing::Lazy
3637        );
3638
3639        let outcome = db
3640            .table_mut("pages")
3641            .unwrap()
3642            .put_with_warnings(
3643                &json!({ "id": "a", "page": { "content": "[missing](/docs/x.md)", "virtualPath": "/docs/a.md" } }),
3644                false,
3645            )
3646            .unwrap();
3647        assert!(outcome.warnings.is_empty());
3648        db.close().unwrap();
3649    }
3650
3651    #[test]
3652    fn a_table_declaring_no_link_parsing_option_defaults_to_eager() {
3653        let dir = tempdir().unwrap();
3654        let pages_schema = parse_schema(&json!({
3655            "schemaVersion": 1,
3656            "table": "pages",
3657            "primaryKey": "id",
3658            "fields": {
3659                "id": { "type": "string" },
3660                "page": { "type": "markdown" },
3661            },
3662        }))
3663        .unwrap();
3664        write_catalog_to(dir.path(), &catalog_of(&[pages_schema]));
3665
3666        let db = Database::open(dir.path()).unwrap();
3667        assert_eq!(
3668            db.table("pages").unwrap().link_parsing(),
3669            crate::catalog::LinkParsing::Eager
3670        );
3671    }
3672
3673    #[test]
3674    fn declare_trigger_writes_the_catalog_and_refreshes_both_tables_live_trigger_set() {
3675        let dir = tempdir().unwrap();
3676        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3677        let mut db = Database::open(dir.path()).unwrap();
3678
3679        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3680
3681        assert_eq!(db.catalog.triggers.as_ref().unwrap().len(), 1);
3682        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
3683        assert!(db.table("orders").unwrap().triggers().is_empty());
3684
3685        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
3686        assert_eq!(reread.triggers.as_ref().unwrap().len(), 1);
3687        assert_eq!(reread.triggers.as_ref().unwrap()[0].name, "users_insert_audit");
3688
3689        db.close().unwrap();
3690    }
3691
3692    #[test]
3693    fn a_declared_trigger_fires_its_effect_immediately_on_the_next_unit_of_work() {
3694        let dir = tempdir().unwrap();
3695        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3696        let mut db = Database::open(dir.path()).unwrap();
3697        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3698
3699        let mut unit = db.unit_of_work();
3700        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
3701        db.commit(&mut unit).unwrap();
3702
3703        assert_eq!(
3704            db.table("orders")
3705                .unwrap()
3706                .get("o-audit")
3707                .unwrap()
3708                .unwrap()
3709                .get("userId"),
3710            Some(&json!("u1"))
3711        );
3712        db.close().unwrap();
3713    }
3714
3715    #[test]
3716    fn a_declared_trigger_still_fires_after_closing_and_reopening_the_database() {
3717        let dir = tempdir().unwrap();
3718        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3719        let mut db = Database::open(dir.path()).unwrap();
3720        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3721        db.close().unwrap();
3722
3723        let mut reopened = Database::open(dir.path()).unwrap();
3724        let mut unit = reopened.unit_of_work();
3725        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
3726        reopened.commit(&mut unit).unwrap();
3727
3728        assert_eq!(
3729            reopened
3730                .table("orders")
3731                .unwrap()
3732                .get("o-audit")
3733                .unwrap()
3734                .unwrap()
3735                .get("userId"),
3736            Some(&json!("u1"))
3737        );
3738        reopened.close().unwrap();
3739    }
3740
3741    #[test]
3742    fn a_declared_trigger_still_fires_after_declare_index_reopened_its_source_table() {
3743        let dir = tempdir().unwrap();
3744        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3745        let mut db = Database::open(dir.path()).unwrap();
3746        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3747
3748        db.declare_index(
3749            "users",
3750            json!({ "name": "users_by_name", "fields": ["name"], "kind": "exact" }),
3751        )
3752        .unwrap();
3753
3754        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
3755
3756        let mut unit = db.unit_of_work();
3757        unit.put("users", json!({ "id": "u1", "name": "Ania" }));
3758        db.commit(&mut unit).unwrap();
3759
3760        assert_eq!(
3761            db.table("orders")
3762                .unwrap()
3763                .get("o-audit")
3764                .unwrap()
3765                .unwrap()
3766                .get("userId"),
3767            Some(&json!("u1"))
3768        );
3769        db.close().unwrap();
3770    }
3771
3772    #[test]
3773    fn a_declared_trigger_still_fires_after_drop_field_reopened_its_source_table() {
3774        let dir = tempdir().unwrap();
3775        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3776        let mut db = Database::open(dir.path()).unwrap();
3777        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3778
3779        db.drop_field("users", "name", false).unwrap();
3780
3781        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
3782
3783        let mut unit = db.unit_of_work();
3784        unit.put("users", json!({ "id": "u1" }));
3785        db.commit(&mut unit).unwrap();
3786
3787        assert_eq!(
3788            db.table("orders")
3789                .unwrap()
3790                .get("o-audit")
3791                .unwrap()
3792                .unwrap()
3793                .get("userId"),
3794            Some(&json!("u1"))
3795        );
3796        db.close().unwrap();
3797    }
3798
3799    #[test]
3800    fn a_declared_trigger_still_fires_after_create_table_migrated_its_source_table() {
3801        let dir = tempdir().unwrap();
3802        write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3803        let mut db = Database::open(dir.path()).unwrap();
3804        db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3805
3806        db.create_table(
3807            "users",
3808            json!({
3809                "schemaVersion": 1,
3810                "table": "users",
3811                "revision": 2,
3812                "primaryKey": "id",
3813                "fields": {
3814                    "id": { "type": "string" },
3815                    "name": { "type": "string", "required": true },
3816                    "city": { "type": "string" },
3817                },
3818            }),
3819        )
3820        .unwrap();
3821
3822        assert_eq!(db.table("users").unwrap().triggers().len(), 1);
3823
3824        let mut unit = db.unit_of_work();
3825        unit.put("users", json!({ "id": "u1", "name": "Ania", "city": "Kraków" }));
3826        db.commit(&mut unit).unwrap();
3827
3828        assert_eq!(
3829            db.table("orders")
3830                .unwrap()
3831                .get("o-audit")
3832                .unwrap()
3833                .unwrap()
3834                .get("userId"),
3835            Some(&json!("u1"))
3836        );
3837        db.close().unwrap();
3838    }
3839
3840    #[test]
3841    fn declare_trigger_refuses_an_unknown_effect_table_and_changes_nothing_on_disk() {
3842        let dir = tempdir().unwrap();
3843        write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3844        let mut db = Database::open(dir.path()).unwrap();
3845
3846        let spec = json!({
3847            "name": "users_insert_audit",
3848            "table": "users",
3849            "event": "insert",
3850            "effect": {
3851                "table": "ghost",
3852                "fields": [
3853                    { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
3854                ],
3855            },
3856        });
3857        let error = db.declare_trigger(spec);
3858        assert!(error.is_err());
3859        assert!(db.catalog.triggers.is_none());
3860
3861        let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
3862        assert!(reread.triggers.is_none());
3863        db.close().unwrap();
3864    }
3865
3866    mod automatic_compaction {
3867        use super::*;
3868        use crate::catalog::TableOptions;
3869
3870        fn enabled(dead_bytes_ratio: f64, min_dead_bytes: u64) -> CatalogCompaction {
3871            CatalogCompaction::Enabled {
3872                dead_bytes_ratio,
3873                min_dead_bytes,
3874            }
3875        }
3876
3877        fn opened(dir: &Path, compaction: Option<CatalogCompaction>) -> Database {
3878            Database::open_with(
3879                dir,
3880                OpenOptions {
3881                    compaction,
3882                    ..Default::default()
3883                },
3884            )
3885            .unwrap()
3886        }
3887
3888        fn lines_of(path: &Path) -> Vec<String> {
3889            fs::read_to_string(path).unwrap().lines().map(str::to_string).collect()
3890        }
3891
3892        /// One put and `overwrites` patches on the same key, so every patch but the last turns a
3893        /// whole line into garbage without changing how many records the table holds.
3894        fn churn(db: &mut Database, table: &str, key: &str, overwrites: usize) {
3895            let users = db.table_mut(table).unwrap();
3896            users.put(&json!({ "id": key, "name": "Ania" }), false).unwrap();
3897            for round in 0..overwrites {
3898                users
3899                    .patch(key, &json!({ "name": format!("Ania {round}") }), false)
3900                    .unwrap();
3901            }
3902        }
3903
3904        #[test]
3905        fn garbage_under_the_absolute_floor_leaves_the_log_alone_however_large_its_share() {
3906            let dir = tempdir().unwrap();
3907            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3908            let mut db = opened(dir.path(), Some(enabled(0.1, 1_000_000)));
3909
3910            churn(&mut db, "users", "u1", 12);
3911
3912            let stats = db.table("users").unwrap().stats();
3913            assert_eq!(stats.gen, 0, "minDeadBytes was never reached");
3914            assert!(stats.dead_bytes > 0, "the overwrites did make garbage");
3915            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
3916            db.close().unwrap();
3917        }
3918
3919        #[test]
3920        fn garbage_below_the_ratio_leaves_the_log_alone_however_many_bytes_it_is() {
3921            let dir = tempdir().unwrap();
3922            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3923            let mut db = opened(dir.path(), Some(enabled(0.99, 1)));
3924
3925            churn(&mut db, "users", "u1", 4);
3926
3927            let stats = db.table("users").unwrap().stats();
3928            assert_eq!(stats.gen, 0, "four of five lines is garbage, but not 99% of the bytes");
3929            assert!(stats.dead_bytes > 1);
3930            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 5);
3931            db.close().unwrap();
3932        }
3933
3934        #[test]
3935        fn crossing_both_thresholds_rewrites_the_log_and_moves_the_generation_on() {
3936            let dir = tempdir().unwrap();
3937            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3938            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
3939
3940            churn(&mut db, "users", "u1", 12);
3941
3942            let stats = db.table("users").unwrap().stats();
3943            assert!(stats.gen > 0, "the write that crossed the thresholds compacted");
3944            assert_eq!(stats.dead_bytes, 0);
3945            assert_eq!(stats.records, 1);
3946            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
3947            assert_eq!(
3948                db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3949                Some(&json!("Ania 11"))
3950            );
3951            db.close().unwrap();
3952        }
3953
3954        #[test]
3955        fn a_log_the_thresholds_rewrote_reopens_at_the_generation_and_the_records_it_left() {
3956            let dir = tempdir().unwrap();
3957            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3958            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
3959            churn(&mut db, "users", "u1", 12);
3960            db.table_mut("users")
3961                .unwrap()
3962                .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3963                .unwrap();
3964            let before = db.table("users").unwrap().stats();
3965            db.close().unwrap();
3966
3967            let reopened = opened(dir.path(), Some(enabled(0.5, 200)));
3968
3969            let after = reopened.table("users").unwrap().stats();
3970            assert_eq!(
3971                after.gen, before.gen,
3972                "the generation is carried by the snapshot, not reset"
3973            );
3974            assert_eq!(after.records, 2);
3975            assert_eq!(after.dead_bytes, 0);
3976            let mut keys: Vec<&String> = reopened.table("users").unwrap().keys().collect();
3977            keys.sort();
3978            assert_eq!(keys, vec!["u1", "u2"]);
3979            assert_eq!(
3980                reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3981                Some(&json!("Ania 11"))
3982            );
3983            reopened.close().unwrap();
3984        }
3985
3986        #[test]
3987        fn a_partitioned_table_is_measured_and_rewritten_one_partition_at_a_time() {
3988            let dir = tempdir().unwrap();
3989            let schema = parse_schema(&json!({
3990                "schemaVersion": 1,
3991                "table": "tickets",
3992                "primaryKey": "id",
3993                "partitionBy": "kind",
3994                "fields": {
3995                    "id": { "type": "string" },
3996                    "kind": { "type": "string", "required": true },
3997                    "note": { "type": "string" },
3998                },
3999            }))
4000            .unwrap();
4001            write_catalog_to(dir.path(), &catalog_of(&[schema]));
4002            // NOTE: a floor this far above what 12 rewrites of one line make can only be crossed once.
4003            let mut db = opened(dir.path(), Some(enabled(0.5, 700)));
4004
4005            let tickets = db.table_mut("tickets").unwrap();
4006            tickets
4007                .put(&json!({ "id": "r1", "kind": "refund", "note": "n" }), false)
4008                .unwrap();
4009            let refund = dir.path().join("tickets").join("refund.jsonl");
4010            let quiet_size = fs::metadata(&refund).unwrap().len();
4011
4012            tickets
4013                .put(&json!({ "id": "s1", "kind": "sale", "note": "n" }), false)
4014                .unwrap();
4015            for round in 0..12 {
4016                tickets
4017                    .patch("s1", &json!({ "note": format!("n{round}") }), false)
4018                    .unwrap();
4019            }
4020
4021            let sale = dir.path().join("tickets").join("sale.jsonl");
4022            assert!(
4023                lines_of(&sale).len() < 13,
4024                "the partition over its thresholds was rewritten, so not all 13 lines are still there"
4025            );
4026            assert_eq!(
4027                fs::metadata(&refund).unwrap().len(),
4028                quiet_size,
4029                "the partition holding no garbage was left byte for byte"
4030            );
4031            assert_eq!(lines_of(&refund).len(), 1);
4032            let stats = db.table("tickets").unwrap().stats();
4033            assert_eq!(
4034                stats.gen, 1,
4035                "one pass moves the generation once, not once per partition"
4036            );
4037            assert_eq!(stats.records, 2);
4038            db.close().unwrap();
4039        }
4040
4041        #[test]
4042        fn a_catalog_declaring_compaction_false_keeps_the_garbage_an_open_call_named_nothing_about() {
4043            let dir = tempdir().unwrap();
4044            let mut catalog = catalog_of(&[users_schema()]);
4045            catalog.defaults = Some(TableOptions {
4046                compaction: Some(CatalogCompaction::Disabled),
4047                ..Default::default()
4048            });
4049            write_catalog_to(dir.path(), &catalog);
4050            let mut db = opened(dir.path(), None);
4051
4052            churn(&mut db, "users", "u1", 12);
4053
4054            assert_eq!(db.table("users").unwrap().stats().gen, 0);
4055            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
4056            db.close().unwrap();
4057        }
4058
4059        #[test]
4060        fn a_tables_own_entry_beats_the_catalog_defaults_and_an_open_call_beats_them_both() {
4061            let dir = tempdir().unwrap();
4062            let mut catalog = catalog_of(&[users_schema()]);
4063            catalog.defaults = Some(TableOptions {
4064                compaction: Some(CatalogCompaction::Disabled),
4065                ..Default::default()
4066            });
4067            catalog.tables.get_mut("users").unwrap().options = Some(TableOptions {
4068                compaction: Some(enabled(0.5, 200)),
4069                ..Default::default()
4070            });
4071            write_catalog_to(dir.path(), &catalog);
4072
4073            let db = opened(dir.path(), None);
4074            assert_eq!(db.table("users").unwrap().compaction(), enabled(0.5, 200));
4075            db.close().unwrap();
4076
4077            let db = opened(dir.path(), Some(CatalogCompaction::Disabled));
4078            assert_eq!(db.table("users").unwrap().compaction(), CatalogCompaction::Disabled);
4079            db.close().unwrap();
4080        }
4081
4082        #[test]
4083        fn a_table_created_under_a_handle_is_armed_on_the_same_terms_as_one_opened_with_it() {
4084            let dir = tempdir().unwrap();
4085            let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4086            db.create_table(
4087                "users",
4088                json!({
4089                    "schemaVersion": 1,
4090                    "table": "users",
4091                    "primaryKey": "id",
4092                    "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
4093                }),
4094            )
4095            .unwrap();
4096
4097            churn(&mut db, "users", "u1", 12);
4098
4099            assert!(db.table("users").unwrap().stats().gen > 0);
4100            assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
4101            db.close().unwrap();
4102        }
4103
4104        #[test]
4105        fn an_open_call_naming_nothing_falls_back_to_the_engine_default_thresholds() {
4106            let dir = tempdir().unwrap();
4107            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4108            let db = opened(dir.path(), None);
4109
4110            assert_eq!(db.table("users").unwrap().compaction(), DEFAULT_COMPACTION);
4111            db.close().unwrap();
4112        }
4113    }
4114
4115    /// `OpenOptions::cache` reaching the tables — the read cache itself is pinned in
4116    /// `table::cache` and `table::handle`.
4117    mod read_cache {
4118        use super::*;
4119
4120        fn schema() -> Value {
4121            json!({
4122                "schemaVersion": 1,
4123                "table": "users",
4124                "primaryKey": "id",
4125                "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
4126            })
4127        }
4128
4129        fn opened(dir: &Path, cache: CachePolicy) -> Database {
4130            Database::open_with(
4131                dir,
4132                OpenOptions {
4133                    cache,
4134                    ..Default::default()
4135                },
4136            )
4137            .unwrap()
4138        }
4139
4140        #[test]
4141        fn an_open_that_names_no_cache_leaves_every_table_caching_nothing() {
4142            let dir = tempdir().unwrap();
4143            let mut db = Database::open(dir.path()).unwrap();
4144            db.create_table("users", schema()).unwrap();
4145            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
4146            db.close().unwrap();
4147
4148            let db = Database::open(dir.path()).unwrap();
4149            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
4150            db.close().unwrap();
4151        }
4152
4153        #[test]
4154        fn the_policy_an_open_call_names_arms_the_tables_it_opens_and_the_ones_it_creates() {
4155            let dir = tempdir().unwrap();
4156            let mut db = opened(dir.path(), CachePolicy::with_entries(64));
4157            db.create_table("users", schema()).unwrap();
4158            assert_eq!(
4159                db.table("users").unwrap().cache_policy(),
4160                CachePolicy::with_entries(64),
4161                "a table created under the handle is armed on the same terms"
4162            );
4163            db.close().unwrap();
4164
4165            let db = opened(dir.path(), CachePolicy::with_entries(64));
4166            assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::with_entries(64));
4167            db.close().unwrap();
4168        }
4169
4170        #[test]
4171        fn a_cached_table_read_through_the_database_answers_the_same_document_twice() {
4172            let dir = tempdir().unwrap();
4173            let mut db = opened(dir.path(), CachePolicy::with_entries(8));
4174            db.create_table("users", schema()).unwrap();
4175            let users = db.table_mut("users").unwrap();
4176            users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
4177
4178            let first = users.get("u1").unwrap();
4179            let second = users.get("u1").unwrap();
4180
4181            assert_eq!(first, second);
4182            assert_eq!(first.unwrap().get("name"), Some(&json!("Ania")));
4183            db.close().unwrap();
4184        }
4185    }
4186
4187    mod cypher {
4188        use super::*;
4189        use crate::crypto::{CipherOptions, MissingKeyPolicy};
4190        use std::collections::HashMap;
4191
4192        fn key(byte: u8) -> Vec<u8> {
4193            vec![byte; 32]
4194        }
4195
4196        fn secrets_schema_json() -> Value {
4197            json!({
4198                "schemaVersion": 1,
4199                "table": "secrets",
4200                "primaryKey": "id",
4201                "fields": {
4202                    "id": { "type": "string" },
4203                    "token": { "type": "string", "cypher": true },
4204                },
4205            })
4206        }
4207
4208        fn secrets_schema_json_with_note() -> Value {
4209            json!({
4210                "schemaVersion": 1,
4211                "table": "secrets",
4212                "primaryKey": "id",
4213                "fields": {
4214                    "id": { "type": "string" },
4215                    "token": { "type": "string", "cypher": true },
4216                    "note": { "type": "string" },
4217                },
4218            })
4219        }
4220
4221        fn open_with_keys(dir: &Path, keys: &[(&str, u8)], key_id: Option<&str>) -> Database {
4222            Database::open_with(
4223                dir,
4224                OpenOptions {
4225                    cypher: Some(CipherOptions {
4226                        keys: keys.iter().map(|(id, byte)| (id.to_string(), key(*byte))).collect(),
4227                        key_id: key_id.map(str::to_string),
4228                        on_missing_key: None,
4229                    }),
4230                    ..OpenOptions::default()
4231                },
4232            )
4233            .unwrap()
4234        }
4235
4236        #[test]
4237        fn a_cypher_field_round_trips_through_open_created_table_and_reopen() {
4238            let dir = tempdir().unwrap();
4239            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4240            db.create_table("secrets", secrets_schema_json()).unwrap();
4241            db.table_mut("secrets")
4242                .unwrap()
4243                .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4244                .unwrap();
4245            db.close().unwrap();
4246
4247            let reopened = open_with_keys(dir.path(), &[("k1", 1)], None);
4248            assert_eq!(
4249                reopened
4250                    .table("secrets")
4251                    .unwrap()
4252                    .get("s1")
4253                    .unwrap()
4254                    .unwrap()
4255                    .get("token"),
4256                Some(&json!("hunter2"))
4257            );
4258            reopened.close().unwrap();
4259        }
4260
4261        #[test]
4262        fn reads_with_an_old_key_and_writes_with_the_new_one_after_rotation() {
4263            let dir = tempdir().unwrap();
4264            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4265            db.create_table("secrets", secrets_schema_json()).unwrap();
4266            db.table_mut("secrets")
4267                .unwrap()
4268                .put(&json!({ "id": "s1", "token": "written-with-k1" }), false)
4269                .unwrap();
4270            db.close().unwrap();
4271
4272            let mut rotated = open_with_keys(dir.path(), &[("k1", 1), ("k2", 2)], Some("k2"));
4273            assert_eq!(
4274                rotated
4275                    .table("secrets")
4276                    .unwrap()
4277                    .get("s1")
4278                    .unwrap()
4279                    .unwrap()
4280                    .get("token"),
4281                Some(&json!("written-with-k1"))
4282            );
4283            rotated
4284                .table_mut("secrets")
4285                .unwrap()
4286                .put(&json!({ "id": "s2", "token": "written-with-k2" }), false)
4287                .unwrap();
4288            rotated.close().unwrap();
4289
4290            let log = std::fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
4291            assert!(log.contains("$cypher$v1$k1$"));
4292            assert!(log.contains("$cypher$v1$k2$"));
4293        }
4294
4295        #[test]
4296        fn keeps_the_keyring_across_drop_field_reopening_the_table() {
4297            let dir = tempdir().unwrap();
4298            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4299            db.create_table("secrets", secrets_schema_json_with_note()).unwrap();
4300            db.table_mut("secrets")
4301                .unwrap()
4302                .put(&json!({ "id": "s1", "token": "hunter2", "note": "x" }), false)
4303                .unwrap();
4304
4305            db.drop_field("secrets", "note", true).unwrap();
4306            assert_eq!(
4307                db.table("secrets").unwrap().get("s1").unwrap().unwrap().get("token"),
4308                Some(&json!("hunter2"))
4309            );
4310            db.close().unwrap();
4311        }
4312
4313        #[test]
4314        fn open_refuses_a_key_that_is_not_32_bytes() {
4315            let dir = tempdir().unwrap();
4316            let error = Database::open_with(
4317                dir.path(),
4318                OpenOptions {
4319                    cypher: Some(CipherOptions {
4320                        keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
4321                        key_id: None,
4322                        on_missing_key: None,
4323                    }),
4324                    ..OpenOptions::default()
4325                },
4326            )
4327            .map(|_| ())
4328            .unwrap_err();
4329            assert!(matches!(error, DatabaseError::Cipher(_)));
4330        }
4331
4332        #[test]
4333        fn missing_key_policy_raw_hands_back_the_envelope_through_the_database() {
4334            let dir = tempdir().unwrap();
4335            let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4336            db.create_table("secrets", secrets_schema_json()).unwrap();
4337            db.table_mut("secrets")
4338                .unwrap()
4339                .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4340                .unwrap();
4341            db.close().unwrap();
4342
4343            let raw = Database::open_with(
4344                dir.path(),
4345                OpenOptions {
4346                    cypher: Some(CipherOptions {
4347                        keys: HashMap::from([("other".to_string(), key(2))]),
4348                        key_id: None,
4349                        on_missing_key: Some(MissingKeyPolicy::Raw),
4350                    }),
4351                    ..OpenOptions::default()
4352                },
4353            )
4354            .unwrap();
4355            let found = raw.table("secrets").unwrap().get("s1").unwrap().unwrap();
4356            assert!(found.get("token").unwrap().as_str().unwrap().contains("$cypher$v1$k1$"));
4357            raw.close().unwrap();
4358        }
4359    }
4360
4361    mod partitions {
4362        use super::*;
4363
4364        fn notes_schema_json(partitioned: bool) -> Value {
4365            let mut spec = json!({
4366                "schemaVersion": 1,
4367                "table": "notes",
4368                "primaryKey": "id",
4369                "fields": {
4370                    "id": { "type": "string" },
4371                    "kind": { "type": "string", "required": true },
4372                    "title": { "type": "string", "required": true },
4373                    "body": { "type": "string" },
4374                },
4375            });
4376            if partitioned {
4377                spec.as_object_mut()
4378                    .unwrap()
4379                    .insert("partitionBy".into(), json!("kind"));
4380            }
4381            spec
4382        }
4383
4384        fn lines_in(dir: &Path, part: &str) -> usize {
4385            fs::read_to_string(dir.join("notes").join(format!("{part}.jsonl")))
4386                .unwrap()
4387                .trim()
4388                .lines()
4389                .filter(|line| !line.is_empty())
4390                .count()
4391        }
4392
4393        #[test]
4394        fn a_partitioned_table_survives_the_catalog_round_trip() {
4395            let dir = tempdir().unwrap();
4396            {
4397                let mut db = Database::open(dir.path()).unwrap();
4398                db.create_table("notes", notes_schema_json(true)).unwrap();
4399                let notes = db.table_mut("notes").unwrap();
4400                notes
4401                    .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4402                    .unwrap();
4403                notes
4404                    .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4405                    .unwrap();
4406                db.close().unwrap();
4407            }
4408
4409            let db = Database::open(dir.path()).unwrap();
4410            let notes = db.table("notes").unwrap();
4411            assert_eq!(notes.schema().partition_by.as_deref(), Some("kind"));
4412            assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4413            assert_eq!(lines_in(dir.path(), "note"), 1);
4414            assert_eq!(lines_in(dir.path(), "adr"), 1);
4415            assert!(!dir.path().join("notes.jsonl").exists());
4416            db.close().unwrap();
4417        }
4418
4419        #[test]
4420        fn declaring_partitions_on_an_existing_table_moves_its_records_into_the_new_layout() {
4421            let dir = tempdir().unwrap();
4422            {
4423                let mut db = Database::open(dir.path()).unwrap();
4424                db.create_table("notes", notes_schema_json(false)).unwrap();
4425                let notes = db.table_mut("notes").unwrap();
4426                notes
4427                    .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4428                    .unwrap();
4429                notes
4430                    .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4431                    .unwrap();
4432                notes.delete("n1", false).unwrap();
4433                db.close().unwrap();
4434            }
4435
4436            let mut db = Database::open(dir.path()).unwrap();
4437            let mut partitioned = notes_schema_json(true);
4438            partitioned.as_object_mut().unwrap().insert("revision".into(), json!(2));
4439            db.create_table("notes", partitioned).unwrap();
4440
4441            let notes = db.table("notes").unwrap();
4442            assert_eq!(notes.get("n1").unwrap(), None);
4443            assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4444            assert!(!dir.path().join("notes.jsonl").exists());
4445            assert_eq!(lines_in(dir.path(), "adr"), 1);
4446            db.close().unwrap();
4447        }
4448
4449        #[test]
4450        fn dropping_the_field_the_table_partitions_by_is_refused() {
4451            let dir = tempdir().unwrap();
4452            let mut db = Database::open(dir.path()).unwrap();
4453            db.create_table("notes", notes_schema_json(true)).unwrap();
4454            db.table_mut("notes")
4455                .unwrap()
4456                .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4457                .unwrap();
4458
4459            let error = db.drop_field("notes", "kind", true).unwrap_err();
4460            assert!(error.to_string().contains("partitioned by"));
4461            assert_eq!(
4462                db.table("notes").unwrap().get("n1").unwrap().unwrap().get("kind"),
4463                Some(&json!("note"))
4464            );
4465            db.close().unwrap();
4466        }
4467
4468        #[test]
4469        fn dropping_an_unrelated_field_keeps_every_partition_readable() {
4470            let dir = tempdir().unwrap();
4471            let mut db = Database::open(dir.path()).unwrap();
4472            db.create_table("notes", notes_schema_json(true)).unwrap();
4473            {
4474                let notes = db.table_mut("notes").unwrap();
4475                notes
4476                    .put(
4477                        &json!({ "id": "n1", "kind": "note", "title": "first", "body": "gone soon" }),
4478                        false,
4479                    )
4480                    .unwrap();
4481                notes
4482                    .put(&json!({ "id": "a1", "kind": "adr", "title": "kept" }), false)
4483                    .unwrap();
4484            }
4485
4486            db.drop_field("notes", "body", true).unwrap();
4487
4488            let notes = db.table("notes").unwrap();
4489            assert!(!notes.schema().fields.contains_key("body"));
4490            assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("first")));
4491            assert_eq!(notes.get("a1").unwrap().unwrap().get("kind"), Some(&json!("adr")));
4492            assert_eq!(lines_in(dir.path(), "note"), 1);
4493            assert_eq!(lines_in(dir.path(), "adr"), 1);
4494            db.close().unwrap();
4495        }
4496
4497        #[test]
4498        fn a_unit_of_work_lands_in_two_partitions_of_one_table() {
4499            let dir = tempdir().unwrap();
4500            let mut db = Database::open(dir.path()).unwrap();
4501            db.create_table("notes", notes_schema_json(true)).unwrap();
4502
4503            let mut unit = db.unit_of_work();
4504            unit.put("notes", json!({ "id": "n1", "kind": "note", "title": "from unit" }));
4505            unit.put("notes", json!({ "id": "a1", "kind": "adr", "title": "from unit too" }));
4506            let result = db.commit(&mut unit).unwrap();
4507            assert_eq!(result.operations, 2);
4508
4509            let notes = db.table("notes").unwrap();
4510            assert_eq!(
4511                notes.get("n1").unwrap().unwrap().get("title"),
4512                Some(&json!("from unit"))
4513            );
4514            assert_eq!(
4515                notes.get("a1").unwrap().unwrap().get("title"),
4516                Some(&json!("from unit too"))
4517            );
4518            assert_eq!(lines_in(dir.path(), "note"), 1);
4519            assert_eq!(lines_in(dir.path(), "adr"), 1);
4520            db.close().unwrap();
4521        }
4522    }
4523    mod following_the_log {
4524        use super::*;
4525
4526        fn readonly(follow: FollowOption) -> OpenOptions {
4527            OpenOptions {
4528                readonly: true,
4529                follow,
4530                ..Default::default()
4531            }
4532        }
4533
4534        /// A writer and a reader on the same directory, as two separate handles — the shape of the
4535        /// TS `multiprocess.test.ts`, which runs the writer in a child process for the lock's sake.
4536        /// One process is enough here: the lock is only taken by the writable handle.
4537        fn writer_and_reader(dir: &Path, follow: FollowOption) -> (Database, Database) {
4538            let mut writer = Database::open(dir).unwrap();
4539            writer.create_table("users", users_schema_spec()).unwrap();
4540            writer
4541                .table_mut("users")
4542                .unwrap()
4543                .put(&json!({ "id": "u1", "name": "Ania" }), false)
4544                .unwrap();
4545            let reader = Database::open_with(dir, readonly(follow)).unwrap();
4546            (writer, reader)
4547        }
4548
4549        fn users_schema_spec() -> Value {
4550            json!({
4551                "schemaVersion": 1,
4552                "table": "users",
4553                "primaryKey": "id",
4554                "fields": {
4555                    "id": { "type": "string" },
4556                    "name": { "type": "string", "required": true },
4557                },
4558            })
4559        }
4560
4561        fn put(writer: &mut Database, key: &str, name: &str) {
4562            writer
4563                .table_mut("users")
4564                .unwrap()
4565                .put(&json!({ "id": key, "name": name }), false)
4566                .unwrap();
4567        }
4568
4569        #[test]
4570        fn a_readonly_open_follows_by_default_and_a_writable_one_does_not() {
4571            let dir = tempdir().unwrap();
4572            let mut writer = Database::open(dir.path()).unwrap();
4573            writer.create_table("users", users_schema_spec()).unwrap();
4574
4575            assert_eq!(writer.follow(), None);
4576            assert_eq!(writer.table("users").unwrap().follow(), None);
4577
4578            let reader = Database::open_with(dir.path(), readonly(FollowOption::Default)).unwrap();
4579            assert_eq!(
4580                reader.follow(),
4581                Some(FollowPolicy {
4582                    interval: DEFAULT_FOLLOW_INTERVAL
4583                })
4584            );
4585            assert_eq!(
4586                reader.table("users").unwrap().follow(),
4587                Some(FollowPolicy {
4588                    interval: DEFAULT_FOLLOW_INTERVAL
4589                })
4590            );
4591        }
4592
4593        #[test]
4594        fn a_plain_get_on_a_following_reader_sees_what_the_writer_committed_after_it_opened() {
4595            let dir = tempdir().unwrap();
4596            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4597
4598            for index in 2..=5 {
4599                put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
4600                let found = reader
4601                    .table_mut("users")
4602                    .unwrap()
4603                    .get_following(&format!("u{index}"))
4604                    .unwrap();
4605                assert_eq!(found.unwrap().get("name"), Some(&json!(format!("Osoba {index}"))));
4606            }
4607        }
4608
4609        #[test]
4610        fn a_plain_query_on_a_following_reader_counts_what_the_writer_committed() {
4611            let dir = tempdir().unwrap();
4612            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4613
4614            let spec = QueryBuilder::new().spec();
4615            assert_eq!(
4616                reader
4617                    .table_mut("users")
4618                    .unwrap()
4619                    .query_following(&spec)
4620                    .unwrap()
4621                    .matched,
4622                1
4623            );
4624
4625            put(&mut writer, "u2", "Osoba 2");
4626
4627            assert_eq!(
4628                reader
4629                    .table_mut("users")
4630                    .unwrap()
4631                    .query_following(&spec)
4632                    .unwrap()
4633                    .matched,
4634                2
4635            );
4636        }
4637
4638        #[test]
4639        fn a_reader_with_follow_off_stands_still_until_refresh_is_called() {
4640            let dir = tempdir().unwrap();
4641            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Off);
4642            put(&mut writer, "u2", "Osoba 2");
4643
4644            assert!(reader
4645                .table_mut("users")
4646                .unwrap()
4647                .get_following("u2")
4648                .unwrap()
4649                .is_none());
4650            assert!(reader.table_mut("users").unwrap().refresh().unwrap().changed());
4651            assert!(reader
4652                .table_mut("users")
4653                .unwrap()
4654                .get_following("u2")
4655                .unwrap()
4656                .is_some());
4657        }
4658
4659        #[test]
4660        fn a_custom_interval_coalesces_the_freshness_check_and_lets_it_through_once_it_elapses() {
4661            let dir = tempdir().unwrap();
4662            let (mut writer, mut reader) =
4663                writer_and_reader(dir.path(), FollowOption::Every(Duration::from_millis(150)));
4664
4665            // The first read is always due — nothing has been followed yet — and starts the clock.
4666            assert!(reader
4667                .table_mut("users")
4668                .unwrap()
4669                .get_following("u1")
4670                .unwrap()
4671                .is_some());
4672            put(&mut writer, "u2", "Osoba 2");
4673            assert!(reader
4674                .table_mut("users")
4675                .unwrap()
4676                .get_following("u2")
4677                .unwrap()
4678                .is_none());
4679
4680            std::thread::sleep(Duration::from_millis(200));
4681
4682            assert!(reader
4683                .table_mut("users")
4684                .unwrap()
4685                .get_following("u2")
4686                .unwrap()
4687                .is_some());
4688        }
4689
4690        #[test]
4691        fn get_many_and_the_aggregates_follow_the_log_the_same_way_a_single_get_does() {
4692            let dir = tempdir().unwrap();
4693            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4694            put(&mut writer, "u2", "Osoba 2");
4695
4696            let keys = vec!["u1".to_string(), "u2".to_string()];
4697            let rows = reader.table_mut("users").unwrap().get_many_following(&keys).unwrap();
4698            assert!(rows.iter().all(Option::is_some), "{rows:?}");
4699
4700            put(&mut writer, "u3", "Osoba 3");
4701            let spec = QueryBuilder::new().spec();
4702            let aggregate = crate::query::aggregate::AggregateSpec {
4703                count: true,
4704                ..Default::default()
4705            };
4706            let counted = reader
4707                .table_mut("users")
4708                .unwrap()
4709                .aggregate_following(&spec, &aggregate)
4710                .unwrap();
4711            assert_eq!(counted.count, 3);
4712        }
4713
4714        #[test]
4715        fn a_following_reader_reloads_instead_of_reading_the_offsets_a_foreign_compaction_moved() {
4716            let dir = tempdir().unwrap();
4717            let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4718            for index in 2..=6 {
4719                put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
4720            }
4721            writer.table_mut("users").unwrap().delete("u5", false).unwrap();
4722            assert!(reader
4723                .table_mut("users")
4724                .unwrap()
4725                .get_following("u6")
4726                .unwrap()
4727                .is_some());
4728
4729            writer.table_mut("users").unwrap().compact().unwrap();
4730
4731            let users = reader.table_mut("users").unwrap();
4732            assert_eq!(
4733                users.get_following("u3").unwrap().unwrap().get("name"),
4734                Some(&json!("Osoba 3"))
4735            );
4736            assert!(users.get_following("u5").unwrap().is_none());
4737            assert_eq!(users.query_following(&QueryBuilder::new().spec()).unwrap().matched, 5);
4738        }
4739    }
4740
4741    mod where_the_files_land {
4742        use super::*;
4743        use crate::database::LayoutKind;
4744
4745        fn notes_schema_spec() -> Value {
4746            json!({
4747                "schemaVersion": 1,
4748                "table": "notes",
4749                "primaryKey": "id",
4750                "partitionBy": "kind",
4751                "fields": {
4752                    "id": { "type": "string" },
4753                    "kind": { "type": "string", "required": true, "index": "exact" },
4754                    "title": { "type": "string", "required": true },
4755                },
4756            })
4757        }
4758
4759        fn nested(metadata_file: Option<&str>) -> OpenOptions {
4760            OpenOptions {
4761                layout: LayoutOptions {
4762                    kind: Some(LayoutKind::Nested),
4763                    metadata_file: metadata_file.map(str::to_string),
4764                    ..Default::default()
4765                },
4766                ..Default::default()
4767            }
4768        }
4769
4770        #[test]
4771        fn the_default_open_writes_exactly_what_it_wrote_before_layouts_existed() {
4772            let dir = tempdir().unwrap();
4773            let mut db = Database::open(dir.path()).unwrap();
4774            db.create_table(
4775                "users",
4776                json!({
4777                    "schemaVersion": 1,
4778                    "table": "users",
4779                    "primaryKey": "id",
4780                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4781                }),
4782            )
4783            .unwrap();
4784            db.table_mut("users")
4785                .unwrap()
4786                .put(&json!({ "id": "u1", "name": "Ania" }), false)
4787                .unwrap();
4788            db.close().unwrap();
4789
4790            assert!(dir.path().join("hayson.json").is_file());
4791            assert!(dir.path().join("users.jsonl").is_file());
4792            assert!(!dir.path().join("tables").exists());
4793            assert!(!dir.path().join("runtime").exists());
4794
4795            let db = Database::open(dir.path()).unwrap();
4796            assert_eq!(db.layout().kind(), LayoutKind::Flat);
4797            assert_eq!(db.layout().metadata_file(), "hayson.json");
4798            assert_eq!(db.layout().metadata_path(), dir.path().join("hayson.json"));
4799        }
4800
4801        #[test]
4802        fn a_custom_metadata_file_holds_the_catalog_and_nothing_is_written_under_the_default_name() {
4803            let dir = tempdir().unwrap();
4804            let options = OpenOptions {
4805                layout: LayoutOptions {
4806                    metadata_file: Some("cerebro.json".into()),
4807                    ..Default::default()
4808                },
4809                ..Default::default()
4810            };
4811
4812            let mut db = Database::open_with(dir.path(), options.clone()).unwrap();
4813            db.create_table(
4814                "users",
4815                json!({
4816                    "schemaVersion": 1,
4817                    "table": "users",
4818                    "primaryKey": "id",
4819                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4820                }),
4821            )
4822            .unwrap();
4823            db.table_mut("users")
4824                .unwrap()
4825                .put(&json!({ "id": "u1", "name": "Ania" }), false)
4826                .unwrap();
4827            db.close().unwrap();
4828
4829            assert!(dir.path().join("cerebro.json").is_file());
4830            assert!(!dir.path().join("hayson.json").exists());
4831
4832            let db = Database::open_with(dir.path(), options).unwrap();
4833            assert_eq!(db.layout().metadata_file(), "cerebro.json");
4834            assert_eq!(db.table_names(), vec!["users"]);
4835            assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
4836        }
4837
4838        #[test]
4839        fn the_deprecated_manifest_name_opens_what_the_new_one_wrote_and_the_other_way_round() {
4840            let dir = tempdir().unwrap();
4841            let legacy = OpenOptions {
4842                layout: LayoutOptions {
4843                    manifest: Some("cerebro.json".into()),
4844                    ..Default::default()
4845                },
4846                ..Default::default()
4847            };
4848            let mut db = Database::open_with(dir.path(), legacy).unwrap();
4849            db.create_table(
4850                "users",
4851                json!({
4852                    "schemaVersion": 1,
4853                    "table": "users",
4854                    "primaryKey": "id",
4855                    "fields": { "id": { "type": "string" } },
4856                }),
4857            )
4858            .unwrap();
4859            db.close().unwrap();
4860
4861            let current = OpenOptions {
4862                layout: LayoutOptions {
4863                    metadata_file: Some("cerebro.json".into()),
4864                    ..Default::default()
4865                },
4866                ..Default::default()
4867            };
4868            let db = Database::open_with(dir.path(), current).unwrap();
4869            assert_eq!(db.table_names(), vec!["users"]);
4870            assert_eq!(db.layout().metadata_path(), dir.path().join("cerebro.json"));
4871        }
4872
4873        #[test]
4874        fn the_two_names_for_the_metadata_document_are_refused_when_they_disagree_touching_nothing() {
4875            let dir = tempdir().unwrap();
4876            let options = OpenOptions {
4877                layout: LayoutOptions {
4878                    metadata_file: Some("cerebro.json".into()),
4879                    manifest: Some("hayson.json".into()),
4880                    ..Default::default()
4881                },
4882                ..Default::default()
4883            };
4884
4885            let Err(error) = Database::open_with(dir.path(), options) else {
4886                panic!("two disagreeing names for the metadata document were accepted")
4887            };
4888
4889            assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
4890            let message = error.to_string();
4891            assert!(
4892                message.contains("metadataFile") && message.contains("manifest"),
4893                "{message}"
4894            );
4895            assert!(!dir.path().join("hayson.json").exists());
4896            assert!(!dir.path().join("cerebro.json").exists());
4897            assert!(!dir.path().join(crate::lock::LOCK_FILE).exists());
4898        }
4899
4900        #[test]
4901        fn the_nested_layout_sorts_the_logs_the_snapshots_and_the_lock_into_their_subdirectories() {
4902            let dir = tempdir().unwrap();
4903            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
4904            db.create_table(
4905                "users",
4906                json!({
4907                    "schemaVersion": 1,
4908                    "table": "users",
4909                    "primaryKey": "id",
4910                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4911                }),
4912            )
4913            .unwrap();
4914            db.table_mut("users")
4915                .unwrap()
4916                .put(&json!({ "id": "u1", "name": "Ania" }), false)
4917                .unwrap();
4918
4919            assert!(dir.path().join("runtime").join(crate::lock::LOCK_FILE).is_file());
4920            db.close().unwrap();
4921
4922            assert!(dir.path().join("hayson.json").is_file());
4923            assert!(dir.path().join("tables").join("users.jsonl").is_file());
4924            assert!(dir.path().join("index").join("users.idx").is_file());
4925            assert!(!dir.path().join("users.jsonl").exists());
4926        }
4927
4928        #[test]
4929        fn a_partitioned_table_under_the_nested_layout_lands_under_tables_slash_its_own_name() {
4930            let dir = tempdir().unwrap();
4931            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
4932            db.create_table("notes", notes_schema_spec()).unwrap();
4933            db.table_mut("notes")
4934                .unwrap()
4935                .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
4936                .unwrap();
4937            db.close().unwrap();
4938
4939            assert!(dir.path().join("tables").join("notes").join("note.jsonl").is_file());
4940        }
4941
4942        #[test]
4943        fn a_nested_database_reopens_as_nested_without_being_told_and_is_refused_as_flat() {
4944            let dir = tempdir().unwrap();
4945            let mut db = Database::open_with(dir.path(), nested(Some("cerebro.json"))).unwrap();
4946            db.create_table(
4947                "users",
4948                json!({
4949                    "schemaVersion": 1,
4950                    "table": "users",
4951                    "primaryKey": "id",
4952                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4953                }),
4954            )
4955            .unwrap();
4956            db.table_mut("users")
4957                .unwrap()
4958                .put(&json!({ "id": "u1", "name": "Ania" }), false)
4959                .unwrap();
4960            db.close().unwrap();
4961
4962            let named_only = OpenOptions {
4963                layout: LayoutOptions {
4964                    metadata_file: Some("cerebro.json".into()),
4965                    ..Default::default()
4966                },
4967                ..Default::default()
4968            };
4969            let db = Database::open_with(dir.path(), named_only).unwrap();
4970            assert_eq!(db.layout().kind(), LayoutKind::Nested);
4971            assert_eq!(
4972                db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
4973                Some(&json!("Ania"))
4974            );
4975            db.close().unwrap();
4976
4977            let as_flat = OpenOptions {
4978                layout: LayoutOptions {
4979                    kind: Some(LayoutKind::Flat),
4980                    metadata_file: Some("cerebro.json".into()),
4981                    ..Default::default()
4982                },
4983                ..Default::default()
4984            };
4985            let Err(error) = Database::open_with(dir.path(), as_flat) else {
4986                panic!("a nested database was opened as flat")
4987            };
4988            assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
4989            assert!(error.to_string().contains("nested"), "{error}");
4990        }
4991
4992        #[test]
4993        fn a_unit_of_work_under_the_nested_layout_journals_into_runtime_and_reopens_whole() {
4994            let dir = tempdir().unwrap();
4995            let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
4996            db.create_table(
4997                "users",
4998                json!({
4999                    "schemaVersion": 1,
5000                    "table": "users",
5001                    "primaryKey": "id",
5002                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5003                }),
5004            )
5005            .unwrap();
5006
5007            let mut unit = db.unit_of_work();
5008            unit.put("users", json!({ "id": "u1", "name": "Ania" }));
5009            unit.put("users", json!({ "id": "u2", "name": "Bartek" }));
5010            db.commit(&mut unit).unwrap();
5011            db.close().unwrap();
5012
5013            assert!(!dir.path().join("runtime").join(".hayson.txn").exists());
5014            let db = Database::open_with(dir.path(), nested(None)).unwrap();
5015            assert_eq!(db.table("users").unwrap().len(), 2);
5016        }
5017    }
5018
5019    /// F8: a catalog naming a table whose log is not on disk. A writable handle makes the log it
5020    /// misses, so the two open modes owe the caller opposite answers here.
5021    mod a_declared_table_whose_log_is_not_on_disk {
5022        use super::*;
5023
5024        fn notes_spec() -> Value {
5025            json!({
5026                "schemaVersion": 1,
5027                "table": "notes",
5028                "primaryKey": "id",
5029                "partitionBy": "kind",
5030                "fields": {
5031                    "id": { "type": "string" },
5032                    "kind": { "type": "string", "required": true },
5033                    "title": { "type": "string", "required": true },
5034                },
5035            })
5036        }
5037
5038        fn readonly() -> OpenOptions {
5039            OpenOptions {
5040                readonly: true,
5041                ..Default::default()
5042            }
5043        }
5044
5045        fn users_with_a_record(dir: &Path) {
5046            let mut db = Database::open(dir).unwrap();
5047            db.create_table(
5048                "users",
5049                json!({
5050                    "schemaVersion": 1,
5051                    "table": "users",
5052                    "primaryKey": "id",
5053                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5054                }),
5055            )
5056            .unwrap();
5057            db.table_mut("users")
5058                .unwrap()
5059                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5060                .unwrap();
5061            db.close().unwrap();
5062        }
5063
5064        fn notes_with_a_record(dir: &Path) {
5065            let mut db = Database::open(dir).unwrap();
5066            db.create_table("notes", notes_spec()).unwrap();
5067            db.table_mut("notes")
5068                .unwrap()
5069                .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
5070                .unwrap();
5071            db.close().unwrap();
5072        }
5073
5074        fn refusal(dir: &Path) -> DatabaseError {
5075            match Database::open_with(dir, readonly()) {
5076                Err(error) => error,
5077                Ok(_) => panic!("expected \"{}\" to be refused", dir.display()),
5078            }
5079        }
5080
5081        fn schema_refusal(error: DatabaseError) -> String {
5082            match error {
5083                DatabaseError::Schema(HaysonSchemaError(message)) => message,
5084                other => panic!("expected a schema refusal, got {other:?}"),
5085            }
5086        }
5087
5088        #[test]
5089        fn a_readonly_open_refuses_naming_the_table_and_the_log_without_advising_create_table() {
5090            let dir = tempdir().unwrap();
5091            users_with_a_record(dir.path());
5092            let log = log_path(dir.path(), "users");
5093            fs::remove_file(&log).unwrap();
5094
5095            let message = schema_refusal(refusal(dir.path()));
5096
5097            assert!(message.contains("\"users\""), "{message}");
5098            assert!(message.contains(&log.display().to_string()), "{message}");
5099            assert!(message.contains("readonly"), "{message}");
5100            assert!(!message.contains("createTable"), "{message}");
5101            assert!(!log.exists(), "the refused open must not have created the log");
5102        }
5103
5104        #[test]
5105        fn a_writable_open_of_the_same_directory_creates_the_log_and_serves_an_empty_table() {
5106            let dir = tempdir().unwrap();
5107            users_with_a_record(dir.path());
5108            fs::remove_file(log_path(dir.path(), "users")).unwrap();
5109
5110            let mut db = Database::open(dir.path()).unwrap();
5111
5112            assert_eq!(db.table_names(), vec!["users"]);
5113            assert!(log_path(dir.path(), "users").is_file());
5114            assert_eq!(db.table("users").unwrap().len(), 0);
5115            db.table_mut("users")
5116                .unwrap()
5117                .put(&json!({ "id": "u1", "name": "Ania" }), false)
5118                .unwrap();
5119            assert_eq!(db.table("users").unwrap().len(), 1);
5120            db.close().unwrap();
5121        }
5122
5123        // NOTE: an empty log is a table nothing was written to; a missing one is a table gone.
5124        #[test]
5125        fn a_readonly_open_accepts_a_log_that_is_present_and_empty() {
5126            let dir = tempdir().unwrap();
5127            users_with_a_record(dir.path());
5128            fs::write(log_path(dir.path(), "users"), "").unwrap();
5129
5130            let db = Database::open_with(dir.path(), readonly()).unwrap();
5131
5132            assert_eq!(db.table_names(), vec!["users"]);
5133            assert_eq!(db.table("users").unwrap().len(), 0);
5134        }
5135
5136        #[test]
5137        fn a_readonly_open_refuses_a_partitioned_table_whose_recorded_partition_is_gone() {
5138            let dir = tempdir().unwrap();
5139            notes_with_a_record(dir.path());
5140            let partition = dir.path().join("notes").join("note.jsonl");
5141            assert!(
5142                index_path(dir.path(), "notes").is_file(),
5143                "the refusal reads the snapshot"
5144            );
5145            fs::remove_dir_all(dir.path().join("notes")).unwrap();
5146
5147            let message = schema_refusal(refusal(dir.path()));
5148
5149            assert!(message.contains("\"notes\""), "{message}");
5150            assert!(message.contains(&partition.display().to_string()), "{message}");
5151            assert!(message.contains("readonly"), "{message}");
5152            assert!(!message.contains("createTable"), "{message}");
5153        }
5154
5155        #[test]
5156        fn a_readonly_open_accepts_a_partitioned_table_that_was_never_written_to() {
5157            let dir = tempdir().unwrap();
5158            let mut db = Database::open(dir.path()).unwrap();
5159            db.create_table("notes", notes_spec()).unwrap();
5160            db.close().unwrap();
5161
5162            let db = Database::open_with(dir.path(), readonly()).unwrap();
5163
5164            assert_eq!(db.table_names(), vec!["notes"]);
5165            assert_eq!(db.table("notes").unwrap().len(), 0);
5166        }
5167
5168        /// A partition the listing shows but the filesystem cannot deliver is corruption, not a
5169        /// table that opens empty — and the log set would otherwise create a fresh file over it.
5170        #[cfg(unix)]
5171        #[test]
5172        fn a_readonly_open_reports_a_partition_that_cannot_be_opened_against_its_own_path() {
5173            let dir = tempdir().unwrap();
5174            notes_with_a_record(dir.path());
5175            let partition = dir.path().join("notes").join("note.jsonl");
5176            fs::remove_file(&partition).unwrap();
5177            std::os::unix::fs::symlink(dir.path().join("notes").join("gone.jsonl"), &partition).unwrap();
5178
5179            let error = refusal(dir.path());
5180
5181            match error {
5182                DatabaseError::Log(path, inner) => {
5183                    assert_eq!(path, partition);
5184                    assert_eq!(inner.kind(), io::ErrorKind::NotFound);
5185                }
5186                other => panic!("expected a log refusal naming the partition, got {other:?}"),
5187            }
5188            assert!(!dir.path().join("notes").join("gone.jsonl").exists());
5189        }
5190    }
5191
5192    mod describing_one_table {
5193        use super::*;
5194
5195        fn users_with_city() -> TableSchema {
5196            parse_schema(&json!({
5197                "schemaVersion": 1,
5198                "table": "users",
5199                "revision": 2,
5200                "primaryKey": "id",
5201                "fields": {
5202                    "id": { "type": "string" },
5203                    "name": { "type": "string", "required": true },
5204                    "city": { "type": "string", "index": "exact" },
5205                },
5206            }))
5207            .unwrap()
5208        }
5209
5210        /// A handle over both tables, with the catalog on disk moved on under it for `users`
5211        /// alone — the drift every gate here has to tell apart from a healthy `orders`.
5212        fn users_drifted_under(dir: &Path) -> Database {
5213            write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
5214            let mut db = Database::open(dir).unwrap();
5215            write_catalog_to(dir, &catalog_of(&[users_with_city(), orders_schema()]));
5216            assert!(db.refresh().unwrap());
5217            db
5218        }
5219
5220        fn refusal_of(answered: Result<Value, DatabaseError>) -> String {
5221            match answered {
5222                Err(DatabaseError::Schema(error)) => error.to_string(),
5223                other => panic!("expected a schema refusal, got {other:?}"),
5224            }
5225        }
5226
5227        #[test]
5228        fn describes_a_table_of_its_own_while_another_one_has_drifted() {
5229            let dir = tempdir().unwrap();
5230            let db = users_drifted_under(dir.path());
5231
5232            let document = db.describe_table("orders").unwrap();
5233
5234            assert_eq!(document["table"], json!("orders"));
5235            assert_eq!(document["revision"], json!(1));
5236            assert!(db.catalog().is_err());
5237        }
5238
5239        #[test]
5240        fn refuses_the_table_that_drifted_itself() {
5241            let dir = tempdir().unwrap();
5242            let db = users_drifted_under(dir.path());
5243
5244            let message = refusal_of(db.describe_table("users"));
5245
5246            assert!(message.contains("\"users\""), "{message}");
5247            assert!(message.contains("revision 2"), "{message}");
5248            // NOTE: refusing replaces the reload, so the stale handle stays stale.
5249            assert_eq!(db.table("users").unwrap().schema().revision, 1);
5250        }
5251
5252        #[test]
5253        fn describes_the_index_set_the_handle_serves_and_leaves_the_named_ones_out() {
5254            let dir = tempdir().unwrap();
5255            let mut catalog = catalog_of(&[users_schema()]);
5256            let entry = catalog.tables.get_mut("users").unwrap();
5257            entry.indexes = Some(vec![IndexEntry {
5258                name: "by_name".to_string(),
5259                fields: vec!["name".to_string()],
5260                kind: crate::catalog::CatalogIndexKind::Exact,
5261                unique: false,
5262                where_: None,
5263                expire_after_seconds: None,
5264            }]);
5265            write_catalog_to(dir.path(), &catalog);
5266            let db = Database::open(dir.path()).unwrap();
5267
5268            let document = db.describe_table("users").unwrap();
5269
5270            assert_eq!(
5271                document["fields"]["name"],
5272                json!({ "type": "string", "required": true })
5273            );
5274            assert!(document.as_object().unwrap().get("indexes").is_none());
5275            assert!(db.catalog().is_ok());
5276        }
5277
5278        #[test]
5279        fn refuses_a_table_this_handle_holds_nowhere() {
5280            let dir = tempdir().unwrap();
5281            write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
5282            let db = Database::open(dir.path()).unwrap();
5283
5284            let message = refusal_of(db.describe_table("ghosts"));
5285
5286            assert!(message.contains("declares no table \"ghosts\""), "{message}");
5287        }
5288    }
5289}