hayson_engine/catalog/

parse.rs

use std::collections::BTreeMap;

use serde_json::{Map, Value};

use crate::errors::HaysonSchemaError;
use crate::schema::{parse_schema, FieldType, SCHEMA_VERSION};

use super::indexes::expand_indexes;
use super::types::{
    Catalog, CatalogCompaction, CatalogFieldSpec, CatalogIndexKind, CatalogLayout, CatalogPredicate, CatalogSyncMode,
    CatalogValidation, CheckEntry, HistoryEntry, IndexEntry, LinkParsing, Relation, RelationAction,
    RelationCardinality, RelationEnd, TableEntry, TableOptions, TriggerEffect, TriggerEntry, TriggerEvent,
    TriggerFieldMapping, TriggerValueSource, ViewEntry, CATALOG_FORMAT_VERSION, CATALOG_PREDICATE_OPS, CATALOG_VERSION,
};

const CATALOG_KEYS: &[&str] = &[
    "catalogVersion",
    // NOTE: "hayson" is the pre-5.0.0 name of "catalogVersion" — still read, never written again.
    "hayson",
    "formatVersion",
    "createdAt",
    "updatedAt",
    "engine",
    "layout",
    "defaults",
    "tables",
    "relations",
    "views",
    "triggers",
];
const TABLE_KEYS: &[&str] = &[
    "revision",
    "description",
    "primaryKey",
    "partitionBy",
    "createdAt",
    "updatedAt",
    "validation",
    "options",
    "fields",
    "indexes",
    "checks",
    "history",
];
const INDEX_KEYS: &[&str] = &["name", "fields", "kind", "unique", "where", "expireAfterSeconds"];
const PREDICATE_KEYS: &[&str] = &["field", "op", "value"];
const CHECK_KEYS: &[&str] = &["name", "expr"];
const HISTORY_KEYS: &[&str] = &["revision", "at", "note"];
const OPTION_KEYS: &[&str] = &["sync", "compaction", "linkParsing", "imageDir"];
const COMPACTION_KEYS: &[&str] = &["deadBytesRatio", "minDeadBytes"];
const RELATION_KEYS: &[&str] = &["name", "from", "to", "cardinality", "onDelete", "enforce"];
const RELATION_END_KEYS: &[&str] = &["table", "field"];
const VIEW_KEYS: &[&str] = &["name", "query", "description"];
const TRIGGER_KEYS: &[&str] = &["name", "table", "event", "effect"];
const TRIGGER_EFFECT_KEYS: &[&str] = &["table", "fields"];
const TRIGGER_FIELD_MAPPING_KEYS: &[&str] = &["target", "source"];
const TRIGGER_SOURCE_KEYS: &[&str] = &["kind", "field", "value"];
const FIELD_META_KEYS: &[&str] = &["description", "deprecated"];

fn fail(path: &str, message: impl Into<String>) -> HaysonSchemaError {
    HaysonSchemaError(format!("{path} {}", message.into()))
}

fn is_object_name(value: &str) -> bool {
    let mut chars = value.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphabetic() => {}
        _ => return false,
    }
    chars.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
}

fn as_object<'a>(value: Option<&'a Value>, path: &str) -> Result<&'a Map<String, Value>, HaysonSchemaError> {
    value
        .and_then(Value::as_object)
        .ok_or_else(|| fail(path, "must be an object"))
}

fn as_array<'a>(value: Option<&'a Value>, path: &str) -> Result<&'a Vec<Value>, HaysonSchemaError> {
    value
        .and_then(Value::as_array)
        .ok_or_else(|| fail(path, "must be an array"))
}

fn as_name(value: Option<&Value>, path: &str, what: &str) -> Result<String, HaysonSchemaError> {
    let text = value.and_then(Value::as_str).filter(|s| is_object_name(s));
    text.map(str::to_string)
        .ok_or_else(|| fail(path, format!("must be a usable {what} name, got {:?}", value)))
}

fn as_text(value: Option<&Value>, path: &str) -> Result<String, HaysonSchemaError> {
    match value.and_then(Value::as_str) {
        Some(s) if !s.is_empty() => Ok(s.to_string()),
        _ => Err(fail(path, "must be a non-empty string")),
    }
}

fn optional_text(value: Option<&Value>, path: &str) -> Result<Option<String>, HaysonSchemaError> {
    match value {
        None => Ok(None),
        Some(_) => as_text(value, path).map(Some),
    }
}

fn optional_flag(value: Option<&Value>, path: &str) -> Result<Option<bool>, HaysonSchemaError> {
    match value {
        None => Ok(None),
        Some(Value::Bool(b)) => Ok(Some(*b)),
        Some(_) => Err(fail(path, "must be a boolean")),
    }
}

fn as_whole_number(value: Option<&Value>, path: &str, minimum: i64) -> Result<u64, HaysonSchemaError> {
    match value.and_then(Value::as_i64) {
        Some(n) if n >= minimum => Ok(n as u64),
        _ => Err(fail(
            path,
            format!("must be an integer of at least {minimum}, got {:?}", value),
        )),
    }
}

fn optional_whole_number(value: Option<&Value>, path: &str, minimum: i64) -> Result<Option<u64>, HaysonSchemaError> {
    match value {
        None => Ok(None),
        Some(_) => as_whole_number(value, path, minimum).map(Some),
    }
}

fn reject_unknown(source: &Map<String, Value>, allowed: &[&str], path: &str) -> Result<(), HaysonSchemaError> {
    for key in source.keys() {
        if !allowed.contains(&key.as_str()) {
            return Err(fail(
                &format!("{path}.{key}"),
                "is not a known catalog option — it was most likely written by a newer version of hayson, and guessing what it means would read the database wrongly. Upgrade @theritual/hayson.",
            ));
        }
    }
    Ok(())
}

fn parse_compaction(raw: Option<&Value>, path: &str) -> Result<Option<CatalogCompaction>, HaysonSchemaError> {
    let Some(raw) = raw else { return Ok(None) };
    if raw == &Value::Bool(false) {
        return Ok(Some(CatalogCompaction::Disabled));
    }
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, COMPACTION_KEYS, path)?;
    let ratio = source.get("deadBytesRatio").and_then(Value::as_f64);
    let ratio = match ratio {
        Some(r) if r > 0.0 && r <= 1.0 => r,
        _ => {
            return Err(fail(
                &format!("{path}.deadBytesRatio"),
                "must be a number greater than 0 and at most 1",
            ))
        }
    };
    let min_dead_bytes = as_whole_number(source.get("minDeadBytes"), &format!("{path}.minDeadBytes"), 0)?;
    Ok(Some(CatalogCompaction::Enabled {
        dead_bytes_ratio: ratio,
        min_dead_bytes,
    }))
}

fn parse_options(raw: Option<&Value>, path: &str) -> Result<Option<TableOptions>, HaysonSchemaError> {
    let Some(raw) = raw else { return Ok(None) };
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, OPTION_KEYS, path)?;
    let sync = match source.get("sync").and_then(Value::as_str) {
        None => None,
        Some("commit") => Some(CatalogSyncMode::Commit),
        Some("never") => Some(CatalogSyncMode::Never),
        Some(_) => return Err(fail(&format!("{path}.sync"), "must be one of \"commit\", \"never\"")),
    };
    let compaction = parse_compaction(source.get("compaction"), &format!("{path}.compaction"))?;
    let link_parsing = match source.get("linkParsing").and_then(Value::as_str) {
        None => None,
        Some("eager") => Some(LinkParsing::Eager),
        Some("lazy") => Some(LinkParsing::Lazy),
        Some(_) => {
            return Err(fail(
                &format!("{path}.linkParsing"),
                "must be one of \"eager\", \"lazy\"",
            ))
        }
    };
    let image_dir = optional_text(source.get("imageDir"), &format!("{path}.imageDir"))?;
    Ok(Some(TableOptions {
        sync,
        compaction,
        link_parsing,
        image_dir,
    }))
}

fn parse_predicate_value(raw: &Value, path: &str) -> Result<Value, HaysonSchemaError> {
    match raw {
        Value::Null | Value::String(_) | Value::Number(_) | Value::Bool(_) => Ok(raw.clone()),
        Value::Array(items) => {
            for (at, item) in items.iter().enumerate() {
                if !(item.is_string() || item.is_number() || item.is_boolean()) {
                    return Err(fail(
                        &format!("{path}[{at}]"),
                        "must be a string, a number or a boolean",
                    ));
                }
            }
            Ok(raw.clone())
        }
        _ => Err(fail(path, "must be a string, number, boolean, array or null")),
    }
}

fn parse_predicate(
    raw: &Value,
    path: &str,
    fields: &BTreeMap<String, CatalogFieldSpec>,
) -> Result<CatalogPredicate, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, PREDICATE_KEYS, path)?;
    let field = as_text(source.get("field"), &format!("{path}.field"))?;
    let spec = fields.get(&field).ok_or_else(|| {
        fail(
            &format!("{path}.field"),
            format!("names a field \"{field}\" the table does not declare"),
        )
    })?;
    if spec.spec.cypher || spec.spec.password_hash {
        return Err(fail(
            &format!("{path}.field"),
            format!("names the encrypted field \"{field}\" — the registry holds it enciphered, so a predicate against it can never match"),
        ));
    }
    let op = as_text(source.get("op"), &format!("{path}.op"))?;
    if !CATALOG_PREDICATE_OPS.contains(&op.as_str()) {
        return Err(fail(
            &format!("{path}.op"),
            format!("must be one of {}", CATALOG_PREDICATE_OPS.join(", ")),
        ));
    }
    let value = parse_predicate_value(source.get("value").unwrap_or(&Value::Null), &format!("{path}.value"))?;
    Ok(CatalogPredicate { field, op, value })
}

fn sortable_in_index_of(field_type: FieldType, width: usize) -> bool {
    if width == 1 {
        return field_type.is_sortable();
    }
    !field_type.is_array() && field_type != FieldType::Json && !field_type.has_virtual_path()
}

pub fn parse_index_entry(
    raw: &Value,
    path: &str,
    fields: &BTreeMap<String, CatalogFieldSpec>,
) -> Result<IndexEntry, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, INDEX_KEYS, path)?;

    let name = as_name(source.get("name"), &format!("{path}.name"), "index")?;
    let kind_raw = source
        .get("kind")
        .and_then(Value::as_str)
        .ok_or_else(|| fail(&format!("{path}.kind"), "is required"))?;
    let kind = match kind_raw {
        "exact" => CatalogIndexKind::Exact,
        "sorted" => CatalogIndexKind::Sorted,
        "fulltext" => CatalogIndexKind::Fulltext,
        "ttl" => CatalogIndexKind::Ttl,
        other => {
            return Err(fail(
                &format!("{path}.kind"),
                format!("must be one of exact, sorted, fulltext, ttl, got {other:?}"),
            ))
        }
    };

    let raw_fields = as_array(source.get("fields"), &format!("{path}.fields"))?;
    if raw_fields.is_empty() {
        return Err(fail(&format!("{path}.fields"), "must name at least one field"));
    }
    let mut named: Vec<String> = Vec::new();
    for (at, entry) in raw_fields.iter().enumerate() {
        let field = as_text(Some(entry), &format!("{path}.fields[{at}]"))?;
        if named.contains(&field) {
            return Err(fail(
                &format!("{path}.fields[{at}]"),
                format!("repeats the field \"{field}\""),
            ));
        }
        let spec = fields.get(&field).ok_or_else(|| {
            fail(
                &format!("{path}.fields[{at}]"),
                format!("names a field \"{field}\" the table does not declare"),
            )
        })?;
        if spec.spec.cypher || spec.spec.password_hash {
            return Err(fail(
                &format!("{path}.fields[{at}]"),
                format!("names the encrypted field \"{field}\" — a random IV makes two equal values differ"),
            ));
        }
        if kind == CatalogIndexKind::Sorted && !sortable_in_index_of(spec.spec.field_type, raw_fields.len()) {
            return Err(fail(
                &format!("{path}.fields[{at}]"),
                format!(
                    "names \"{field}\" of type {}, which cannot be sorted",
                    spec.spec.field_type.as_str()
                ),
            ));
        }
        if kind == CatalogIndexKind::Fulltext
            && spec.spec.field_type != FieldType::String
            && spec.spec.field_type != FieldType::StringArray
        {
            return Err(fail(
                &format!("{path}.fields[{at}]"),
                format!(
                    "names \"{field}\" of type {}, which cannot be searched as text",
                    spec.spec.field_type.as_str()
                ),
            ));
        }
        if kind == CatalogIndexKind::Ttl && spec.spec.field_type != FieldType::Timestamp {
            return Err(fail(
                &format!("{path}.fields[{at}]"),
                format!(
                    "names \"{field}\" of type {}, which is not a timestamp",
                    spec.spec.field_type.as_str()
                ),
            ));
        }
        named.push(field);
    }
    if kind == CatalogIndexKind::Fulltext && named.len() > 1 {
        return Err(fail(
            &format!("{path}.fields"),
            "must name exactly one field for a fulltext index",
        ));
    }
    if kind == CatalogIndexKind::Ttl && named.len() > 1 {
        return Err(fail(
            &format!("{path}.fields"),
            "must name exactly one field for a ttl index",
        ));
    }
    if kind == CatalogIndexKind::Fulltext && source.contains_key("where") {
        return Err(fail(
            &format!("{path}.where"),
            "cannot filter a fulltext index — a record excluded by where would simply not be searchable, which defeats the point",
        ));
    }

    let unique = optional_flag(source.get("unique"), &format!("{path}.unique"))?.unwrap_or(false);
    if unique {
        if kind != CatalogIndexKind::Exact && kind != CatalogIndexKind::Sorted {
            return Err(fail(
                &format!("{path}.unique"),
                format!("cannot hold on a {} index", kind.as_str()),
            ));
        }
        for field in &named {
            let spec = &fields[field].spec;
            if spec.field_type.is_array() {
                return Err(fail(
                    &format!("{path}.unique"),
                    format!("cannot cover \"{field}\" of type {}", spec.field_type.as_str()),
                ));
            }
        }
    }

    let expire_after_seconds = optional_whole_number(
        source.get("expireAfterSeconds"),
        &format!("{path}.expireAfterSeconds"),
        1,
    )?;
    if kind == CatalogIndexKind::Ttl && expire_after_seconds.is_none() {
        return Err(fail(
            &format!("{path}.expireAfterSeconds"),
            "is required on a ttl index",
        ));
    }
    if kind != CatalogIndexKind::Ttl && expire_after_seconds.is_some() {
        return Err(fail(
            &format!("{path}.expireAfterSeconds"),
            format!("only belongs on a ttl index, not on a {} one", kind.as_str()),
        ));
    }

    let where_ = match source.get("where") {
        None => None,
        Some(raw) => {
            let clauses = as_array(Some(raw), &format!("{path}.where"))?;
            if clauses.is_empty() {
                return Err(fail(&format!("{path}.where"), "must hold at least one predicate"));
            }
            let mut parsed = Vec::with_capacity(clauses.len());
            for (at, clause) in clauses.iter().enumerate() {
                parsed.push(parse_predicate(clause, &format!("{path}.where[{at}]"), fields)?);
            }
            Some(parsed)
        }
    };

    Ok(IndexEntry {
        name,
        fields: named,
        kind,
        unique,
        where_,
        expire_after_seconds,
    })
}

fn parse_check_entry(raw: &Value, path: &str) -> Result<CheckEntry, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, CHECK_KEYS, path)?;
    Ok(CheckEntry {
        name: as_name(source.get("name"), &format!("{path}.name"), "check")?,
        expr: as_text(source.get("expr"), &format!("{path}.expr"))?,
    })
}

fn parse_history_entry(raw: &Value, path: &str) -> Result<HistoryEntry, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, HISTORY_KEYS, path)?;
    Ok(HistoryEntry {
        revision: as_whole_number(source.get("revision"), &format!("{path}.revision"), 1)?,
        at: as_whole_number(source.get("at"), &format!("{path}.at"), 0)?,
        note: optional_text(source.get("note"), &format!("{path}.note"))?,
    })
}

fn section_document(name: &str, source: &Map<String, Value>) -> Result<Value, HaysonSchemaError> {
    let raw_fields = as_object(source.get("fields"), &format!("tables.{name}.fields"))?;
    let mut fields = Map::new();
    for (field, raw) in raw_fields {
        let spec = as_object(Some(raw), &format!("tables.{name}.fields.{field}"))?;
        let mut bare = Map::new();
        for (key, value) in spec {
            if FIELD_META_KEYS.contains(&key.as_str()) {
                continue;
            }
            bare.insert(key.clone(), value.clone());
        }
        fields.insert(field.clone(), Value::Object(bare));
    }
    let mut doc = Map::new();
    doc.insert("schemaVersion".to_string(), Value::from(SCHEMA_VERSION));
    doc.insert("table".to_string(), Value::from(name));
    if let Some(revision) = source.get("revision") {
        doc.insert("revision".to_string(), revision.clone());
    }
    if let Some(primary_key) = source.get("primaryKey") {
        doc.insert("primaryKey".to_string(), primary_key.clone());
    }
    if let Some(partition_by) = source.get("partitionBy") {
        doc.insert("partitionBy".to_string(), partition_by.clone());
    }
    doc.insert("fields".to_string(), Value::Object(fields));
    Ok(Value::Object(doc))
}

fn parse_section(name: &str, source: &Map<String, Value>) -> Result<crate::schema::TableSchema, HaysonSchemaError> {
    let document = section_document(name, source)?;
    parse_schema(&document)
        .map_err(|error| HaysonSchemaError(format!("tables.{name} is not a usable table: {}", error.0)))
}

fn catalog_fields(
    name: &str,
    schema: &crate::schema::TableSchema,
    source: &Map<String, Value>,
) -> Result<BTreeMap<String, CatalogFieldSpec>, HaysonSchemaError> {
    let raw_fields = as_object(source.get("fields"), &format!("tables.{name}.fields"))?;
    let mut fields = BTreeMap::new();
    for (field, spec) in &schema.fields {
        let raw = as_object(raw_fields.get(field), &format!("tables.{name}.fields.{field}"))?;
        let description = optional_text(
            raw.get("description"),
            &format!("tables.{name}.fields.{field}.description"),
        )?;
        let deprecated = optional_text(
            raw.get("deprecated"),
            &format!("tables.{name}.fields.{field}.deprecated"),
        )?;
        fields.insert(
            field.clone(),
            CatalogFieldSpec {
                spec: spec.clone(),
                description,
                deprecated,
            },
        );
    }
    Ok(fields)
}

fn parse_table_entry(name: &str, raw: &Value) -> Result<TableEntry, HaysonSchemaError> {
    let source = as_object(Some(raw), &format!("tables.{name}"))?;
    if let Some(declared) = source.get("table") {
        let message = match declared.as_str() {
            Some(other) if other != name => {
                format!("names the table \"{other}\" while the catalog keys the section as \"{name}\"")
            }
            _ => "must not repeat the table name — the key in \"tables\" is the name".to_string(),
        };
        return Err(fail(&format!("tables.{name}.table"), message));
    }
    for marker in ["catalogVersion", "hayson"] {
        if source.contains_key(marker) {
            return Err(fail(
                &format!("tables.{name}.{marker}"),
                "must not repeat the document marker of the catalog",
            ));
        }
    }
    reject_unknown(source, TABLE_KEYS, &format!("tables.{name}"))?;

    if let Some(revision) = source.get("revision") {
        as_whole_number(Some(revision), &format!("tables.{name}.revision"), 1)?;
    }
    if let Some(primary_key) = source.get("primaryKey") {
        as_text(Some(primary_key), &format!("tables.{name}.primaryKey"))?;
    }

    let schema = parse_section(name, source)?;
    let fields = catalog_fields(name, &schema, source)?;

    let description = optional_text(source.get("description"), &format!("tables.{name}.description"))?;
    let created_at = optional_whole_number(source.get("createdAt"), &format!("tables.{name}.createdAt"), 0)?;
    let updated_at = optional_whole_number(source.get("updatedAt"), &format!("tables.{name}.updatedAt"), 0)?;
    let validation = match source.get("validation").and_then(Value::as_str) {
        None => None,
        Some("strict") => Some(CatalogValidation::Strict),
        Some("warn") => Some(CatalogValidation::Warn),
        Some(_) => {
            return Err(fail(
                &format!("tables.{name}.validation"),
                "must be one of \"strict\", \"warn\"",
            ))
        }
    };
    let options = parse_options(source.get("options"), &format!("tables.{name}.options"))?;

    let mut entry = TableEntry {
        revision: schema.revision as u64,
        primary_key: schema.primary_key.clone(),
        fields,
        partition_by: schema.partition_by.clone(),
        description,
        created_at,
        updated_at,
        validation,
        options,
        indexes: None,
        checks: None,
        history: None,
    };

    if let Some(raw_indexes) = source.get("indexes") {
        let raw_indexes = as_array(Some(raw_indexes), &format!("tables.{name}.indexes"))?;
        let mut indexes = Vec::with_capacity(raw_indexes.len());
        for (at, raw) in raw_indexes.iter().enumerate() {
            indexes.push(parse_index_entry(
                raw,
                &format!("tables.{name}.indexes[{at}]"),
                &entry.fields,
            )?);
        }
        for at in 0..indexes.len() {
            if indexes[..at].iter().any(|other| other.name == indexes[at].name) {
                return Err(fail(
                    &format!("tables.{name}.indexes[{at}].name"),
                    format!("repeats the index name \"{}\"", indexes[at].name),
                ));
            }
        }
        entry.indexes = Some(indexes);
    }

    if let Some(raw_checks) = source.get("checks") {
        let raw_checks = as_array(Some(raw_checks), &format!("tables.{name}.checks"))?;
        let mut checks = Vec::with_capacity(raw_checks.len());
        for (at, raw) in raw_checks.iter().enumerate() {
            checks.push(parse_check_entry(raw, &format!("tables.{name}.checks[{at}]"))?);
        }
        for at in 0..checks.len() {
            if checks[..at].iter().any(|other| other.name == checks[at].name) {
                return Err(fail(
                    &format!("tables.{name}.checks[{at}].name"),
                    format!("repeats the check name \"{}\"", checks[at].name),
                ));
            }
        }
        entry.checks = Some(checks);
    }

    if let Some(raw_history) = source.get("history") {
        let raw_history = as_array(Some(raw_history), &format!("tables.{name}.history"))?;
        let mut history = Vec::with_capacity(raw_history.len());
        for (at, raw) in raw_history.iter().enumerate() {
            history.push(parse_history_entry(raw, &format!("tables.{name}.history[{at}]"))?);
        }
        entry.history = Some(history);
    }

    let expanded = expand_indexes(name, &entry);
    for at in 0..expanded.len() {
        let clash = expanded[..at].iter().any(|other| other.name == expanded[at].name);
        if !clash {
            continue;
        }
        let declared_at = entry
            .indexes
            .iter()
            .flatten()
            .position(|other| other.name == expanded[at].name);
        let path = match declared_at {
            Some(declared_at) => format!("tables.{name}.indexes[{declared_at}].name"),
            None => format!("tables.{name}.indexes"),
        };
        return Err(fail(
            &path,
            format!(
                "collides with the index the field modifiers expand to under the name \"{}\"",
                expanded[at].name
            ),
        ));
    }

    Ok(entry)
}

fn parse_relation_end(
    raw: &Value,
    path: &str,
    tables: &BTreeMap<String, TableEntry>,
) -> Result<RelationEnd, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, RELATION_END_KEYS, path)?;
    let table = as_text(source.get("table"), &format!("{path}.table"))?;
    let entry = tables.get(&table).ok_or_else(|| {
        fail(
            &format!("{path}.table"),
            format!("names a table \"{table}\" the catalog does not declare"),
        )
    })?;
    let field = as_text(source.get("field"), &format!("{path}.field"))?;
    if !entry.fields.contains_key(&field) {
        return Err(fail(
            &format!("{path}.field"),
            format!("names a field \"{field}\" table \"{table}\" does not declare"),
        ));
    }
    Ok(RelationEnd { table, field })
}

fn is_unique_side(entry: &TableEntry, table: &str, field: &str) -> bool {
    if entry.primary_key == field {
        return true;
    }
    expand_indexes(table, entry)
        .iter()
        .any(|index| index.unique && index.fields.len() == 1 && index.fields[0] == field)
}

pub fn parse_relation(
    raw: &Value,
    path: &str,
    tables: &BTreeMap<String, TableEntry>,
) -> Result<Relation, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, RELATION_KEYS, path)?;
    let name = as_name(source.get("name"), &format!("{path}.name"), "relation")?;
    let from = parse_relation_end(
        source
            .get("from")
            .ok_or_else(|| fail(&format!("{path}.from"), "is required"))?,
        &format!("{path}.from"),
        tables,
    )?;
    let to = parse_relation_end(
        source
            .get("to")
            .ok_or_else(|| fail(&format!("{path}.to"), "is required"))?,
        &format!("{path}.to"),
        tables,
    )?;
    if from.table == to.table && from.field == to.field {
        return Err(fail(path, "points at the same field on both sides"));
    }
    let target = &tables[&to.table];
    if !is_unique_side(target, &to.table, &to.field) {
        return Err(fail(
            &format!("{path}.to.field"),
            format!(
                "names \"{}\", which is neither the primary key of \"{}\" nor unique in it",
                to.field, to.table
            ),
        ));
    }
    let cardinality = match source.get("cardinality").and_then(Value::as_str) {
        Some("one-to-one") => RelationCardinality::OneToOne,
        Some("many-to-one") => RelationCardinality::ManyToOne,
        Some(_) => {
            return Err(fail(
                &format!("{path}.cardinality"),
                "must be one of \"one-to-one\", \"many-to-one\"",
            ))
        }
        None => return Err(fail(&format!("{path}.cardinality"), "is required")),
    };
    let on_delete = match source.get("onDelete").and_then(Value::as_str) {
        None => None,
        Some("restrict") => Some(RelationAction::Restrict),
        Some("cascade") => Some(RelationAction::Cascade),
        Some("set-null") => Some(RelationAction::SetNull),
        Some("no-action") => Some(RelationAction::NoAction),
        Some(_) => {
            return Err(fail(
                &format!("{path}.onDelete"),
                "must be one of \"restrict\", \"cascade\", \"set-null\", \"no-action\"",
            ))
        }
    };
    let origin = &tables[&from.table];
    if on_delete == Some(RelationAction::SetNull) {
        let required = origin
            .fields
            .get(&from.field)
            .map(|spec| spec.spec.required)
            .unwrap_or(false);
        if required {
            return Err(fail(
                &format!("{path}.onDelete"),
                format!("cannot null the required field \"{}.{}\"", from.table, from.field),
            ));
        }
    }
    let enforce = optional_flag(source.get("enforce"), &format!("{path}.enforce"))?;
    Ok(Relation {
        name,
        from,
        to,
        cardinality,
        on_delete,
        enforce,
    })
}

fn parse_view_entry(raw: &Value, path: &str) -> Result<ViewEntry, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, VIEW_KEYS, path)?;
    Ok(ViewEntry {
        name: as_name(source.get("name"), &format!("{path}.name"), "view")?,
        query: as_text(source.get("query"), &format!("{path}.query"))?,
        description: optional_text(source.get("description"), &format!("{path}.description"))?,
    })
}

fn parse_trigger_value_source(
    raw: &Value,
    path: &str,
    fields: &BTreeMap<String, CatalogFieldSpec>,
) -> Result<TriggerValueSource, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, TRIGGER_SOURCE_KEYS, path)?;
    let kind = source
        .get("kind")
        .and_then(Value::as_str)
        .ok_or_else(|| fail(&format!("{path}.kind"), "is required"))?;
    match kind {
        "new" | "old" => {
            let field = as_text(source.get("field"), &format!("{path}.field"))?;
            if !fields.contains_key(&field) {
                return Err(fail(
                    &format!("{path}.field"),
                    format!("names a field \"{field}\" the source table does not declare"),
                ));
            }
            Ok(if kind == "new" {
                TriggerValueSource::NewField(field)
            } else {
                TriggerValueSource::OldField(field)
            })
        }
        "constant" => {
            let value = source
                .get("value")
                .ok_or_else(|| fail(&format!("{path}.value"), "is required"))?;
            Ok(TriggerValueSource::Constant(value.clone()))
        }
        "now" => Ok(TriggerValueSource::Now),
        other => Err(fail(
            &format!("{path}.kind"),
            format!("must be one of \"new\", \"old\", \"constant\", \"now\", got {other:?}"),
        )),
    }
}

fn parse_trigger_field_mapping(
    raw: &Value,
    path: &str,
    source_fields: &BTreeMap<String, CatalogFieldSpec>,
    effect_fields: &BTreeMap<String, CatalogFieldSpec>,
) -> Result<TriggerFieldMapping, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, TRIGGER_FIELD_MAPPING_KEYS, path)?;
    let target = as_text(source.get("target"), &format!("{path}.target"))?;
    if !effect_fields.contains_key(&target) {
        return Err(fail(
            &format!("{path}.target"),
            format!("names a field \"{target}\" the effect table does not declare"),
        ));
    }
    let value_source = parse_trigger_value_source(
        source
            .get("source")
            .ok_or_else(|| fail(&format!("{path}.source"), "is required"))?,
        &format!("{path}.source"),
        source_fields,
    )?;
    Ok(TriggerFieldMapping {
        target,
        source: value_source,
    })
}

fn parse_trigger_effect(
    raw: &Value,
    path: &str,
    source_fields: &BTreeMap<String, CatalogFieldSpec>,
    tables: &BTreeMap<String, TableEntry>,
) -> Result<TriggerEffect, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, TRIGGER_EFFECT_KEYS, path)?;
    let table = as_text(source.get("table"), &format!("{path}.table"))?;
    let effect_entry = tables.get(&table).ok_or_else(|| {
        fail(
            &format!("{path}.table"),
            format!("names a table \"{table}\" the catalog does not declare"),
        )
    })?;

    let raw_fields = as_array(source.get("fields"), &format!("{path}.fields"))?;
    if raw_fields.is_empty() {
        return Err(fail(&format!("{path}.fields"), "must name at least one field"));
    }
    let mut fields = Vec::with_capacity(raw_fields.len());
    for (at, raw) in raw_fields.iter().enumerate() {
        fields.push(parse_trigger_field_mapping(
            raw,
            &format!("{path}.fields[{at}]"),
            source_fields,
            &effect_entry.fields,
        )?);
    }
    Ok(TriggerEffect { table, fields })
}

/// Validates `table` and `effect.table` against `tables` before returning anything, so a trigger
/// naming a table the catalog does not declare is refused here — before `Database::declare_trigger`
/// ever reaches disk (ADR-s21j/0029, same rule `declare_index`/relations already follow).
pub fn parse_trigger_entry(
    raw: &Value,
    path: &str,
    tables: &BTreeMap<String, TableEntry>,
) -> Result<TriggerEntry, HaysonSchemaError> {
    let source = as_object(Some(raw), path)?;
    reject_unknown(source, TRIGGER_KEYS, path)?;
    let name = as_name(source.get("name"), &format!("{path}.name"), "trigger")?;
    let table = as_text(source.get("table"), &format!("{path}.table"))?;
    let source_entry = tables.get(&table).ok_or_else(|| {
        fail(
            &format!("{path}.table"),
            format!("names a table \"{table}\" the catalog does not declare"),
        )
    })?;
    let event = match source.get("event").and_then(Value::as_str) {
        Some("insert") => TriggerEvent::Insert,
        Some("update") => TriggerEvent::Update,
        Some("delete") => TriggerEvent::Delete,
        Some(other) => {
            return Err(fail(
                &format!("{path}.event"),
                format!("must be one of \"insert\", \"update\", \"delete\", got {other:?}"),
            ))
        }
        None => return Err(fail(&format!("{path}.event"), "is required")),
    };
    let effect = parse_trigger_effect(
        source
            .get("effect")
            .ok_or_else(|| fail(&format!("{path}.effect"), "is required"))?,
        &format!("{path}.effect"),
        &source_entry.fields,
        tables,
    )?;
    Ok(TriggerEntry {
        name,
        table,
        event,
        effect,
    })
}

/// Format 1 is not opened by this engine at all, so every refusal that recognises it has to name
/// the one way forward: the offline rewrite in `hayson-cli`.
const FORMAT_1_MIGRATION_HINT: &str =
    "format 1 is no longer opened — rewrite the directory once with \"hayson-cli migrate-format1 <dir>\", on a copy you can go back to, and open it again";

fn parse_format_version(raw: Option<&Value>) -> Result<u64, HaysonSchemaError> {
    let value = as_whole_number(raw, "formatVersion", 1)?;
    if value == CATALOG_FORMAT_VERSION {
        return Ok(value);
    }
    if value < CATALOG_FORMAT_VERSION {
        return Err(fail(
            "formatVersion",
            format!(
                "is {value}, which describes the manifest of a database that predates the catalog — migrate it to format {CATALOG_FORMAT_VERSION}; {FORMAT_1_MIGRATION_HINT}"
            ),
        ));
    }
    Err(fail(
        "formatVersion",
        format!("is {value}, and this build of hayson reads format {CATALOG_FORMAT_VERSION} — upgrade @theritual/hayson to open it"),
    ))
}

/// Validates a parsed `hayson.json` of a format-2 database and returns it typed, without opening
/// the database. Every message names the path it failed on, such as `tables.users.indexes[1].fields`.
pub fn parse_catalog(input: &Value) -> Result<Catalog, HaysonSchemaError> {
    let source = as_object(Some(input), "catalog")?;
    let marker = source.get("catalogVersion").or_else(|| source.get("hayson"));
    if marker != Some(&Value::from(CATALOG_VERSION)) {
        let hint = if marker.and_then(Value::as_u64) == Some(1) {
            format!("; {FORMAT_1_MIGRATION_HINT}")
        } else {
            String::new()
        };
        return Err(fail(
            "catalogVersion",
            format!(
                "must be {CATALOG_VERSION}, got {marker:?} — {CATALOG_VERSION} is the marker of a metadata catalog, 1 marks a manifest of the layout that predates it; a catalog written before 5.0.0 carries the same marker under the old key \"hayson\", which is read as well{hint}"
            ),
        ));
    }
    let format_version = parse_format_version(source.get("formatVersion"))?;
    reject_unknown(source, CATALOG_KEYS, "catalog")?;

    let created_at = as_whole_number(source.get("createdAt"), "createdAt", 0)?;
    let updated_at = match source.get("updatedAt") {
        None => created_at,
        Some(_) => as_whole_number(source.get("updatedAt"), "updatedAt", 0)?,
    };

    let raw_tables = as_object(source.get("tables"), "tables")?;
    let mut tables = BTreeMap::new();
    for (name, raw) in raw_tables {
        tables.insert(name.clone(), parse_table_entry(name, raw)?);
    }

    let engine = optional_text(source.get("engine"), "engine")?;
    let layout = match source.get("layout").and_then(Value::as_str) {
        None => None,
        Some("flat") => Some(CatalogLayout::Flat),
        Some("nested") => Some(CatalogLayout::Nested),
        Some(_) => return Err(fail("layout", "must be one of \"flat\", \"nested\"")),
    };
    let defaults = parse_options(source.get("defaults"), "defaults")?;

    let relations = match source.get("relations") {
        None => None,
        Some(raw) => {
            let raw_relations = as_array(Some(raw), "relations")?;
            let mut relations = Vec::with_capacity(raw_relations.len());
            for (at, raw) in raw_relations.iter().enumerate() {
                relations.push(parse_relation(raw, &format!("relations[{at}]"), &tables)?);
            }
            for at in 0..relations.len() {
                if relations[..at].iter().any(|other| other.name == relations[at].name) {
                    return Err(fail(
                        &format!("relations[{at}].name"),
                        format!("repeats the relation name \"{}\"", relations[at].name),
                    ));
                }
            }
            Some(relations)
        }
    };

    let views = match source.get("views") {
        None => None,
        Some(raw) => {
            let raw_views = as_array(Some(raw), "views")?;
            let mut views = Vec::with_capacity(raw_views.len());
            for (at, raw) in raw_views.iter().enumerate() {
                views.push(parse_view_entry(raw, &format!("views[{at}]"))?);
            }
            for at in 0..views.len() {
                if views[..at].iter().any(|other| other.name == views[at].name) {
                    return Err(fail(
                        &format!("views[{at}].name"),
                        format!("repeats the view name \"{}\"", views[at].name),
                    ));
                }
            }
            Some(views)
        }
    };

    let triggers = match source.get("triggers") {
        None => None,
        Some(raw) => {
            let raw_triggers = as_array(Some(raw), "triggers")?;
            let mut triggers = Vec::with_capacity(raw_triggers.len());
            for (at, raw) in raw_triggers.iter().enumerate() {
                triggers.push(parse_trigger_entry(raw, &format!("triggers[{at}]"), &tables)?);
            }
            for at in 0..triggers.len() {
                if triggers[..at].iter().any(|other| other.name == triggers[at].name) {
                    return Err(fail(
                        &format!("triggers[{at}].name"),
                        format!("repeats the trigger name \"{}\"", triggers[at].name),
                    ));
                }
            }
            Some(triggers)
        }
    };

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

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

    use super::*;
    use crate::catalog::serialize::{catalog_document, serialize_catalog};

    fn reference_document() -> Value {
        json!({
            "catalogVersion": 2,
            "formatVersion": 2,
            "createdAt": 1786995100337_u64,
            "updatedAt": 1786995400000_u64,
            "engine": "1.10.1",
            "layout": "flat",
            "defaults": { "sync": "commit", "compaction": { "deadBytesRatio": 0.4, "minDeadBytes": 65536 } },
            "tables": {
                "orders": {
                    "revision": 1,
                    "primaryKey": "id",
                    "fields": {
                        "id": { "type": "string" },
                        "userId": { "type": "string", "index": "exact" }
                    }
                },
                "users": {
                    "revision": 2,
                    "description": "konta w panelu",
                    "primaryKey": "id",
                    "createdAt": 1786995100337_u64,
                    "updatedAt": 1786995400000_u64,
                    "validation": "strict",
                    "fields": {
                        "id": { "type": "string" },
                        "email": { "type": "string", "unique": true, "description": "login" },
                        "role": { "type": "enum", "values": ["admin", "user"], "index": "exact" }
                    },
                    "indexes": [
                        { "name": "users_role_email", "fields": ["role", "email"], "kind": "exact", "unique": true }
                    ],
                    "checks": [{ "name": "role_present", "expr": "role != ''" }],
                    "history": [{ "revision": 2, "at": 1786995400000_u64, "note": "dodane pole role" }]
                }
            },
            "relations": [
                {
                    "name": "orders_user",
                    "from": { "table": "orders", "field": "userId" },
                    "to": { "table": "users", "field": "id" },
                    "cardinality": "many-to-one",
                    "onDelete": "restrict",
                    "enforce": false
                }
            ]
        })
    }

    #[test]
    fn parses_the_format_md_reference_document() {
        let catalog = parse_catalog(&reference_document()).unwrap();
        assert_eq!(catalog.tables.len(), 2);
        assert_eq!(catalog.tables["users"].indexes.as_ref().unwrap().len(), 1);
        assert_eq!(catalog.relations.as_ref().unwrap().len(), 1);
    }

    #[test]
    fn round_trips_parse_serialize_parse_to_an_identical_catalog() {
        let once = parse_catalog(&reference_document()).unwrap();
        let reparsed = parse_catalog(&catalog_document(&once)).unwrap();
        assert_eq!(once, reparsed);
    }

    #[test]
    fn a_non_addressable_markdown_field_survives_the_catalog_it_is_written_into() {
        let mut doc = reference_document();
        doc["tables"]["users"]["fields"]["notes"] = json!({ "type": "markdown", "addressable": false });

        let once = parse_catalog(&doc).unwrap();
        let reparsed = parse_catalog(&catalog_document(&once)).unwrap();

        assert!(!once.tables["users"].fields["notes"].spec.addressable);
        assert_eq!(once, reparsed);
    }

    #[test]
    fn serializes_to_the_same_bytes_whatever_order_the_keys_arrived_in() {
        let forward = parse_catalog(&reference_document()).unwrap();

        let reversed_text = {
            let mut reordered = Map::new();
            let original = reference_document();
            let source = original.as_object().unwrap();
            for key in source.keys().rev() {
                reordered.insert(key.clone(), source[key].clone());
            }
            Value::Object(reordered)
        };
        let backward = parse_catalog(&reversed_text).unwrap();

        assert_eq!(serialize_catalog(&forward), serialize_catalog(&backward));
    }

    #[test]
    fn rejects_markdown_and_image_in_a_sorted_index_of_any_width() {
        for field_type in ["markdown", "image"] {
            for fields in [json!(["body"]), json!(["body", "email"])] {
                let mut doc = reference_document();
                doc["tables"]["users"]["fields"]["body"] = json!({ "type": field_type });
                doc["tables"]["users"]["indexes"] =
                    json!([{ "name": "users_body", "fields": fields, "kind": "sorted" }]);
                let failure = parse_catalog(&doc).unwrap_err();
                assert!(
                    failure
                        .0
                        .contains(&format!("of type {field_type}, which cannot be sorted")),
                    "{}",
                    failure.0
                );
            }
        }
    }

    /// A format-1 manifest carries marker 1 under the old key and no `catalogVersion` at all.
    #[test]
    fn rejects_a_catalog_with_marker_1() {
        let mut doc = reference_document();
        let object = doc.as_object_mut().unwrap();
        object.remove("catalogVersion");
        object.insert("hayson".to_string(), Value::from(1));
        let failure = parse_catalog(&doc).unwrap_err();
        assert!(failure.0.contains("hayson-cli migrate-format1"), "{}", failure.0);
    }

    #[test]
    fn rejects_a_marker_that_is_not_format_1_without_naming_the_migration_tool() {
        let mut doc = reference_document();
        doc["catalogVersion"] = Value::from(7);
        let failure = parse_catalog(&doc).unwrap_err();
        assert!(!failure.0.contains("hayson-cli migrate-format1"), "{}", failure.0);
    }

    #[test]
    fn parses_a_catalog_that_carries_the_marker_under_the_current_key_alone() {
        let doc = reference_document();
        assert!(
            doc.get("hayson").is_none(),
            "the fixture is written the way 5.0.0 writes it"
        );
        assert_eq!(parse_catalog(&doc).unwrap().catalog_version, CATALOG_VERSION);
    }

    /// A catalog written before the rename has the marker under `"hayson"`; nothing migrates it, so
    /// the read side has to take it as it stands until the next write replaces the key.
    #[test]
    fn parses_a_catalog_that_still_carries_the_marker_under_the_old_key() {
        let mut doc = reference_document();
        let object = doc.as_object_mut().unwrap();
        let marker = object.remove("catalogVersion").unwrap();
        object.insert("hayson".to_string(), marker);
        assert_eq!(parse_catalog(&doc).unwrap().catalog_version, CATALOG_VERSION);
    }

    #[test]
    fn rewrites_an_old_keyed_catalog_under_the_current_key_alone() {
        let mut doc = reference_document();
        let object = doc.as_object_mut().unwrap();
        let marker = object.remove("catalogVersion").unwrap();
        object.insert("hayson".to_string(), marker);

        let rendered = catalog_document(&parse_catalog(&doc).unwrap());

        assert_eq!(rendered["catalogVersion"], Value::from(CATALOG_VERSION));
        assert!(rendered.get("hayson").is_none(), "the old key is never written again");
    }

    #[test]
    fn rejects_a_formatversion_older_than_the_catalog_and_names_the_migration_tool() {
        let mut doc = reference_document();
        doc["formatVersion"] = Value::from(1);
        let failure = parse_catalog(&doc).unwrap_err();
        assert!(failure.0.contains("hayson-cli migrate-format1"), "{}", failure.0);
    }

    #[test]
    fn rejects_a_formatversion_newer_than_this_build_reads() {
        let mut doc = reference_document();
        doc["formatVersion"] = Value::from(3);
        assert!(parse_catalog(&doc).is_err());
    }

    #[test]
    fn rejects_an_unknown_top_level_option() {
        let mut doc = reference_document();
        doc.as_object_mut()
            .unwrap()
            .insert("somethingNew".to_string(), Value::from(true));
        assert!(parse_catalog(&doc).is_err());
    }

    #[test]
    fn rejects_a_declared_index_whose_name_collides_with_field_modifier_sugar_under_different_content() {
        // "email" sugars to an exact index named "users_email_exact" (docs/format.md, "Rozwinięty
        // zbiór indeksów"); declaring a *sorted* index under that same generated name is a distinct
        // identity that still collides on the name, which parseTableEntry refuses.
        let doc = json!({
            "catalogVersion": 2,
            "formatVersion": 2,
            "createdAt": 0,
            "tables": {
                "users": {
                    "revision": 1,
                    "primaryKey": "id",
                    "fields": {
                        "id": { "type": "string" },
                        "email": { "type": "string", "index": "exact" }
                    },
                    "indexes": [
                        { "name": "users_email_exact", "fields": ["email"], "kind": "sorted" }
                    ]
                }
            }
        });
        assert!(parse_catalog(&doc).is_err());
    }

    #[test]
    fn rejects_a_relation_pointing_at_a_non_unique_field() {
        let doc = json!({
            "catalogVersion": 2,
            "formatVersion": 2,
            "createdAt": 0,
            "tables": {
                "orders": {
                    "revision": 1,
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "userId": { "type": "string" } }
                },
                "users": {
                    "revision": 1,
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } }
                }
            },
            "relations": [
                {
                    "name": "orders_user",
                    "from": { "table": "orders", "field": "userId" },
                    "to": { "table": "users", "field": "name" },
                    "cardinality": "many-to-one"
                }
            ]
        });
        assert!(parse_catalog(&doc).is_err());
    }

    #[test]
    fn rejects_a_where_predicate_naming_an_unknown_field() {
        let doc = json!({
            "catalogVersion": 2,
            "formatVersion": 2,
            "createdAt": 0,
            "tables": {
                "users": {
                    "revision": 1,
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "age": { "type": "int" } },
                    "indexes": [
                        {
                            "name": "adults",
                            "fields": ["id"],
                            "kind": "exact",
                            "where": [{ "field": "missing", "op": ">=", "value": 18 }]
                        }
                    ]
                }
            }
        });
        assert!(parse_catalog(&doc).is_err());
    }

    fn users_and_orders() -> Value {
        json!({
            "catalogVersion": 2,
            "formatVersion": 2,
            "createdAt": 0,
            "tables": {
                "orders": {
                    "revision": 1,
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "userId": { "type": "string" } }
                },
                "users": {
                    "revision": 1,
                    "primaryKey": "id",
                    "fields": { "id": { "type": "string" }, "name": { "type": "string" } }
                }
            }
        })
    }

    fn with_trigger(mut doc: Value, trigger: Value) -> Value {
        doc["triggers"] = Value::Array(vec![trigger]);
        doc
    }

    fn users_insert_audit_trigger() -> Value {
        json!({
            "name": "users_insert_audit",
            "table": "users",
            "event": "insert",
            "effect": {
                "table": "orders",
                "fields": [
                    { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
                    { "target": "userId", "source": { "kind": "new", "field": "id" } }
                ]
            }
        })
    }

    #[test]
    fn parses_a_declared_trigger_and_round_trips_it() {
        let doc = with_trigger(users_and_orders(), users_insert_audit_trigger());
        let catalog = parse_catalog(&doc).unwrap();

        let triggers = catalog.triggers.as_ref().unwrap();
        assert_eq!(triggers.len(), 1);
        assert_eq!(triggers[0].name, "users_insert_audit");
        assert_eq!(triggers[0].table, "users");
        assert_eq!(triggers[0].effect.table, "orders");

        let reparsed = parse_catalog(&catalog_document(&catalog)).unwrap();
        assert_eq!(catalog, reparsed);
    }

    #[test]
    fn rejects_a_trigger_naming_a_source_table_the_catalog_does_not_declare() {
        let mut trigger = users_insert_audit_trigger();
        trigger["table"] = Value::from("ghost");
        let doc = with_trigger(users_and_orders(), trigger);
        assert!(parse_catalog(&doc).is_err());
    }

    #[test]
    fn rejects_a_trigger_whose_effect_names_a_table_the_catalog_does_not_declare() {
        let mut trigger = users_insert_audit_trigger();
        trigger["effect"]["table"] = Value::from("ghost");
        let doc = with_trigger(users_and_orders(), trigger);
        assert!(parse_catalog(&doc).is_err());
    }

    #[test]
    fn rejects_two_triggers_sharing_a_name() {
        let doc = users_and_orders();
        let mut doc = with_trigger(doc, users_insert_audit_trigger());
        doc["triggers"]
            .as_array_mut()
            .unwrap()
            .push(users_insert_audit_trigger());
        assert!(parse_catalog(&doc).is_err());
    }

    #[test]
    fn parses_a_table_declared_lazy_link_parsing_option() {
        let mut doc = users_and_orders();
        doc["tables"]["users"]["options"] = json!({ "linkParsing": "lazy" });
        let catalog = parse_catalog(&doc).unwrap();
        assert_eq!(
            catalog.tables["users"].options.as_ref().unwrap().link_parsing,
            Some(LinkParsing::Lazy)
        );
    }

    #[test]
    fn rejects_an_unsupported_link_parsing_mode() {
        let mut doc = users_and_orders();
        doc["tables"]["users"]["options"] = json!({ "linkParsing": "sometimes" });
        assert!(parse_catalog(&doc).is_err());
    }

    #[test]
    fn parses_a_table_declared_image_dir_option() {
        let mut doc = users_and_orders();
        doc["tables"]["users"]["options"] = json!({ "imageDir": "images/users" });
        let catalog = parse_catalog(&doc).unwrap();
        assert_eq!(
            catalog.tables["users"].options.as_ref().unwrap().image_dir,
            Some("images/users".to_string())
        );
    }
}