hayson-engine 0.16.0

use std::collections::{HashMap, HashSet};
use std::io;

use serde_json::Value;

use crate::query::builder::QueryBuilder;
use crate::schema::HaysonRecord;
use crate::table::Table;

use super::types::{Catalog, Relation};

/// One record holding a reference no record on the other side of the relation answers.
#[derive(Debug, Clone, PartialEq)]
pub struct OrphanRecord {
    pub relation: String,
    pub table: String,
    pub field: String,
    pub primary_key: Value,
    pub value: Value,
}

/// What `check_integrity` found. An empty `orphans` means every declared relation holds right now.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct IntegrityReport {
    pub orphans: Vec<OrphanRecord>,
}

pub fn relations_touching(catalog: &Catalog, table: Option<&str>) -> Vec<Relation> {
    let declared = catalog.relations.iter().flatten();
    match table {
        None => declared.cloned().collect(),
        Some(table) => declared
            .filter(|relation| relation.from.table == table || relation.to.table == table)
            .cloned()
            .collect(),
    }
}

fn reference_key(value: &Value) -> String {
    match value {
        Value::String(s) => format!("s:{s}"),
        Value::Number(n) => format!("n:{n}"),
        Value::Bool(b) => format!("b:{b}"),
        other => format!("j:{other}"),
    }
}

/// A reference is a scalar the other side of a relation can hold as a primary key or a unique
/// value — an array or object can never match one, and `null`/missing is unlinked, not orphaned.
pub(crate) fn reference_of(value: Option<&Value>) -> Option<Value> {
    match value {
        None | Some(Value::Null) | Some(Value::Array(_)) | Some(Value::Object(_)) => None,
        Some(other) => Some(other.clone()),
    }
}

fn records_of(
    tables: &HashMap<String, &Table>,
    cache: &mut HashMap<String, Vec<HaysonRecord>>,
    name: &str,
) -> io::Result<()> {
    if cache.contains_key(name) {
        return Ok(());
    }
    let rows = match tables.get(name) {
        None => Vec::new(),
        Some(table) => {
            let spec = QueryBuilder::new().spec();
            let result = table.query(&spec)?;
            let mut rows = Vec::with_capacity(result.keys.len());
            for key in &result.keys {
                if let Some(record) = table.get(key)? {
                    rows.push(record);
                }
            }
            rows
        }
    };
    cache.insert(name.to_string(), rows);
    Ok(())
}

/// `db.checkIntegrity()` — walks every declared relation and reports the records whose reference
/// has no counterpart. A diagnosis, not a gate: nothing here is repaired, removed or written.
/// `tables` supplies the open handles this call can read; a relation naming a table not in it
/// reads as empty, exactly like the TS reference's `this.#tables.has(name) ? ... : []`.
pub fn check_integrity(catalog: &Catalog, tables: &HashMap<String, &Table>) -> io::Result<IntegrityReport> {
    let mut orphans = Vec::new();
    let mut cache: HashMap<String, Vec<HaysonRecord>> = HashMap::new();

    for relation in catalog.relations.iter().flatten() {
        records_of(tables, &mut cache, &relation.from.table)?;
        if cache[&relation.from.table].is_empty() {
            continue;
        }

        records_of(tables, &mut cache, &relation.to.table)?;
        let known: HashSet<String> = cache[&relation.to.table]
            .iter()
            .filter_map(|target| reference_of(target.get(&relation.to.field)))
            .map(|value| reference_key(&value))
            .collect();

        let primary_key = catalog
            .tables
            .get(&relation.from.table)
            .map(|entry| entry.primary_key.clone());
        for origin in &cache[&relation.from.table] {
            let Some(value) = reference_of(origin.get(&relation.from.field)) else {
                continue;
            };
            if known.contains(&reference_key(&value)) {
                continue;
            }
            let primary_key_value = primary_key
                .as_ref()
                .and_then(|key| origin.get(key))
                .cloned()
                .unwrap_or(Value::Null);
            orphans.push(OrphanRecord {
                relation: relation.name.clone(),
                table: relation.from.table.clone(),
                field: relation.from.field.clone(),
                primary_key: primary_key_value,
                value,
            });
        }
    }

    Ok(IntegrityReport { orphans })
}

#[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, RelationAction, RelationCardinality, RelationEnd};
    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();
        let mut catalog = create_table(&catalog, "orders", &orders, 1_000);
        catalog.relations = Some(vec![Relation {
            name: "orders_user".to_string(),
            from: RelationEnd {
                table: "orders".to_string(),
                field: "userId".to_string(),
            },
            to: RelationEnd {
                table: "users".to_string(),
                field: "id".to_string(),
            },
            cardinality: RelationCardinality::ManyToOne,
            on_delete: Some(RelationAction::Restrict),
            enforce: Some(false),
        }]);
        catalog
    }

    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()
    }

    #[test]
    fn relations_touching_filters_by_either_end() {
        let catalog = users_orders_catalog();
        assert_eq!(relations_touching(&catalog, Some("users")).len(), 1);
        assert_eq!(relations_touching(&catalog, Some("orders")).len(), 1);
        assert_eq!(relations_touching(&catalog, Some("nope")).len(), 0);
        assert_eq!(relations_touching(&catalog, None).len(), 1);
    }

    #[test]
    fn reports_a_record_whose_reference_has_no_counterpart() {
        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();
        orders.put(&json!({ "id": "o2", "userId": "ghost" }), false).unwrap();

        let catalog = users_orders_catalog();
        let mut tables: HashMap<String, &Table> = HashMap::new();
        tables.insert("users".to_string(), &users);
        tables.insert("orders".to_string(), &orders);

        let report = check_integrity(&catalog, &tables).unwrap();
        assert_eq!(report.orphans.len(), 1);
        assert_eq!(report.orphans[0].primary_key, json!("o2"));
        assert_eq!(report.orphans[0].value, json!("ghost"));
    }

    #[test]
    fn a_record_with_no_value_on_the_from_field_is_unlinked_not_orphaned() {
        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" }), false).unwrap();

        let catalog = users_orders_catalog();
        let mut tables: HashMap<String, &Table> = HashMap::new();
        tables.insert("users".to_string(), &users);
        tables.insert("orders".to_string(), &orders);

        let report = check_integrity(&catalog, &tables).unwrap();
        assert!(report.orphans.is_empty());
    }

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

        let catalog = users_orders_catalog();
        let mut tables: HashMap<String, &Table> = HashMap::new();
        tables.insert("orders".to_string(), &orders);

        let report = check_integrity(&catalog, &tables).unwrap();
        assert_eq!(report.orphans.len(), 1);
    }
}