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