hayson_engine/catalog/
migrate.rs

1use std::collections::{BTreeMap, HashSet};
2use std::fmt;
3use std::fs;
4use std::path::{Path, PathBuf};
5
6use serde_json::Value;
7
8use crate::errors::HaysonSchemaError;
9use crate::registry::table_registry::TableRegistry;
10use crate::schema::parse::with_current_schema_marker;
11use crate::schema::{is_partitioned, parse_schema, TableSchema, UNPARTITIONED};
12use crate::storage::{LogFile, LOG_SUFFIX};
13
14use super::build::table_entry_of;
15use super::types::{Catalog, CatalogLayout, CATALOG_FORMAT_VERSION, CATALOG_VERSION};
16
17const JOURNAL_FILE_NAME: &str = ".hayson.txn";
18const SCHEMA_FILE_SUFFIX: &str = ".schema.json";
19
20fn journal_paths_of(dir: &Path) -> Vec<PathBuf> {
21    vec![dir.join(JOURNAL_FILE_NAME), dir.join("runtime").join(JOURNAL_FILE_NAME)]
22}
23
24fn assert_journal_gone(dir: &Path) -> Result<(), HaysonSchemaError> {
25    for path in journal_paths_of(dir) {
26        if path.exists() {
27            return Err(HaysonSchemaError(format!(
28                "refusing to migrate to the metadata catalog while \"{}\" is on disk: an unfinished unit of work is redone through the layout it was written in, so it has to be finished before the format changes — nothing on disk was changed",
29                path.display()
30            )));
31        }
32    }
33    Ok(())
34}
35
36fn layout_of(dir: &Path) -> CatalogLayout {
37    let manifest = dir.join("hayson.json");
38    let Ok(contents) = fs::read_to_string(&manifest) else {
39        return CatalogLayout::Flat;
40    };
41    let Ok(value) = serde_json::from_str::<Value>(&contents) else {
42        return CatalogLayout::Flat;
43    };
44    match value.get("layout").and_then(Value::as_str) {
45        Some("nested") => CatalogLayout::Nested,
46        _ => CatalogLayout::Flat,
47    }
48}
49
50fn schema_dirs_of(dir: &Path) -> Vec<PathBuf> {
51    let nested = dir.join("schema");
52    if nested.is_dir() {
53        vec![nested]
54    } else {
55        vec![dir.to_path_buf()]
56    }
57}
58
59/// Every `<table>.schema.json` a format-1 database at `dir` declares a table through, sorted.
60/// [`migrate_format_1_to_2`] reads exactly these, so the caller that persists the catalog it
61/// returns can retire exactly these once the catalog is safely on disk.
62pub fn schema_files_of(dir: &Path) -> Result<Vec<PathBuf>, HaysonSchemaError> {
63    let mut files = Vec::new();
64    for schema_dir in schema_dirs_of(dir) {
65        let entries = fs::read_dir(&schema_dir).map_err(|error| {
66            HaysonSchemaError(format!("reading \"{}\" to migrate it: {error}", schema_dir.display()))
67        })?;
68        for entry in entries {
69            let entry = entry.map_err(|error| {
70                HaysonSchemaError(format!("reading \"{}\" to migrate it: {error}", schema_dir.display()))
71            })?;
72            let path = entry.path();
73            let is_schema_file = path
74                .file_name()
75                .and_then(|name| name.to_str())
76                .is_some_and(|name| name.ends_with(SCHEMA_FILE_SUFFIX));
77            if is_schema_file {
78                files.push(path);
79            }
80        }
81    }
82    files.sort();
83    Ok(files)
84}
85
86fn tables_dir_of(dir: &Path, layout: CatalogLayout) -> PathBuf {
87    match layout {
88        CatalogLayout::Nested => dir.join("tables"),
89        CatalogLayout::Flat => dir.to_path_buf(),
90    }
91}
92
93fn partition_logs_in(dir: &Path) -> Result<Vec<(String, PathBuf)>, HaysonSchemaError> {
94    let reading = |error: std::io::Error| {
95        HaysonSchemaError(format!(
96            "reading \"{}\" to check unique fields before migrating it: {error}",
97            dir.display()
98        ))
99    };
100    // NOTE: read_dir here, not partition_names_in() — that one reports an unreadable dir as empty.
101    let mut logs = Vec::new();
102    for entry in fs::read_dir(dir).map_err(reading)? {
103        let name = entry.map_err(reading)?.file_name();
104        let Some(part) = name.to_str().and_then(|name| name.strip_suffix(LOG_SUFFIX)) else {
105            continue;
106        };
107        if !part.is_empty() {
108            logs.push((part.to_string(), dir.join(&name)));
109        }
110    }
111    logs.sort();
112    Ok(logs)
113}
114
115/// Every log file a format-1 table keeps its records in, by partition name, for each of the layout
116/// shapes a format-1 database can be on disk in: flat or `nested` (`docs/format.md`, "Układ
117/// katalogu"), partitioned or not. An empty answer means the table has no log at all — nothing was
118/// ever written to it — and never "the log is somewhere this cannot look".
119fn log_paths_of(
120    dir: &Path,
121    layout: CatalogLayout,
122    schema: &TableSchema,
123) -> Result<Vec<(String, PathBuf)>, HaysonSchemaError> {
124    let tables = tables_dir_of(dir, layout);
125    let flat = tables.join(format!("{}{LOG_SUFFIX}", schema.table));
126    let flat_only = || {
127        if flat.exists() {
128            vec![(UNPARTITIONED.to_string(), flat.clone())]
129        } else {
130            Vec::new()
131        }
132    };
133    if !is_partitioned(schema) {
134        return Ok(flat_only());
135    }
136    let partitions = tables.join(&schema.table);
137    if !partitions.is_dir() {
138        // NOTE: a partitioned schema whose records were never split yet still keeps one flat log.
139        return Ok(flat_only());
140    }
141    let logs = partition_logs_in(&partitions)?;
142    Ok(if logs.is_empty() { flat_only() } else { logs })
143}
144
145struct UniqueViolation {
146    table: String,
147    field: String,
148    keys: Vec<String>,
149}
150
151fn describe_unique_violation(violation: &UniqueViolation) -> String {
152    format!(
153        "field \"{}\" of table \"{}\" is declared unique, and {} record(s) already share a value: {}",
154        violation.field,
155        violation.table,
156        violation.keys.len(),
157        violation.keys.join(", ")
158    )
159}
160
161/// Scans every existing (format-1) log of the table for values that clash on a field declared
162/// `unique` in the candidate schema. Mirrors `uniqueViolationsOfTable` in
163/// `packages/hayson/src/catalog/migrate.ts`, but reads live data through the same `TableRegistry`
164/// the rest of the engine opens a table with, rather than the TS engine's own paged `read(keys)`
165/// callback.
166///
167/// Answers "no violations" only for a table whose logs it did read — a layout it cannot resolve
168/// the logs of is an error, since a silent empty answer would let the migration declare a field
169/// `unique` over data that already breaks it.
170fn unique_violations_of_table(
171    dir: &Path,
172    layout: CatalogLayout,
173    schema: &TableSchema,
174) -> Result<Vec<UniqueViolation>, HaysonSchemaError> {
175    let unique_fields: Vec<String> = schema
176        .fields
177        .iter()
178        .filter(|(name, spec)| spec.unique && name.as_str() != schema.primary_key)
179        .map(|(name, _)| name.clone())
180        .collect();
181    if unique_fields.is_empty() {
182        return Ok(Vec::new());
183    }
184
185    let logs = log_paths_of(dir, layout, schema)?;
186    if logs.is_empty() {
187        return Ok(Vec::new());
188    }
189    let mut registry = TableRegistry::new(schema.clone());
190    for (part, path) in &logs {
191        let reading = |error: std::io::Error| {
192            HaysonSchemaError(format!(
193                "reading \"{}\" to check unique fields before migrating it: {error}",
194                path.display()
195            ))
196        };
197        // NOTE: readonly open on purpose — this check runs before anything on disk may change.
198        let log = LogFile::open(path, true).map_err(reading)?;
199        let scan = log.scan(0).map_err(reading)?;
200        registry.apply_scan_in(scan, part);
201    }
202
203    let mut violations = Vec::new();
204    for field in unique_fields {
205        let mut clashing: HashSet<String> = HashSet::new();
206        for key in registry.keys() {
207            let Some(projection) = registry.projection_of(key) else {
208                continue;
209            };
210            let Some(value) = projection.get(&field) else { continue };
211            if let Some(holders) = registry.lookup_exact(&field, value) {
212                if holders.len() > 1 {
213                    clashing.extend(holders.iter().cloned());
214                }
215            }
216        }
217        if !clashing.is_empty() {
218            let mut keys: Vec<String> = clashing.into_iter().collect();
219            keys.sort();
220            violations.push(UniqueViolation {
221                table: schema.table.clone(),
222                field,
223                keys,
224            });
225        }
226    }
227    Ok(violations)
228}
229
230/// Reads every `*.schema.json` of a format-1 database at `dir` and rewrites it into one format-2
231/// `Catalog`, ready for `write_catalog` — the migration itself is left to the caller, which also
232/// decides when `createdAt`/`updatedAt` are pinned. Refuses while `.hayson.txn` sits on disk
233/// (`assertJournalGone` in `packages/hayson/src/catalog/migrate.ts`): an unfinished unit of work is
234/// redone through the layout it was written in, so the format cannot change out from under it.
235/// Also refuses — without writing anything — when a field declared `unique` already has two live
236/// records sharing a value (`uniqueViolationsOf` in migrate.ts).
237pub fn migrate_format_1_to_2(dir: &Path) -> Result<Catalog, HaysonSchemaError> {
238    assert_journal_gone(dir)?;
239
240    let now = chrono::Utc::now().timestamp_millis() as u64;
241    let layout = layout_of(dir);
242    let mut tables = BTreeMap::new();
243    let mut schemas = Vec::new();
244    for path in schema_files_of(dir)? {
245        let contents = fs::read_to_string(&path)
246            .map_err(|error| HaysonSchemaError(format!("reading \"{}\" to migrate it: {error}", path.display())))?;
247        let document: Value = serde_json::from_str(&contents)
248            .map_err(|error| HaysonSchemaError(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
249        let schema = parse_schema(&with_current_schema_marker(document))?;
250        let entry = table_entry_of(&schema, None, now);
251        tables.insert(schema.table.clone(), entry);
252        schemas.push(schema);
253    }
254
255    let mut violations = Vec::new();
256    for schema in &schemas {
257        violations.extend(unique_violations_of_table(dir, layout, schema)?);
258    }
259    if !violations.is_empty() {
260        let detail = violations
261            .iter()
262            .map(describe_unique_violation)
263            .collect::<Vec<_>>()
264            .join("; ");
265        return Err(HaysonSchemaError(format!(
266            "refusing to migrate \"{}\" to the metadata catalog: {detail} — nothing on disk was changed; make those keys unique (or take the constraint off the schema) and migrate again",
267            dir.display()
268        )));
269    }
270
271    Ok(Catalog {
272        catalog_version: CATALOG_VERSION,
273        format_version: CATALOG_FORMAT_VERSION,
274        created_at: now,
275        updated_at: now,
276        engine: None,
277        layout: Some(layout),
278        defaults: None,
279        tables,
280        relations: None,
281        views: None,
282        triggers: None,
283    })
284}
285
286const CATALOG_FILE: &str = "hayson.json";
287
288/// Outcome of [`migrate_format1`] on a directory. `AlreadyMigrated` is not an error — the
289/// operation is safe to put in a script that runs on every deployment.
290#[derive(Debug)]
291pub enum MigrateOutcome {
292    AlreadyMigrated {
293        catalog: PathBuf,
294    },
295    Migrated {
296        catalog: PathBuf,
297        tables: Vec<String>,
298        removed_schema_files: Vec<PathBuf>,
299    },
300}
301
302#[derive(Debug)]
303pub enum MigrateFormat1Error {
304    NotADirectory(PathBuf),
305    UnreadableMarker(PathBuf, String),
306    UnknownMarker(PathBuf, String),
307    NoSchemaFiles(PathBuf),
308    Schema(HaysonSchemaError),
309    Io(std::io::Error),
310    Write(String),
311}
312
313impl fmt::Display for MigrateFormat1Error {
314    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
315        match self {
316            MigrateFormat1Error::NotADirectory(path) => {
317                write!(f, "\"{}\" is not a directory — name the database directory itself", path.display())
318            }
319            MigrateFormat1Error::UnreadableMarker(path, detail) => {
320                write!(f, "\"{}\" cannot be read as JSON: {detail} — nothing on disk was changed", path.display())
321            }
322            MigrateFormat1Error::UnknownMarker(path, marker) => write!(
323                f,
324                "\"{}\" carries the marker {marker}, which is neither 1 (the manifest this rewrites) nor 2 (the metadata catalog it writes) — nothing on disk was changed",
325                path.display()
326            ),
327            MigrateFormat1Error::NoSchemaFiles(path) => write!(
328                f,
329                "\"{}\" holds no <table>.schema.json, so there is no format-1 database here to rewrite — nothing on disk was changed",
330                path.display()
331            ),
332            MigrateFormat1Error::Schema(error) => error.fmt(f),
333            MigrateFormat1Error::Io(error) => error.fmt(f),
334            MigrateFormat1Error::Write(detail) => f.write_str(detail),
335        }
336    }
337}
338
339impl std::error::Error for MigrateFormat1Error {}
340
341impl From<HaysonSchemaError> for MigrateFormat1Error {
342    fn from(value: HaysonSchemaError) -> Self {
343        MigrateFormat1Error::Schema(value)
344    }
345}
346
347impl From<std::io::Error> for MigrateFormat1Error {
348    fn from(value: std::io::Error) -> Self {
349        MigrateFormat1Error::Io(value)
350    }
351}
352
353fn marker_of(path: &Path) -> Result<Option<u64>, MigrateFormat1Error> {
354    let raw = match fs::read_to_string(path) {
355        Ok(raw) => raw,
356        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
357        Err(error) => return Err(error.into()),
358    };
359    let document: Value = serde_json::from_str(&raw)
360        .map_err(|error| MigrateFormat1Error::UnreadableMarker(path.to_path_buf(), error.to_string()))?;
361    // NOTE: "hayson" is the pre-5.0.0 name of "catalogVersion" — a format-1 manifest only has that one.
362    let marker = document.get("catalogVersion").or_else(|| document.get("hayson"));
363    match marker.and_then(Value::as_u64) {
364        Some(marker) => Ok(Some(marker)),
365        None => Err(MigrateFormat1Error::UnknownMarker(
366            path.to_path_buf(),
367            marker.map_or_else(|| "nothing".to_string(), ToString::to_string),
368        )),
369    }
370}
371
372/// Whether `migrate_format1(dir)` would find something to rewrite right now — the same guard
373/// clauses it runs before touching disk, without the write. `false` on a directory already at
374/// format 2, or on one with neither a catalog nor any `<table>.schema.json` (a fresh, empty
375/// database directory): a caller deciding whether to call `migrate_format1` "just in case" on
376/// every open does not have to reinvent this check, which duplicates `schema_files_of`'s own
377/// notion of what counts as format 1.
378pub fn is_format_one_directory(dir: &Path) -> Result<bool, MigrateFormat1Error> {
379    if !dir.is_dir() {
380        return Err(MigrateFormat1Error::NotADirectory(dir.to_path_buf()));
381    }
382    let catalog_path = dir.join(CATALOG_FILE);
383    match marker_of(&catalog_path)? {
384        Some(1) | None => Ok(!schema_files_of(dir)?.is_empty()),
385        Some(2) => Ok(false),
386        Some(other) => Err(MigrateFormat1Error::UnknownMarker(catalog_path, other.to_string())),
387    }
388}
389
390/// Rewrites the format-1 database at `dir` — a `<table>.schema.json` per table, next to the logs —
391/// into the format-2 metadata catalog the engine reads, in place and offline. The catalog lands
392/// first, atomically, and only then do the schema files go: interrupted in between, the directory
393/// is a working format-2 database with residue, which is the state `Database::open` already reads.
394///
395/// Nothing here opens the database, on purpose — the engine refuses format 1, and this is the one
396/// way out of it. Take a copy of the directory first; the rewrite is one-way. Shared by
397/// `hayson-cli`'s `migrate-format1` command and both bindings' `migrateFormat1`.
398pub fn migrate_format1(dir: &Path) -> Result<MigrateOutcome, MigrateFormat1Error> {
399    if !dir.is_dir() {
400        return Err(MigrateFormat1Error::NotADirectory(dir.to_path_buf()));
401    }
402    let catalog_path = dir.join(CATALOG_FILE);
403
404    match marker_of(&catalog_path)? {
405        Some(1) | None => {}
406        Some(2) => return Ok(MigrateOutcome::AlreadyMigrated { catalog: catalog_path }),
407        Some(other) => return Err(MigrateFormat1Error::UnknownMarker(catalog_path, other.to_string())),
408    }
409
410    let schema_files = schema_files_of(dir)?;
411    if schema_files.is_empty() {
412        return Err(MigrateFormat1Error::NoSchemaFiles(dir.to_path_buf()));
413    }
414
415    let catalog = migrate_format_1_to_2(dir)?;
416    let tables = catalog.tables.keys().cloned().collect();
417    super::write::write_catalog(&catalog_path, &catalog)
418        .map_err(|error| MigrateFormat1Error::Write(error.to_string()))?;
419
420    let mut removed_schema_files = Vec::new();
421    for path in schema_files {
422        fs::remove_file(&path)?;
423        removed_schema_files.push(path);
424    }
425
426    Ok(MigrateOutcome::Migrated {
427        catalog: catalog_path,
428        tables,
429        removed_schema_files,
430    })
431}
432
433#[cfg(test)]
434mod tests {
435    use serde_json::json;
436    use tempfile::tempdir;
437
438    use super::*;
439    use crate::catalog::parse::parse_catalog;
440    use crate::catalog::serialize::catalog_document;
441
442    /// A format-1 `<table>.schema.json`, marker key and all: what sits on the disk being migrated.
443    fn schema_document(table: &str, fields: Value) -> Value {
444        let mut document = json!({
445            "hayson": 1,
446            "table": table,
447            "primaryKey": "id",
448            "fields": { "id": { "type": "string" } },
449        });
450        for (key, value) in fields.as_object().unwrap() {
451            document["fields"][key] = value.clone();
452        }
453        document
454    }
455
456    fn write_schema_document(dir: &Path, table: &str, document: &Value) {
457        fs::create_dir_all(dir).unwrap();
458        fs::write(
459            dir.join(format!("{table}.schema.json")),
460            serde_json::to_string(document).unwrap(),
461        )
462        .unwrap();
463    }
464
465    fn write_schema_file(dir: &Path, table: &str, fields: Value) {
466        write_schema_document(dir, table, &schema_document(table, fields));
467    }
468
469    fn write_log_at(path: &Path, records: &[(&str, Value)]) {
470        use crate::schema::HaysonRecord;
471        use crate::storage::{serialize_entry, LogMeta, LogOp};
472
473        let mut lines = Vec::new();
474        for (position, (key, fields)) in records.iter().enumerate() {
475            let record: HaysonRecord = fields
476                .as_object()
477                .unwrap()
478                .iter()
479                .map(|(k, v)| (k.clone(), v.clone()))
480                .collect();
481            let meta = LogMeta {
482                key: (*key).to_string(),
483                op: LogOp::Put,
484                seq: position as u64 + 1,
485                ts: 1000 + position as i64,
486            };
487            lines.push(serialize_entry(&meta, Some(&record)));
488        }
489        fs::create_dir_all(path.parent().unwrap()).unwrap();
490        fs::write(path, format!("{}\n", lines.join("\n"))).unwrap();
491    }
492
493    fn write_log_file(dir: &Path, table: &str, records: &[(&str, Value)]) {
494        write_log_at(&dir.join(format!("{table}{LOG_SUFFIX}")), records);
495    }
496
497    #[test]
498    fn migrates_every_schema_file_of_the_directory_into_one_catalog() {
499        let dir = tempdir().unwrap();
500        write_schema_file(dir.path(), "users", json!({ "name": { "type": "string" } }));
501        write_schema_file(dir.path(), "orders", json!({ "userId": { "type": "string" } }));
502
503        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
504
505        assert_eq!(catalog.tables.len(), 2);
506        assert!(catalog.tables.contains_key("users"));
507        assert!(catalog.tables.contains_key("orders"));
508    }
509
510    #[test]
511    fn round_trips_through_serialize_and_parse() {
512        let dir = tempdir().unwrap();
513        write_schema_file(dir.path(), "users", json!({ "name": { "type": "string" } }));
514        write_schema_file(dir.path(), "orders", json!({ "userId": { "type": "string" } }));
515
516        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
517        let reparsed = parse_catalog(&catalog_document(&catalog)).unwrap();
518
519        assert_eq!(reparsed, catalog);
520    }
521
522    #[test]
523    fn refuses_while_a_journal_is_on_disk() {
524        let dir = tempdir().unwrap();
525        write_schema_file(dir.path(), "users", json!({}));
526        fs::write(dir.path().join(".hayson.txn"), "{}").unwrap();
527
528        let error = migrate_format_1_to_2(dir.path()).unwrap_err();
529        assert!(error.0.contains(".hayson.txn"));
530    }
531
532    #[test]
533    fn migrates_when_no_table_declares_a_unique_field() {
534        let dir = tempdir().unwrap();
535        write_schema_file(dir.path(), "users", json!({ "name": { "type": "string" } }));
536        write_log_file(
537            dir.path(),
538            "users",
539            &[
540                ("user_1", json!({ "id": "user_1", "name": "Ania" })),
541                ("user_2", json!({ "id": "user_2", "name": "Ania" })),
542            ],
543        );
544
545        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
546        assert!(catalog.tables.contains_key("users"));
547    }
548
549    #[test]
550    fn migrates_a_unique_field_that_has_no_live_duplicate() {
551        let dir = tempdir().unwrap();
552        write_schema_file(
553            dir.path(),
554            "users",
555            json!({ "email": { "type": "string", "unique": true } }),
556        );
557        write_log_file(
558            dir.path(),
559            "users",
560            &[
561                ("user_1", json!({ "id": "user_1", "email": "ania@example.test" })),
562                ("user_2", json!({ "id": "user_2", "email": "piotr@example.test" })),
563            ],
564        );
565
566        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
567        assert!(catalog.tables.contains_key("users"));
568    }
569
570    #[test]
571    fn refuses_the_whole_migration_when_one_table_has_a_live_unique_duplicate() {
572        let dir = tempdir().unwrap();
573        write_schema_file(
574            dir.path(),
575            "users",
576            json!({ "email": { "type": "string", "unique": true } }),
577        );
578        write_log_file(
579            dir.path(),
580            "users",
581            &[
582                ("user_1", json!({ "id": "user_1", "email": "ania@example.test" })),
583                ("user_2", json!({ "id": "user_2", "email": "ania@example.test" })),
584            ],
585        );
586        write_schema_file(dir.path(), "orders", json!({ "userId": { "type": "string" } }));
587        write_log_file(
588            dir.path(),
589            "orders",
590            &[("order_1", json!({ "id": "order_1", "userId": "user_1" }))],
591        );
592
593        let error = migrate_format_1_to_2(dir.path()).unwrap_err();
594        assert!(error.0.contains("\"email\""));
595        assert!(error.0.contains("\"users\""));
596        assert!(error.0.contains("nothing on disk was changed"));
597    }
598    #[test]
599    fn refuses_a_partitioned_table_whose_partitions_together_hold_a_unique_duplicate() {
600        let dir = tempdir().unwrap();
601        let mut document = schema_document(
602            "users",
603            json!({
604                "kind": { "type": "string", "required": true },
605                "email": { "type": "string", "unique": true },
606            }),
607        );
608        document["partitionBy"] = json!("kind");
609        write_schema_document(dir.path(), "users", &document);
610        write_log_at(
611            &dir.path().join("users").join(format!("staff{LOG_SUFFIX}")),
612            &[(
613                "user_1",
614                json!({ "id": "user_1", "kind": "staff", "email": "ania@example.test" }),
615            )],
616        );
617        write_log_at(
618            &dir.path().join("users").join(format!("guest{LOG_SUFFIX}")),
619            &[(
620                "user_2",
621                json!({ "id": "user_2", "kind": "guest", "email": "ania@example.test" }),
622            )],
623        );
624
625        let error = migrate_format_1_to_2(dir.path()).unwrap_err();
626
627        assert!(error.0.contains("\"email\""), "{}", error.0);
628        assert!(error.0.contains("user_1") && error.0.contains("user_2"), "{}", error.0);
629        assert!(error.0.contains("nothing on disk was changed"), "{}", error.0);
630    }
631
632    #[test]
633    fn migrates_a_partitioned_table_whose_partitions_hold_no_duplicate() {
634        let dir = tempdir().unwrap();
635        let mut document = schema_document(
636            "users",
637            json!({
638                "kind": { "type": "string", "required": true },
639                "email": { "type": "string", "unique": true },
640            }),
641        );
642        document["partitionBy"] = json!("kind");
643        write_schema_document(dir.path(), "users", &document);
644        write_log_at(
645            &dir.path().join("users").join(format!("staff{LOG_SUFFIX}")),
646            &[(
647                "user_1",
648                json!({ "id": "user_1", "kind": "staff", "email": "ania@example.test" }),
649            )],
650        );
651        write_log_at(
652            &dir.path().join("users").join(format!("guest{LOG_SUFFIX}")),
653            &[(
654                "user_2",
655                json!({ "id": "user_2", "kind": "guest", "email": "piotr@example.test" }),
656            )],
657        );
658
659        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
660
661        assert!(catalog.tables.contains_key("users"));
662    }
663
664    #[test]
665    fn refuses_a_nested_layout_table_whose_log_holds_a_unique_duplicate() {
666        let dir = tempdir().unwrap();
667        fs::write(
668            dir.path().join("hayson.json"),
669            json!({ "layout": "nested" }).to_string(),
670        )
671        .unwrap();
672        write_schema_file(
673            &dir.path().join("schema"),
674            "users",
675            json!({ "email": { "type": "string", "unique": true } }),
676        );
677        write_log_file(
678            &dir.path().join("tables"),
679            "users",
680            &[
681                ("user_1", json!({ "id": "user_1", "email": "ania@example.test" })),
682                ("user_2", json!({ "id": "user_2", "email": "ania@example.test" })),
683            ],
684        );
685
686        let error = migrate_format_1_to_2(dir.path()).unwrap_err();
687
688        assert!(error.0.contains("\"email\""), "{}", error.0);
689        assert!(error.0.contains("user_1") && error.0.contains("user_2"), "{}", error.0);
690        assert!(error.0.contains("nothing on disk was changed"), "{}", error.0);
691    }
692
693    #[test]
694    fn migrates_a_nested_layout_table_whose_log_holds_no_duplicate() {
695        let dir = tempdir().unwrap();
696        fs::write(
697            dir.path().join("hayson.json"),
698            json!({ "layout": "nested" }).to_string(),
699        )
700        .unwrap();
701        write_schema_file(
702            &dir.path().join("schema"),
703            "users",
704            json!({ "email": { "type": "string", "unique": true } }),
705        );
706        write_log_file(
707            &dir.path().join("tables"),
708            "users",
709            &[
710                ("user_1", json!({ "id": "user_1", "email": "ania@example.test" })),
711                ("user_2", json!({ "id": "user_2", "email": "piotr@example.test" })),
712            ],
713        );
714
715        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
716
717        assert_eq!(catalog.layout, Some(CatalogLayout::Nested));
718        assert!(catalog.tables.contains_key("users"));
719    }
720
721    fn write_format_1(dir: &Path) {
722        fs::write(
723            dir.join(CATALOG_FILE),
724            "{\n  \"hayson\": 1,\n  \"tables\": [\"users\"]\n}\n",
725        )
726        .unwrap();
727        fs::write(
728            dir.join("users.schema.json"),
729            serde_json::to_string(&json!({
730                "hayson": 1,
731                "table": "users",
732                "primaryKey": "id",
733                "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
734            }))
735            .unwrap(),
736        )
737        .unwrap();
738        fs::write(
739            dir.join("users.jsonl"),
740            "{\"_k\":\"user_1\",\"_op\":\"put\",\"_seq\":1,\"_ts\":1000,\"id\":\"user_1\",\"name\":\"Ania\"}\n",
741        )
742        .unwrap();
743    }
744
745    #[test]
746    fn migrate_format1_rewrites_a_format_1_directory_into_a_catalog_the_engine_opens() {
747        use crate::database::Database;
748
749        let dir = tempdir().unwrap();
750        write_format_1(dir.path());
751        assert!(Database::open(dir.path()).is_err());
752
753        let outcome = migrate_format1(dir.path()).unwrap();
754
755        let MigrateOutcome::Migrated {
756            tables,
757            removed_schema_files,
758            ..
759        } = outcome
760        else {
761            panic!("a format-1 directory has to report a migration, not a no-op");
762        };
763        assert_eq!(tables, vec!["users".to_string()]);
764        assert_eq!(removed_schema_files, vec![dir.path().join("users.schema.json")]);
765        assert!(!dir.path().join("users.schema.json").exists());
766
767        let db = Database::open(dir.path()).unwrap();
768        assert_eq!(db.table_names(), vec!["users"]);
769        assert_eq!(
770            db.table("users").unwrap().get("user_1").unwrap().unwrap().get("name"),
771            Some(&Value::from("Ania"))
772        );
773    }
774
775    #[test]
776    fn migrate_format1_leaves_the_log_byte_identical() {
777        let dir = tempdir().unwrap();
778        write_format_1(dir.path());
779        let before = fs::read(dir.path().join("users.jsonl")).unwrap();
780
781        migrate_format1(dir.path()).unwrap();
782
783        assert_eq!(fs::read(dir.path().join("users.jsonl")).unwrap(), before);
784    }
785
786    #[test]
787    fn migrate_format1_a_second_run_reports_a_no_op_and_rewrites_nothing() {
788        let dir = tempdir().unwrap();
789        write_format_1(dir.path());
790        migrate_format1(dir.path()).unwrap();
791        let bytes = fs::read_to_string(dir.path().join(CATALOG_FILE)).unwrap();
792
793        let outcome = migrate_format1(dir.path()).unwrap();
794
795        assert!(matches!(outcome, MigrateOutcome::AlreadyMigrated { .. }));
796        assert_eq!(fs::read_to_string(dir.path().join(CATALOG_FILE)).unwrap(), bytes);
797    }
798
799    #[test]
800    fn is_format_one_directory_is_false_on_an_empty_directory() {
801        let dir = tempdir().unwrap();
802
803        assert!(!is_format_one_directory(dir.path()).unwrap());
804    }
805
806    #[test]
807    fn is_format_one_directory_is_true_where_migrate_format1_would_find_something_to_rewrite() {
808        let dir = tempdir().unwrap();
809        write_format_1(dir.path());
810
811        assert!(is_format_one_directory(dir.path()).unwrap());
812    }
813
814    #[test]
815    fn is_format_one_directory_is_false_on_a_directory_already_at_format_2() {
816        let dir = tempdir().unwrap();
817        write_format_1(dir.path());
818        migrate_format1(dir.path()).unwrap();
819
820        assert!(!is_format_one_directory(dir.path()).unwrap());
821    }
822
823    #[test]
824    fn is_format_one_directory_refuses_an_unknown_marker_the_same_way_migrate_format1_does() {
825        let dir = tempdir().unwrap();
826        write_format_1(dir.path());
827        fs::write(dir.path().join(CATALOG_FILE), "{ \"hayson\": 3 }\n").unwrap();
828
829        let failure = is_format_one_directory(dir.path()).unwrap_err();
830
831        assert!(matches!(failure, MigrateFormat1Error::UnknownMarker(_, _)));
832    }
833
834    #[test]
835    fn is_format_one_directory_refuses_a_path_that_is_not_a_directory() {
836        let dir = tempdir().unwrap();
837        let file = dir.path().join("not-a-dir");
838        fs::write(&file, "").unwrap();
839
840        let failure = is_format_one_directory(&file).unwrap_err();
841
842        assert!(matches!(failure, MigrateFormat1Error::NotADirectory(_)));
843    }
844
845    #[test]
846    fn migrate_format1_refuses_a_directory_holding_no_schema_files_without_writing_a_catalog() {
847        let dir = tempdir().unwrap();
848
849        let failure = migrate_format1(dir.path()).unwrap_err();
850
851        assert!(matches!(failure, MigrateFormat1Error::NoSchemaFiles(_)));
852        assert!(!dir.path().join(CATALOG_FILE).exists());
853    }
854
855    #[test]
856    fn migrate_format1_refuses_a_marker_that_is_neither_1_nor_2_and_keeps_the_schema_files() {
857        let dir = tempdir().unwrap();
858        write_format_1(dir.path());
859        fs::write(dir.path().join(CATALOG_FILE), "{ \"hayson\": 3 }\n").unwrap();
860
861        let failure = migrate_format1(dir.path()).unwrap_err();
862
863        assert!(matches!(failure, MigrateFormat1Error::UnknownMarker(_, _)));
864        assert!(dir.path().join("users.schema.json").exists());
865    }
866
867    #[test]
868    fn migrate_format1_refuses_while_an_unfinished_unit_of_work_is_on_disk() {
869        let dir = tempdir().unwrap();
870        write_format_1(dir.path());
871        fs::write(dir.path().join(".hayson.txn"), "{}\n").unwrap();
872
873        let failure = migrate_format1(dir.path()).unwrap_err();
874
875        assert!(matches!(failure, MigrateFormat1Error::Schema(_)));
876        assert!(failure.to_string().contains(".hayson.txn"));
877        assert!(dir.path().join("users.schema.json").exists());
878        assert_eq!(marker_of(&dir.path().join(CATALOG_FILE)).unwrap(), Some(1));
879    }
880
881    #[test]
882    fn migrate_format1_refuses_a_schema_file_naming_a_table_outside_the_directory_before_writing_anything() {
883        let dir = tempdir().unwrap();
884        write_format_1(dir.path());
885        let schema: Value =
886            serde_json::from_str(&fs::read_to_string(dir.path().join("users.schema.json")).unwrap()).unwrap();
887        let mut traversing = schema.clone();
888        traversing["table"] = Value::from("../outside");
889        fs::write(
890            dir.path().join("outside.schema.json"),
891            serde_json::to_string(&traversing).unwrap(),
892        )
893        .unwrap();
894
895        let failure = migrate_format1(dir.path()).unwrap_err();
896
897        assert!(failure.to_string().contains("unusable table name"));
898        assert_eq!(marker_of(&dir.path().join(CATALOG_FILE)).unwrap(), Some(1));
899        assert!(dir.path().join("users.schema.json").exists());
900    }
901
902    #[cfg(unix)]
903    #[test]
904    fn migrate_format1_writes_the_catalog_no_looser_than_the_manifest_it_replaces() {
905        use std::os::unix::fs::PermissionsExt;
906
907        let dir = tempdir().unwrap();
908        write_format_1(dir.path());
909        let catalog_path = dir.path().join(CATALOG_FILE);
910        fs::set_permissions(&catalog_path, fs::Permissions::from_mode(0o600)).unwrap();
911
912        migrate_format1(dir.path()).unwrap();
913
914        assert_eq!(fs::metadata(&catalog_path).unwrap().permissions().mode() & 0o777, 0o600);
915    }
916
917    #[test]
918    fn migrate_format1_refuses_a_path_that_is_not_a_directory() {
919        let dir = tempdir().unwrap();
920        let file = dir.path().join("hayson.json");
921        fs::write(&file, "{}").unwrap();
922
923        assert!(matches!(
924            migrate_format1(&file).unwrap_err(),
925            MigrateFormat1Error::NotADirectory(_)
926        ));
927    }
928}
hayson_engine/catalog/migrate.rs

hayson_engine/catalog/
migrate.rs
