hayson-engine 0.8.0

use std::collections::{BTreeMap, HashSet};
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,
    })
}

#[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"));
    }
}