hayson-engine 0.10.0

use std::collections::{BTreeMap, HashSet};
use std::fmt;
use std::fs;
use std::path::{Path, PathBuf};

use serde_json::Value;

use crate::errors::HaysonSchemaError;
use crate::registry::table_registry::TableRegistry;
use crate::schema::parse::with_current_schema_marker;
use crate::schema::{is_partitioned, parse_schema, TableSchema, UNPARTITIONED};
use crate::storage::{LogFile, LOG_SUFFIX};

use super::build::table_entry_of;
use super::types::{Catalog, CatalogLayout, CATALOG_FORMAT_VERSION, CATALOG_VERSION};

const JOURNAL_FILE_NAME: &str = ".hayson.txn";
const SCHEMA_FILE_SUFFIX: &str = ".schema.json";

fn journal_paths_of(dir: &Path) -> Vec<PathBuf> {
    vec![dir.join(JOURNAL_FILE_NAME), dir.join("runtime").join(JOURNAL_FILE_NAME)]
}

fn assert_journal_gone(dir: &Path) -> Result<(), HaysonSchemaError> {
    for path in journal_paths_of(dir) {
        if path.exists() {
            return Err(HaysonSchemaError(format!(
                "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",
                path.display()
            )));
        }
    }
    Ok(())
}

fn layout_of(dir: &Path) -> CatalogLayout {
    let manifest = dir.join("hayson.json");
    let Ok(contents) = fs::read_to_string(&manifest) else {
        return CatalogLayout::Flat;
    };
    let Ok(value) = serde_json::from_str::<Value>(&contents) else {
        return CatalogLayout::Flat;
    };
    match value.get("layout").and_then(Value::as_str) {
        Some("nested") => CatalogLayout::Nested,
        _ => CatalogLayout::Flat,
    }
}

fn schema_dirs_of(dir: &Path) -> Vec<PathBuf> {
    let nested = dir.join("schema");
    if nested.is_dir() {
        vec![nested]
    } else {
        vec![dir.to_path_buf()]
    }
}

/// Every `<table>.schema.json` a format-1 database at `dir` declares a table through, sorted.
/// [`migrate_format_1_to_2`] reads exactly these, so the caller that persists the catalog it
/// returns can retire exactly these once the catalog is safely on disk.
pub fn schema_files_of(dir: &Path) -> Result<Vec<PathBuf>, HaysonSchemaError> {
    let mut files = Vec::new();
    for schema_dir in schema_dirs_of(dir) {
        let entries = fs::read_dir(&schema_dir).map_err(|error| {
            HaysonSchemaError(format!("reading \"{}\" to migrate it: {error}", schema_dir.display()))
        })?;
        for entry in entries {
            let entry = entry.map_err(|error| {
                HaysonSchemaError(format!("reading \"{}\" to migrate it: {error}", schema_dir.display()))
            })?;
            let path = entry.path();
            let is_schema_file = path
                .file_name()
                .and_then(|name| name.to_str())
                .is_some_and(|name| name.ends_with(SCHEMA_FILE_SUFFIX));
            if is_schema_file {
                files.push(path);
            }
        }
    }
    files.sort();
    Ok(files)
}

fn tables_dir_of(dir: &Path, layout: CatalogLayout) -> PathBuf {
    match layout {
        CatalogLayout::Nested => dir.join("tables"),
        CatalogLayout::Flat => dir.to_path_buf(),
    }
}

fn partition_logs_in(dir: &Path) -> Result<Vec<(String, PathBuf)>, HaysonSchemaError> {
    let reading = |error: std::io::Error| {
        HaysonSchemaError(format!(
            "reading \"{}\" to check unique fields before migrating it: {error}",
            dir.display()
        ))
    };
    // NOTE: read_dir here, not partition_names_in() — that one reports an unreadable dir as empty.
    let mut logs = Vec::new();
    for entry in fs::read_dir(dir).map_err(reading)? {
        let name = entry.map_err(reading)?.file_name();
        let Some(part) = name.to_str().and_then(|name| name.strip_suffix(LOG_SUFFIX)) else {
            continue;
        };
        if !part.is_empty() {
            logs.push((part.to_string(), dir.join(&name)));
        }
    }
    logs.sort();
    Ok(logs)
}

/// Every log file a format-1 table keeps its records in, by partition name, for each of the layout
/// shapes a format-1 database can be on disk in: flat or `nested` (`docs/format.md`, "Układ
/// katalogu"), partitioned or not. An empty answer means the table has no log at all — nothing was
/// ever written to it — and never "the log is somewhere this cannot look".
fn log_paths_of(
    dir: &Path,
    layout: CatalogLayout,
    schema: &TableSchema,
) -> Result<Vec<(String, PathBuf)>, HaysonSchemaError> {
    let tables = tables_dir_of(dir, layout);
    let flat = tables.join(format!("{}{LOG_SUFFIX}", schema.table));
    let flat_only = || {
        if flat.exists() {
            vec![(UNPARTITIONED.to_string(), flat.clone())]
        } else {
            Vec::new()
        }
    };
    if !is_partitioned(schema) {
        return Ok(flat_only());
    }
    let partitions = tables.join(&schema.table);
    if !partitions.is_dir() {
        // NOTE: a partitioned schema whose records were never split yet still keeps one flat log.
        return Ok(flat_only());
    }
    let logs = partition_logs_in(&partitions)?;
    Ok(if logs.is_empty() { flat_only() } else { logs })
}

struct UniqueViolation {
    table: String,
    field: String,
    keys: Vec<String>,
}

fn describe_unique_violation(violation: &UniqueViolation) -> String {
    format!(
        "field \"{}\" of table \"{}\" is declared unique, and {} record(s) already share a value: {}",
        violation.field,
        violation.table,
        violation.keys.len(),
        violation.keys.join(", ")
    )
}

/// Scans every existing (format-1) log of the table for values that clash on a field declared
/// `unique` in the candidate schema. Mirrors `uniqueViolationsOfTable` in
/// `packages/hayson/src/catalog/migrate.ts`, but reads live data through the same `TableRegistry`
/// the rest of the engine opens a table with, rather than the TS engine's own paged `read(keys)`
/// callback.
///
/// Answers "no violations" only for a table whose logs it did read — a layout it cannot resolve
/// the logs of is an error, since a silent empty answer would let the migration declare a field
/// `unique` over data that already breaks it.
fn unique_violations_of_table(
    dir: &Path,
    layout: CatalogLayout,
    schema: &TableSchema,
) -> Result<Vec<UniqueViolation>, HaysonSchemaError> {
    let unique_fields: Vec<String> = schema
        .fields
        .iter()
        .filter(|(name, spec)| spec.unique && name.as_str() != schema.primary_key)
        .map(|(name, _)| name.clone())
        .collect();
    if unique_fields.is_empty() {
        return Ok(Vec::new());
    }

    let logs = log_paths_of(dir, layout, schema)?;
    if logs.is_empty() {
        return Ok(Vec::new());
    }
    let mut registry = TableRegistry::new(schema.clone());
    for (part, path) in &logs {
        let reading = |error: std::io::Error| {
            HaysonSchemaError(format!(
                "reading \"{}\" to check unique fields before migrating it: {error}",
                path.display()
            ))
        };
        // NOTE: readonly open on purpose — this check runs before anything on disk may change.
        let log = LogFile::open(path, true).map_err(reading)?;
        let scan = log.scan(0).map_err(reading)?;
        registry.apply_scan_in(scan, part);
    }

    let mut violations = Vec::new();
    for field in unique_fields {
        let mut clashing: HashSet<String> = HashSet::new();
        for key in registry.keys() {
            let Some(projection) = registry.projection_of(key) else {
                continue;
            };
            let Some(value) = projection.get(&field) else { continue };
            if let Some(holders) = registry.lookup_exact(&field, value) {
                if holders.len() > 1 {
                    clashing.extend(holders.iter().cloned());
                }
            }
        }
        if !clashing.is_empty() {
            let mut keys: Vec<String> = clashing.into_iter().collect();
            keys.sort();
            violations.push(UniqueViolation {
                table: schema.table.clone(),
                field,
                keys,
            });
        }
    }
    Ok(violations)
}

/// Reads every `*.schema.json` of a format-1 database at `dir` and rewrites it into one format-2
/// `Catalog`, ready for `write_catalog` — the migration itself is left to the caller, which also
/// decides when `createdAt`/`updatedAt` are pinned. Refuses while `.hayson.txn` sits on disk
/// (`assertJournalGone` in `packages/hayson/src/catalog/migrate.ts`): an unfinished unit of work is
/// redone through the layout it was written in, so the format cannot change out from under it.
/// Also refuses — without writing anything — when a field declared `unique` already has two live
/// records sharing a value (`uniqueViolationsOf` in migrate.ts).
pub fn migrate_format_1_to_2(dir: &Path) -> Result<Catalog, HaysonSchemaError> {
    assert_journal_gone(dir)?;

    let now = chrono::Utc::now().timestamp_millis() as u64;
    let layout = layout_of(dir);
    let mut tables = BTreeMap::new();
    let mut schemas = Vec::new();
    for path in schema_files_of(dir)? {
        let contents = fs::read_to_string(&path)
            .map_err(|error| HaysonSchemaError(format!("reading \"{}\" to migrate it: {error}", path.display())))?;
        let document: Value = serde_json::from_str(&contents)
            .map_err(|error| HaysonSchemaError(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
        let schema = parse_schema(&with_current_schema_marker(document))?;
        let entry = table_entry_of(&schema, None, now);
        tables.insert(schema.table.clone(), entry);
        schemas.push(schema);
    }

    let mut violations = Vec::new();
    for schema in &schemas {
        violations.extend(unique_violations_of_table(dir, layout, schema)?);
    }
    if !violations.is_empty() {
        let detail = violations
            .iter()
            .map(describe_unique_violation)
            .collect::<Vec<_>>()
            .join("; ");
        return Err(HaysonSchemaError(format!(
            "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",
            dir.display()
        )));
    }

    Ok(Catalog {
        catalog_version: CATALOG_VERSION,
        format_version: CATALOG_FORMAT_VERSION,
        created_at: now,
        updated_at: now,
        engine: None,
        layout: Some(layout),
        defaults: None,
        tables,
        relations: None,
        views: None,
        triggers: None,
    })
}

const CATALOG_FILE: &str = "hayson.json";

/// Outcome of [`migrate_format1`] on a directory. `AlreadyMigrated` is not an error — the
/// operation is safe to put in a script that runs on every deployment.
#[derive(Debug)]
pub enum MigrateOutcome {
    AlreadyMigrated {
        catalog: PathBuf,
    },
    Migrated {
        catalog: PathBuf,
        tables: Vec<String>,
        removed_schema_files: Vec<PathBuf>,
    },
}

#[derive(Debug)]
pub enum MigrateFormat1Error {
    NotADirectory(PathBuf),
    UnreadableMarker(PathBuf, String),
    UnknownMarker(PathBuf, String),
    NoSchemaFiles(PathBuf),
    Schema(HaysonSchemaError),
    Io(std::io::Error),
    Write(String),
}

impl fmt::Display for MigrateFormat1Error {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            MigrateFormat1Error::NotADirectory(path) => {
                write!(f, "\"{}\" is not a directory — name the database directory itself", path.display())
            }
            MigrateFormat1Error::UnreadableMarker(path, detail) => {
                write!(f, "\"{}\" cannot be read as JSON: {detail} — nothing on disk was changed", path.display())
            }
            MigrateFormat1Error::UnknownMarker(path, marker) => write!(
                f,
                "\"{}\" carries the marker {marker}, which is neither 1 (the manifest this rewrites) nor 2 (the metadata catalog it writes) — nothing on disk was changed",
                path.display()
            ),
            MigrateFormat1Error::NoSchemaFiles(path) => write!(
                f,
                "\"{}\" holds no <table>.schema.json, so there is no format-1 database here to rewrite — nothing on disk was changed",
                path.display()
            ),
            MigrateFormat1Error::Schema(error) => error.fmt(f),
            MigrateFormat1Error::Io(error) => error.fmt(f),
            MigrateFormat1Error::Write(detail) => f.write_str(detail),
        }
    }
}

impl std::error::Error for MigrateFormat1Error {}

impl From<HaysonSchemaError> for MigrateFormat1Error {
    fn from(value: HaysonSchemaError) -> Self {
        MigrateFormat1Error::Schema(value)
    }
}

impl From<std::io::Error> for MigrateFormat1Error {
    fn from(value: std::io::Error) -> Self {
        MigrateFormat1Error::Io(value)
    }
}

fn marker_of(path: &Path) -> Result<Option<u64>, MigrateFormat1Error> {
    let raw = match fs::read_to_string(path) {
        Ok(raw) => raw,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error.into()),
    };
    let document: Value = serde_json::from_str(&raw)
        .map_err(|error| MigrateFormat1Error::UnreadableMarker(path.to_path_buf(), error.to_string()))?;
    // NOTE: "hayson" is the pre-5.0.0 name of "catalogVersion" — a format-1 manifest only has that one.
    let marker = document.get("catalogVersion").or_else(|| document.get("hayson"));
    match marker.and_then(Value::as_u64) {
        Some(marker) => Ok(Some(marker)),
        None => Err(MigrateFormat1Error::UnknownMarker(
            path.to_path_buf(),
            marker.map_or_else(|| "nothing".to_string(), ToString::to_string),
        )),
    }
}

/// Rewrites the format-1 database at `dir` — a `<table>.schema.json` per table, next to the logs —
/// into the format-2 metadata catalog the engine reads, in place and offline. The catalog lands
/// first, atomically, and only then do the schema files go: interrupted in between, the directory
/// is a working format-2 database with residue, which is the state `Database::open` already reads.
///
/// Nothing here opens the database, on purpose — the engine refuses format 1, and this is the one
/// way out of it. Take a copy of the directory first; the rewrite is one-way. Shared by
/// `hayson-cli`'s `migrate-format1` command and both bindings' `migrateFormat1`.
pub fn migrate_format1(dir: &Path) -> Result<MigrateOutcome, MigrateFormat1Error> {
    if !dir.is_dir() {
        return Err(MigrateFormat1Error::NotADirectory(dir.to_path_buf()));
    }
    let catalog_path = dir.join(CATALOG_FILE);

    match marker_of(&catalog_path)? {
        Some(1) | None => {}
        Some(2) => return Ok(MigrateOutcome::AlreadyMigrated { catalog: catalog_path }),
        Some(other) => return Err(MigrateFormat1Error::UnknownMarker(catalog_path, other.to_string())),
    }

    let schema_files = schema_files_of(dir)?;
    if schema_files.is_empty() {
        return Err(MigrateFormat1Error::NoSchemaFiles(dir.to_path_buf()));
    }

    let catalog = migrate_format_1_to_2(dir)?;
    let tables = catalog.tables.keys().cloned().collect();
    super::write::write_catalog(&catalog_path, &catalog)
        .map_err(|error| MigrateFormat1Error::Write(error.to_string()))?;

    let mut removed_schema_files = Vec::new();
    for path in schema_files {
        fs::remove_file(&path)?;
        removed_schema_files.push(path);
    }

    Ok(MigrateOutcome::Migrated {
        catalog: catalog_path,
        tables,
        removed_schema_files,
    })
}

#[cfg(test)]
mod tests {
    use serde_json::json;
    use tempfile::tempdir;

    use super::*;
    use crate::catalog::parse::parse_catalog;
    use crate::catalog::serialize::catalog_document;

    /// A format-1 `<table>.schema.json`, marker key and all: what sits on the disk being migrated.
    fn schema_document(table: &str, fields: Value) -> Value {
        let mut document = json!({
            "hayson": 1,
            "table": table,
            "primaryKey": "id",
            "fields": { "id": { "type": "string" } },
        });
        for (key, value) in fields.as_object().unwrap() {
            document["fields"][key] = value.clone();
        }
        document
    }

    fn write_schema_document(dir: &Path, table: &str, document: &Value) {
        fs::create_dir_all(dir).unwrap();
        fs::write(
            dir.join(format!("{table}.schema.json")),
            serde_json::to_string(document).unwrap(),
        )
        .unwrap();
    }

    fn write_schema_file(dir: &Path, table: &str, fields: Value) {
        write_schema_document(dir, table, &schema_document(table, fields));
    }

    fn write_log_at(path: &Path, records: &[(&str, Value)]) {
        use crate::schema::HaysonRecord;
        use crate::storage::{serialize_entry, LogMeta, LogOp};

        let mut lines = Vec::new();
        for (position, (key, fields)) in records.iter().enumerate() {
            let record: HaysonRecord = fields
                .as_object()
                .unwrap()
                .iter()
                .map(|(k, v)| (k.clone(), v.clone()))
                .collect();
            let meta = LogMeta {
                key: (*key).to_string(),
                op: LogOp::Put,
                seq: position as u64 + 1,
                ts: 1000 + position as i64,
            };
            lines.push(serialize_entry(&meta, Some(&record)));
        }
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(path, format!("{}\n", lines.join("\n"))).unwrap();
    }

    fn write_log_file(dir: &Path, table: &str, records: &[(&str, Value)]) {
        write_log_at(&dir.join(format!("{table}{LOG_SUFFIX}")), records);
    }

    #[test]
    fn migrates_every_schema_file_of_the_directory_into_one_catalog() {
        let dir = tempdir().unwrap();
        write_schema_file(dir.path(), "users", json!({ "name": { "type": "string" } }));
        write_schema_file(dir.path(), "orders", json!({ "userId": { "type": "string" } }));

        let catalog = migrate_format_1_to_2(dir.path()).unwrap();

        assert_eq!(catalog.tables.len(), 2);
        assert!(catalog.tables.contains_key("users"));
        assert!(catalog.tables.contains_key("orders"));
    }

    #[test]
    fn round_trips_through_serialize_and_parse() {
        let dir = tempdir().unwrap();
        write_schema_file(dir.path(), "users", json!({ "name": { "type": "string" } }));
        write_schema_file(dir.path(), "orders", json!({ "userId": { "type": "string" } }));

        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
        let reparsed = parse_catalog(&catalog_document(&catalog)).unwrap();

        assert_eq!(reparsed, catalog);
    }

    #[test]
    fn refuses_while_a_journal_is_on_disk() {
        let dir = tempdir().unwrap();
        write_schema_file(dir.path(), "users", json!({}));
        fs::write(dir.path().join(".hayson.txn"), "{}").unwrap();

        let error = migrate_format_1_to_2(dir.path()).unwrap_err();
        assert!(error.0.contains(".hayson.txn"));
    }

    #[test]
    fn migrates_when_no_table_declares_a_unique_field() {
        let dir = tempdir().unwrap();
        write_schema_file(dir.path(), "users", json!({ "name": { "type": "string" } }));
        write_log_file(
            dir.path(),
            "users",
            &[
                ("user_1", json!({ "id": "user_1", "name": "Ania" })),
                ("user_2", json!({ "id": "user_2", "name": "Ania" })),
            ],
        );

        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
        assert!(catalog.tables.contains_key("users"));
    }

    #[test]
    fn migrates_a_unique_field_that_has_no_live_duplicate() {
        let dir = tempdir().unwrap();
        write_schema_file(
            dir.path(),
            "users",
            json!({ "email": { "type": "string", "unique": true } }),
        );
        write_log_file(
            dir.path(),
            "users",
            &[
                ("user_1", json!({ "id": "user_1", "email": "ania@example.test" })),
                ("user_2", json!({ "id": "user_2", "email": "piotr@example.test" })),
            ],
        );

        let catalog = migrate_format_1_to_2(dir.path()).unwrap();
        assert!(catalog.tables.contains_key("users"));
    }

    #[test]
    fn refuses_the_whole_migration_when_one_table_has_a_live_unique_duplicate() {
        let dir = tempdir().unwrap();
        write_schema_file(
            dir.path(),
            "users",
            json!({ "email": { "type": "string", "unique": true } }),
        );
        write_log_file(
            dir.path(),
            "users",
            &[
                ("user_1", json!({ "id": "user_1", "email": "ania@example.test" })),
                ("user_2", json!({ "id": "user_2", "email": "ania@example.test" })),
            ],
        );
        write_schema_file(dir.path(), "orders", json!({ "userId": { "type": "string" } }));
        write_log_file(
            dir.path(),
            "orders",
            &[("order_1", json!({ "id": "order_1", "userId": "user_1" }))],
        );

        let error = migrate_format_1_to_2(dir.path()).unwrap_err();
        assert!(error.0.contains("\"email\""));
        assert!(error.0.contains("\"users\""));
        assert!(error.0.contains("nothing on disk was changed"));
    }
    #[test]
    fn refuses_a_partitioned_table_whose_partitions_together_hold_a_unique_duplicate() {
        let dir = tempdir().unwrap();
        let mut document = schema_document(
            "users",
            json!({
                "kind": { "type": "string", "required": true },
                "email": { "type": "string", "unique": true },
            }),
        );
        document["partitionBy"] = json!("kind");
        write_schema_document(dir.path(), "users", &document);
        write_log_at(
            &dir.path().join("users").join(format!("staff{LOG_SUFFIX}")),
            &[(
                "user_1",
                json!({ "id": "user_1", "kind": "staff", "email": "ania@example.test" }),
            )],
        );
        write_log_at(
            &dir.path().join("users").join(format!("guest{LOG_SUFFIX}")),
            &[(
                "user_2",
                json!({ "id": "user_2", "kind": "guest", "email": "ania@example.test" }),
            )],
        );

        let error = migrate_format_1_to_2(dir.path()).unwrap_err();

        assert!(error.0.contains("\"email\""), "{}", error.0);
        assert!(error.0.contains("user_1") && error.0.contains("user_2"), "{}", error.0);
        assert!(error.0.contains("nothing on disk was changed"), "{}", error.0);
    }

    #[test]
    fn migrates_a_partitioned_table_whose_partitions_hold_no_duplicate() {
        let dir = tempdir().unwrap();
        let mut document = schema_document(
            "users",
            json!({
                "kind": { "type": "string", "required": true },
                "email": { "type": "string", "unique": true },
            }),
        );
        document["partitionBy"] = json!("kind");
        write_schema_document(dir.path(), "users", &document);
        write_log_at(
            &dir.path().join("users").join(format!("staff{LOG_SUFFIX}")),
            &[(
                "user_1",
                json!({ "id": "user_1", "kind": "staff", "email": "ania@example.test" }),
            )],
        );
        write_log_at(
            &dir.path().join("users").join(format!("guest{LOG_SUFFIX}")),
            &[(
                "user_2",
                json!({ "id": "user_2", "kind": "guest", "email": "piotr@example.test" }),
            )],
        );

        let catalog = migrate_format_1_to_2(dir.path()).unwrap();

        assert!(catalog.tables.contains_key("users"));
    }

    #[test]
    fn refuses_a_nested_layout_table_whose_log_holds_a_unique_duplicate() {
        let dir = tempdir().unwrap();
        fs::write(
            dir.path().join("hayson.json"),
            json!({ "layout": "nested" }).to_string(),
        )
        .unwrap();
        write_schema_file(
            &dir.path().join("schema"),
            "users",
            json!({ "email": { "type": "string", "unique": true } }),
        );
        write_log_file(
            &dir.path().join("tables"),
            "users",
            &[
                ("user_1", json!({ "id": "user_1", "email": "ania@example.test" })),
                ("user_2", json!({ "id": "user_2", "email": "ania@example.test" })),
            ],
        );

        let error = migrate_format_1_to_2(dir.path()).unwrap_err();

        assert!(error.0.contains("\"email\""), "{}", error.0);
        assert!(error.0.contains("user_1") && error.0.contains("user_2"), "{}", error.0);
        assert!(error.0.contains("nothing on disk was changed"), "{}", error.0);
    }

    #[test]
    fn migrates_a_nested_layout_table_whose_log_holds_no_duplicate() {
        let dir = tempdir().unwrap();
        fs::write(
            dir.path().join("hayson.json"),
            json!({ "layout": "nested" }).to_string(),
        )
        .unwrap();
        write_schema_file(
            &dir.path().join("schema"),
            "users",
            json!({ "email": { "type": "string", "unique": true } }),
        );
        write_log_file(
            &dir.path().join("tables"),
            "users",
            &[
                ("user_1", json!({ "id": "user_1", "email": "ania@example.test" })),
                ("user_2", json!({ "id": "user_2", "email": "piotr@example.test" })),
            ],
        );

        let catalog = migrate_format_1_to_2(dir.path()).unwrap();

        assert_eq!(catalog.layout, Some(CatalogLayout::Nested));
        assert!(catalog.tables.contains_key("users"));
    }

    fn write_format_1(dir: &Path) {
        fs::write(
            dir.join(CATALOG_FILE),
            "{\n  \"hayson\": 1,\n  \"tables\": [\"users\"]\n}\n",
        )
        .unwrap();
        fs::write(
            dir.join("users.schema.json"),
            serde_json::to_string(&json!({
                "hayson": 1,
                "table": "users",
                "primaryKey": "id",
                "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
            }))
            .unwrap(),
        )
        .unwrap();
        fs::write(
            dir.join("users.jsonl"),
            "{\"_k\":\"user_1\",\"_op\":\"put\",\"_seq\":1,\"_ts\":1000,\"id\":\"user_1\",\"name\":\"Ania\"}\n",
        )
        .unwrap();
    }

    #[test]
    fn migrate_format1_rewrites_a_format_1_directory_into_a_catalog_the_engine_opens() {
        use crate::database::Database;

        let dir = tempdir().unwrap();
        write_format_1(dir.path());
        assert!(Database::open(dir.path()).is_err());

        let outcome = migrate_format1(dir.path()).unwrap();

        let MigrateOutcome::Migrated {
            tables,
            removed_schema_files,
            ..
        } = outcome
        else {
            panic!("a format-1 directory has to report a migration, not a no-op");
        };
        assert_eq!(tables, vec!["users".to_string()]);
        assert_eq!(removed_schema_files, vec![dir.path().join("users.schema.json")]);
        assert!(!dir.path().join("users.schema.json").exists());

        let db = Database::open(dir.path()).unwrap();
        assert_eq!(db.table_names(), vec!["users"]);
        assert_eq!(
            db.table("users").unwrap().get("user_1").unwrap().unwrap().get("name"),
            Some(&Value::from("Ania"))
        );
    }

    #[test]
    fn migrate_format1_leaves_the_log_byte_identical() {
        let dir = tempdir().unwrap();
        write_format_1(dir.path());
        let before = fs::read(dir.path().join("users.jsonl")).unwrap();

        migrate_format1(dir.path()).unwrap();

        assert_eq!(fs::read(dir.path().join("users.jsonl")).unwrap(), before);
    }

    #[test]
    fn migrate_format1_a_second_run_reports_a_no_op_and_rewrites_nothing() {
        let dir = tempdir().unwrap();
        write_format_1(dir.path());
        migrate_format1(dir.path()).unwrap();
        let bytes = fs::read_to_string(dir.path().join(CATALOG_FILE)).unwrap();

        let outcome = migrate_format1(dir.path()).unwrap();

        assert!(matches!(outcome, MigrateOutcome::AlreadyMigrated { .. }));
        assert_eq!(fs::read_to_string(dir.path().join(CATALOG_FILE)).unwrap(), bytes);
    }

    #[test]
    fn migrate_format1_refuses_a_directory_holding_no_schema_files_without_writing_a_catalog() {
        let dir = tempdir().unwrap();

        let failure = migrate_format1(dir.path()).unwrap_err();

        assert!(matches!(failure, MigrateFormat1Error::NoSchemaFiles(_)));
        assert!(!dir.path().join(CATALOG_FILE).exists());
    }

    #[test]
    fn migrate_format1_refuses_a_marker_that_is_neither_1_nor_2_and_keeps_the_schema_files() {
        let dir = tempdir().unwrap();
        write_format_1(dir.path());
        fs::write(dir.path().join(CATALOG_FILE), "{ \"hayson\": 3 }\n").unwrap();

        let failure = migrate_format1(dir.path()).unwrap_err();

        assert!(matches!(failure, MigrateFormat1Error::UnknownMarker(_, _)));
        assert!(dir.path().join("users.schema.json").exists());
    }

    #[test]
    fn migrate_format1_refuses_while_an_unfinished_unit_of_work_is_on_disk() {
        let dir = tempdir().unwrap();
        write_format_1(dir.path());
        fs::write(dir.path().join(".hayson.txn"), "{}\n").unwrap();

        let failure = migrate_format1(dir.path()).unwrap_err();

        assert!(matches!(failure, MigrateFormat1Error::Schema(_)));
        assert!(failure.to_string().contains(".hayson.txn"));
        assert!(dir.path().join("users.schema.json").exists());
        assert_eq!(marker_of(&dir.path().join(CATALOG_FILE)).unwrap(), Some(1));
    }

    #[test]
    fn migrate_format1_refuses_a_schema_file_naming_a_table_outside_the_directory_before_writing_anything() {
        let dir = tempdir().unwrap();
        write_format_1(dir.path());
        let schema: Value =
            serde_json::from_str(&fs::read_to_string(dir.path().join("users.schema.json")).unwrap()).unwrap();
        let mut traversing = schema.clone();
        traversing["table"] = Value::from("../outside");
        fs::write(
            dir.path().join("outside.schema.json"),
            serde_json::to_string(&traversing).unwrap(),
        )
        .unwrap();

        let failure = migrate_format1(dir.path()).unwrap_err();

        assert!(failure.to_string().contains("unusable table name"));
        assert_eq!(marker_of(&dir.path().join(CATALOG_FILE)).unwrap(), Some(1));
        assert!(dir.path().join("users.schema.json").exists());
    }

    #[cfg(unix)]
    #[test]
    fn migrate_format1_writes_the_catalog_no_looser_than_the_manifest_it_replaces() {
        use std::os::unix::fs::PermissionsExt;

        let dir = tempdir().unwrap();
        write_format_1(dir.path());
        let catalog_path = dir.path().join(CATALOG_FILE);
        fs::set_permissions(&catalog_path, fs::Permissions::from_mode(0o600)).unwrap();

        migrate_format1(dir.path()).unwrap();

        assert_eq!(fs::metadata(&catalog_path).unwrap().permissions().mode() & 0o777, 0o600);
    }

    #[test]
    fn migrate_format1_refuses_a_path_that_is_not_a_directory() {
        let dir = tempdir().unwrap();
        let file = dir.path().join("hayson.json");
        fs::write(&file, "{}").unwrap();

        assert!(matches!(
            migrate_format1(&file).unwrap_err(),
            MigrateFormat1Error::NotADirectory(_)
        ));
    }
}