hayson-engine 0.27.0

use std::collections::BTreeMap;

use serde_json::Value;

use crate::errors::HaysonSchemaError;
use crate::schema::TableSchema;

use super::parse::{parse_catalog, parse_trigger_entry};
use super::serialize::catalog_document;
use super::types::{
    CatalogFieldSpec, CatalogLayout, CheckEntry, IndexEntry, TableEntry, CATALOG_FORMAT_VERSION, CATALOG_VERSION,
};
use super::Catalog;

pub fn empty_catalog(layout: CatalogLayout, now: u64) -> Catalog {
    Catalog {
        catalog_version: CATALOG_VERSION,
        format_version: CATALOG_FORMAT_VERSION,
        created_at: now,
        updated_at: now,
        engine: None,
        layout: Some(layout),
        defaults: None,
        tables: BTreeMap::new(),
        relations: None,
        views: None,
        triggers: None,
    }
}

fn is_keyed_by_present_fields(index: &IndexEntry, fields: &BTreeMap<String, CatalogFieldSpec>) -> bool {
    index.fields.iter().all(|field| fields.contains_key(field))
}

fn kept_indexes(
    indexes: &Option<Vec<IndexEntry>>,
    fields: &BTreeMap<String, CatalogFieldSpec>,
) -> Option<Vec<IndexEntry>> {
    let indexes = indexes.as_ref()?;
    let kept: Vec<IndexEntry> = indexes
        .iter()
        .filter(|index| is_keyed_by_present_fields(index, fields))
        .cloned()
        .collect();
    if kept.is_empty() {
        None
    } else {
        Some(kept)
    }
}

pub fn dropped_indexes_of(previous: &TableEntry, fields: &BTreeMap<String, CatalogFieldSpec>) -> Vec<IndexEntry> {
    previous
        .indexes
        .iter()
        .flatten()
        .filter(|index| !is_keyed_by_present_fields(index, fields))
        .cloned()
        .collect()
}

fn is_word_char(byte: u8) -> bool {
    byte.is_ascii_alphanumeric() || byte == b'_'
}

/// `\b<field>\b` on ASCII bytes — good enough for field names, which schema already restricts to
/// `[A-Za-z][A-Za-z0-9_]*`.
fn mentions_word(haystack: &str, word: &str) -> bool {
    if word.is_empty() {
        return false;
    }
    let bytes = haystack.as_bytes();
    let mut search_from = 0;
    while let Some(offset) = haystack[search_from..].find(word) {
        let at = search_from + offset;
        let before_ok = at == 0 || !is_word_char(bytes[at - 1]);
        let after = at + word.len();
        let after_ok = after == bytes.len() || !is_word_char(bytes[after]);
        if before_ok && after_ok {
            return true;
        }
        search_from = at + 1;
    }
    false
}

pub fn checks_mentioning(previous: &TableEntry, field: &str) -> Vec<CheckEntry> {
    previous
        .checks
        .iter()
        .flatten()
        .filter(|check| mentions_word(&check.expr, field))
        .cloned()
        .collect()
}

/// Builds the next `TableEntry` for `createTable`/migration: field specs come from `schema`, but
/// documentation (`description`/`deprecated`), `options`, `checks` and `history` of a same-named
/// previous entry are carried over, and a declared index survives only if every field it names is
/// still present.
pub fn table_entry_of(schema: &TableSchema, previous: Option<&TableEntry>, now: u64) -> TableEntry {
    let mut fields = BTreeMap::new();
    for (name, spec) in &schema.fields {
        let previous_field = previous.and_then(|entry| entry.fields.get(name));
        fields.insert(
            name.clone(),
            CatalogFieldSpec {
                spec: spec.clone(),
                description: previous_field.and_then(|f| f.description.clone()),
                deprecated: previous_field.and_then(|f| f.deprecated.clone()),
            },
        );
    }
    let indexes = kept_indexes(&previous.and_then(|p| p.indexes.clone()), &fields);
    TableEntry {
        revision: schema.revision as u64,
        primary_key: schema.primary_key.clone(),
        fields,
        partition_by: schema.partition_by.clone(),
        description: previous.and_then(|p| p.description.clone()),
        created_at: Some(previous.and_then(|p| p.created_at).unwrap_or(now)),
        updated_at: Some(now),
        validation: previous.and_then(|p| p.validation),
        options: previous.and_then(|p| p.options.clone()),
        indexes,
        checks: previous.and_then(|p| p.checks.clone()),
        history: previous.and_then(|p| p.history.clone()),
    }
}

pub fn with_table_entry(catalog: &Catalog, name: &str, entry: TableEntry, now: u64) -> Catalog {
    let mut next = catalog.clone();
    next.updated_at = now;
    next.tables.insert(name.to_string(), entry);
    next
}

/// `db.createTable()` — addytywna zmiana albo nowa tabela, patrz `docs/schema.md` „Ewolucja schemy".
/// Migracja niekompatybilna (zmiana typu, zwężenie itd.) nie jest tu sprawdzana: `schema` już
/// przeszła przez `schema::parse_schema`, a diff rewizji zostaje na liście TODO niżej.
pub fn create_table(catalog: &Catalog, name: &str, schema: &TableSchema, now: u64) -> Catalog {
    let entry = table_entry_of(schema, catalog.tables.get(name), now);
    with_table_entry(catalog, name, entry, now)
}

/// `db.dropField(table, field)` na poziomie samego katalogu — `docs/format.md`
/// „`dropField` odmawia jednolicie na wszystkich czterech klasach zależności" i ADR-qtr2 (cerebro,
/// „0030 — force nie wyburza deklaracji"): `force` potwierdza tylko rekordy nadal niosące pole
/// w logu, nigdy deklaracje katalogu — ten test jest więc zawsze egzekwowany, bez parametru
/// `force`, bo nic tutaj go nie omija.
///
/// TODO: sprawdzenie rezydualnych rekordów (`HaysonFieldInUseError`, `{ force: true }`) potrzebuje
/// żywej tabeli/logu i należy do `database::drop_field`, gdy ta warstwa powstanie — patrz ADR-qtr2.
pub fn drop_field(
    catalog: &Catalog,
    table: &str,
    field: &str,
    now: u64,
) -> Result<(Catalog, TableEntry), HaysonSchemaError> {
    let entry = catalog
        .tables
        .get(table)
        .ok_or_else(|| HaysonSchemaError(format!("table \"{table}\" does not exist in the catalog")))?;
    if !entry.fields.contains_key(field) {
        return Err(HaysonSchemaError(format!(
            "field \"{field}\" is not declared in the schema of table \"{table}\""
        )));
    }
    if entry.primary_key == field {
        return Err(HaysonSchemaError(format!(
            "field \"{field}\" is the primary key of table \"{table}\" and cannot be dropped — rewriting the table is the only way out"
        )));
    }
    if entry.partition_by.as_deref() == Some(field) {
        return Err(HaysonSchemaError(format!(
            "field \"{field}\" is what table \"{table}\" is partitioned by and cannot be dropped — every record would lose the file it belongs to"
        )));
    }

    let mut remaining_fields = entry.fields.clone();
    remaining_fields.remove(field);

    let keyed_by = dropped_indexes_of(entry, &remaining_fields);
    if !keyed_by.is_empty() {
        let named = keyed_by
            .iter()
            .map(|index| format!("\"{}\"", index.name))
            .collect::<Vec<_>>()
            .join(", ");
        return Err(HaysonSchemaError(format!(
            "field \"{field}\" is used by index(es) {named} of table \"{table}\" and dropping it would silently discard them — drop the index(es) first"
        )));
    }
    let mentioned_in = checks_mentioning(entry, field);
    if !mentioned_in.is_empty() {
        let named = mentioned_in
            .iter()
            .map(|check| format!("\"{}\"", check.name))
            .collect::<Vec<_>>()
            .join(", ");
        return Err(HaysonSchemaError(format!(
            "check(s) {named} of table \"{table}\" mention \"{field}\" — dropping \"{field}\" may break the check; this is a text-only match on the expression, not semantic analysis, so if the check does not actually depend on \"{field}\", drop it and recreate it without the mention"
        )));
    }

    let mut next_entry = entry.clone();
    next_entry.fields = remaining_fields;
    next_entry.updated_at = Some(now);

    let mut next_catalog = catalog.clone();
    next_catalog.updated_at = now;
    next_catalog.tables.insert(table.to_string(), next_entry.clone());

    // NOTE: catches indexes[].where[]/relations[] pointing at the dropped field — the checks above
    // only cover indexes[].fields and checks[].expr (docs/format.md, the four dependency classes).
    parse_catalog(&catalog_document(&next_catalog)).map_err(|error| {
        HaysonSchemaError(format!(
            "dropping field \"{field}\" from table \"{table}\" would leave a catalog this build of hayson cannot read back — {}; nothing was changed",
            error.0
        ))
    })?;

    Ok((next_catalog, next_entry))
}

/// `db.declareTrigger()` — adds a declarative AFTER trigger to `triggers[]`. `trigger_spec.table`
/// and `trigger_spec.effect.table` must already be declared tables of `catalog` (checked by
/// `parse_trigger_entry` before anything else runs, ADR-s21j/0029): a spec naming an undeclared
/// table is refused here, before `Database::declare_trigger` ever reaches disk. Nothing is written
/// when it refuses — the catalog handed back on success is the only side effect, `write_catalog` is
/// left to the caller (mirrors `declare_index`).
pub fn declare_trigger(catalog: &Catalog, trigger_spec: &Value, now: u64) -> Result<Catalog, HaysonSchemaError> {
    let declared = catalog.triggers.clone().unwrap_or_default();
    let path = format!("triggers[{}]", declared.len());
    let trigger = parse_trigger_entry(trigger_spec, &path, &catalog.tables)?;

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

    let mut next_declared = declared;
    next_declared.push(trigger.clone());

    let mut next_catalog = catalog.clone();
    next_catalog.updated_at = now;
    next_catalog.triggers = Some(next_declared);

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

    Ok(next_catalog)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::catalog::types::CatalogIndexKind;
    use crate::schema::parse_schema;
    use serde_json::{json, Value};

    fn users_schema(fields_extra: Value) -> TableSchema {
        let mut fields = json!({ "id": { "type": "string" } });
        for (key, value) in fields_extra.as_object().unwrap() {
            fields[key] = value.clone();
        }
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": fields,
        }))
        .unwrap()
    }

    #[test]
    fn create_table_adds_a_fresh_table_to_an_empty_catalog() {
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let schema = users_schema(json!({ "name": { "type": "string" } }));

        let next = create_table(&catalog, "users", &schema, 2_000);

        assert!(next.tables.contains_key("users"));
        assert_eq!(next.updated_at, 2_000);
        assert_eq!(next.tables["users"].created_at, Some(2_000));
    }

    #[test]
    fn create_table_keeps_documentation_and_created_at_of_a_previous_entry() {
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let schema = users_schema(json!({ "name": { "type": "string" } }));
        let first = create_table(&catalog, "users", &schema, 1_000);

        let mut with_docs = first.clone();
        with_docs
            .tables
            .get_mut("users")
            .unwrap()
            .fields
            .get_mut("name")
            .unwrap()
            .description = Some("display name".to_string());

        let migrated_schema = users_schema(json!({
            "name": { "type": "string" },
            "age": { "type": "int" },
        }));
        let second = create_table(&with_docs, "users", &migrated_schema, 3_000);

        assert_eq!(second.tables["users"].created_at, Some(1_000));
        assert_eq!(second.tables["users"].updated_at, Some(3_000));
        assert_eq!(
            second.tables["users"].fields["name"].description,
            Some("display name".to_string())
        );
    }

    #[test]
    fn drop_field_refuses_when_a_named_index_still_uses_the_field() {
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let schema = users_schema(json!({ "email": { "type": "string" } }));
        let mut catalog = create_table(&catalog, "users", &schema, 1_000);
        catalog.tables.get_mut("users").unwrap().indexes = Some(vec![IndexEntry {
            name: "users_email_custom".to_string(),
            fields: vec!["email".to_string()],
            kind: CatalogIndexKind::Exact,
            unique: false,
            where_: None,
            expire_after_seconds: None,
        }]);

        let result = drop_field(&catalog, "users", "email", 2_000);
        assert!(result.is_err());
        assert!(result.unwrap_err().0.contains("users_email_custom"));
    }

    #[test]
    fn drop_field_refuses_when_a_check_mentions_the_field_by_name() {
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let schema = users_schema(json!({ "age": { "type": "int" } }));
        let mut catalog = create_table(&catalog, "users", &schema, 1_000);
        catalog.tables.get_mut("users").unwrap().checks = Some(vec![CheckEntry {
            name: "adult".to_string(),
            expr: "age >= 18".to_string(),
        }]);

        let result = drop_field(&catalog, "users", "age", 2_000);
        assert!(result.is_err());
        assert!(result.unwrap_err().0.contains("adult"));
    }

    #[test]
    fn drop_field_succeeds_when_nothing_depends_on_the_field() {
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let schema = users_schema(json!({ "nickname": { "type": "string" } }));
        let catalog = create_table(&catalog, "users", &schema, 1_000);

        let (next, entry) = drop_field(&catalog, "users", "nickname", 2_000).unwrap();
        assert!(!entry.fields.contains_key("nickname"));
        assert!(!next.tables["users"].fields.contains_key("nickname"));
        assert_eq!(next.tables["users"].updated_at, Some(2_000));
    }

    #[test]
    fn drop_field_refuses_the_primary_key() {
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let schema = users_schema(json!({ "name": { "type": "string" } }));
        let catalog = create_table(&catalog, "users", &schema, 1_000);

        assert!(drop_field(&catalog, "users", "id", 2_000).is_err());
    }
}