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