hayson-engine 0.11.0

use std::collections::HashMap;

use serde_json::Value;

use crate::errors::HaysonSchemaError;
use crate::table::Table;

use super::integrity::check_integrity;
use super::parse::{parse_catalog, parse_relation};
use super::serialize::catalog_document;
use super::types::{Catalog, Relation};

const LISTED_KEYS: usize = 20;

fn describe_orphans(relation: &Relation, report: &super::integrity::IntegrityReport) -> Option<String> {
    let mut keys: Vec<String> = report
        .orphans
        .iter()
        .filter(|orphan| orphan.relation == relation.name)
        .map(|orphan| orphan.primary_key.to_string())
        .collect();
    if keys.is_empty() {
        return None;
    }
    keys.sort();
    let listed = keys.iter().take(LISTED_KEYS).cloned().collect::<Vec<_>>().join(", ");
    let rest = if keys.len() > LISTED_KEYS {
        format!(" and {} more", keys.len() - LISTED_KEYS)
    } else {
        String::new()
    };
    Some(format!(
        "refusing to declare relation \"{}\" with enforce: true: {} record(s) of \"{}\" already point at a value \"{}\" has no counterpart for: {listed}{rest} — nothing on disk was changed; pass force: true to declare it anyway (the orphans stay orphans, enforcement just starts from here on), or fix the data first",
        relation.name,
        keys.len(),
        relation.from.table,
        relation.to.table,
    ))
}

/// `db.declareRelation()` — adds a named relation to the catalog. Mirrors `declare_index`'s shape:
/// parses and validates against the catalog held in memory, refuses without writing anything, and
/// leaves `write_catalog` to the caller.
///
/// When the new relation declares `enforce: true`, this additionally runs the same orphan check
/// `checkIntegrity` does (`catalog::integrity::check_integrity`, scoped to just this relation) and
/// refuses — unless `force` is `true` — if any live record already breaks it: turning enforcement
/// on does not retroactively catch what was written before it existed, so declaring it silently
/// over existing orphans would make a promise the data has already broken. `force: true` accepts
/// that and declares anyway; the orphans already there stay orphans, enforcement starts from here.
pub fn declare_relation(
    catalog: &Catalog,
    spec: &Value,
    from_table_handle: &Table,
    to_table_handle: &Table,
    now: u64,
    force: bool,
) -> Result<Catalog, HaysonSchemaError> {
    let declared = catalog.relations.clone().unwrap_or_default();
    let path = format!("relations[{}]", declared.len());
    let relation = parse_relation(spec, &path, &catalog.tables)?;

    if declared.iter().any(|other| other.name == relation.name) {
        return Err(HaysonSchemaError(format!(
            "catalog already declares a relation named \"{}\" — a relation name is what the catalog identifies it by, so take the declared one off the catalog before declaring another under that name",
            relation.name
        )));
    }

    let mut next_declared = declared;
    next_declared.push(relation.clone());
    let mut next_catalog = catalog.clone();
    next_catalog.updated_at = now;
    next_catalog.relations = Some(next_declared);

    parse_catalog(&catalog_document(&next_catalog)).map_err(|error| {
        HaysonSchemaError(format!(
            "declaring relation \"{}\" would leave a catalog this build of hayson cannot read back — {}; nothing was changed",
            relation.name, error.0
        ))
    })?;

    if relation.enforce == Some(true) && !force {
        let mut tables: HashMap<String, &Table> = HashMap::new();
        tables.insert(relation.from.table.clone(), from_table_handle);
        tables.insert(relation.to.table.clone(), to_table_handle);
        let report = check_integrity(&next_catalog, &tables).map_err(|error| {
            HaysonSchemaError(format!(
                "reading \"{}\"/\"{}\" to check relation \"{}\": {error}",
                relation.from.table, relation.to.table, relation.name
            ))
        })?;
        if let Some(message) = describe_orphans(&relation, &report) {
            return Err(HaysonSchemaError(message));
        }
    }

    Ok(next_catalog)
}

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

    use super::*;
    use crate::catalog::build::{create_table, empty_catalog};
    use crate::catalog::types::CatalogLayout;
    use crate::schema::parse_schema;

    fn users_orders_catalog() -> Catalog {
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let users = parse_schema(&json!({
            "schemaVersion": 1, "table": "users", "primaryKey": "id",
            "fields": { "id": { "type": "string" } },
        }))
        .unwrap();
        let catalog = create_table(&catalog, "users", &users, 1_000);
        let orders = parse_schema(&json!({
            "schemaVersion": 1, "table": "orders", "primaryKey": "id",
            "fields": { "id": { "type": "string" }, "userId": { "type": "string" } },
        }))
        .unwrap();
        create_table(&catalog, "orders", &orders, 1_000)
    }

    fn open_table(dir: &tempfile::TempDir, name: &str, fields: Value) -> Table {
        let mut document =
            json!({ "schemaVersion": 1, "table": name, "primaryKey": "id", "fields": { "id": { "type": "string" } } });
        for (key, value) in fields.as_object().unwrap() {
            document["fields"][key] = value.clone();
        }
        let schema = parse_schema(&document).unwrap();
        Table::open(schema, dir.path().join(format!("{name}.jsonl"))).unwrap()
    }

    fn relation_spec(enforce: bool) -> Value {
        json!({
            "name": "orders_user",
            "from": { "table": "orders", "field": "userId" },
            "to": { "table": "users", "field": "id" },
            "cardinality": "many-to-one",
            "enforce": enforce,
        })
    }

    #[test]
    fn declares_a_relation_with_no_orphans_to_check() {
        let dir = tempdir().unwrap();
        let users = open_table(&dir, "users", json!({}));
        let orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
        let catalog = users_orders_catalog();

        let next = declare_relation(&catalog, &relation_spec(false), &orders, &users, 2_000, false).unwrap();

        let relations = next.relations.unwrap();
        assert_eq!(relations.len(), 1);
        assert_eq!(relations[0].name, "orders_user");
        assert_eq!(next.updated_at, 2_000);
    }

    #[test]
    fn refuses_enforce_true_when_a_live_record_already_orphans_it() {
        let dir = tempdir().unwrap();
        let users = open_table(&dir, "users", json!({}));
        let mut orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
        orders.put(&json!({ "id": "o1", "userId": "ghost" }), false).unwrap();
        let catalog = users_orders_catalog();

        let error = declare_relation(&catalog, &relation_spec(true), &orders, &users, 2_000, false).unwrap_err();

        assert!(error.0.contains("orders_user"));
        assert!(error.0.contains("o1"));
        assert!(error.0.contains("force"));
        assert!(catalog.relations.is_none());
    }

    #[test]
    fn force_declares_enforce_true_over_existing_orphans() {
        let dir = tempdir().unwrap();
        let users = open_table(&dir, "users", json!({}));
        let mut orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
        orders.put(&json!({ "id": "o1", "userId": "ghost" }), false).unwrap();
        let catalog = users_orders_catalog();

        let next = declare_relation(&catalog, &relation_spec(true), &orders, &users, 2_000, true).unwrap();

        assert_eq!(next.relations.unwrap()[0].enforce, Some(true));
    }

    #[test]
    fn enforce_true_declares_cleanly_when_every_reference_resolves() {
        let dir = tempdir().unwrap();
        let mut users = open_table(&dir, "users", json!({}));
        let mut orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
        users.put(&json!({ "id": "u1" }), false).unwrap();
        orders.put(&json!({ "id": "o1", "userId": "u1" }), false).unwrap();
        let catalog = users_orders_catalog();

        let next = declare_relation(&catalog, &relation_spec(true), &orders, &users, 2_000, false).unwrap();

        assert_eq!(next.relations.unwrap()[0].enforce, Some(true));
    }

    #[test]
    fn refuses_a_duplicate_relation_name() {
        let dir = tempdir().unwrap();
        let users = open_table(&dir, "users", json!({}));
        let orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
        let catalog = users_orders_catalog();
        let catalog = declare_relation(&catalog, &relation_spec(false), &orders, &users, 2_000, false).unwrap();

        let error = declare_relation(&catalog, &relation_spec(false), &orders, &users, 3_000, false).unwrap_err();
        assert!(error.0.contains("already declares a relation named \"orders_user\""));
    }
}