hayson-engine 0.11.0

use std::collections::{BTreeSet, HashMap};

use serde_json::Value;

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

use super::parse::{parse_catalog, parse_index_entry};
use super::serialize::catalog_document;
use super::types::{Catalog, IndexEntry, TableEntry};

const LISTED_KEYS: usize = 20;

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

fn composite_key_of(record: &HaysonRecord, fields: &[String]) -> Option<String> {
    let mut parts = Vec::with_capacity(fields.len());
    for field in fields {
        match record.get(field) {
            Some(Value::Null) | None => return None,
            Some(value) => parts.push(value_key(value)),
        }
    }
    Some(parts.join("\u{1}"))
}

/// Mirrors TS `enforcedUniqueIndexes`/ADR-0027: a `where`-scoped (partial) unique index stays
/// unchecked here, and a single-field index over the primary key is unique by construction.
fn is_checked_unique(entry: &TableEntry, index: &IndexEntry) -> bool {
    index.unique && index.where_.is_none() && !(index.fields.len() == 1 && index.fields[0] == entry.primary_key)
}

fn assert_records_hold_index(
    table_handle: &Table,
    entry: &TableEntry,
    index: &IndexEntry,
) -> Result<(), HaysonSchemaError> {
    if !is_checked_unique(entry, index) {
        return Ok(());
    }

    let spec = QueryBuilder::new().spec();
    let all = table_handle.query(&spec).map_err(|error| {
        HaysonSchemaError(format!(
            "reading table \"{}\" to check index \"{}\": {error}",
            table_handle.name(),
            index.name
        ))
    })?;

    let mut holders: HashMap<String, String> = HashMap::new();
    let mut clashing: BTreeSet<String> = BTreeSet::new();
    for key in &all.keys {
        let record = table_handle.get(key).map_err(|error| {
            HaysonSchemaError(format!(
                "reading table \"{}\" to check index \"{}\": {error}",
                table_handle.name(),
                index.name
            ))
        })?;
        let Some(record) = record else { continue };
        let Some(composite) = composite_key_of(&record, &index.fields) else {
            continue;
        };
        match holders.get(&composite) {
            None => {
                holders.insert(composite, key.clone());
            }
            Some(held) => {
                clashing.insert(held.clone());
                clashing.insert(key.clone());
            }
        }
    }

    if clashing.is_empty() {
        return Ok(());
    }
    let mut keys: Vec<String> = clashing.into_iter().collect();
    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()
    };
    Err(HaysonSchemaError(format!(
        "refusing to declare index \"{}\" on table \"{}\": index \"{}\" over ({}) is declared unique, and {} record(s) already share a value: {listed}{rest} — nothing on disk was changed; make those keys unique (or declare the index without unique) and try again",
        index.name,
        table_handle.name(),
        index.name,
        index.fields.join(", "),
        keys.len(),
    )))
}

/// `db.declareIndex()` — adds a named index to `tables.<table>.indexes[]` and, when it is unique,
/// refuses the whole operation if any live record (read through `table_handle`) already shares a
/// value on it. 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 `create_table`/`drop_field`).
pub fn declare_index(
    catalog: &Catalog,
    table: &str,
    spec: &Value,
    table_handle: &Table,
    now: u64,
) -> Result<Catalog, HaysonSchemaError> {
    let entry = catalog
        .tables
        .get(table)
        .ok_or_else(|| HaysonSchemaError(format!("table \"{table}\" does not exist in the catalog")))?;

    let declared = entry.indexes.clone().unwrap_or_default();
    let path = format!("tables.{table}.indexes[{}]", declared.len());
    let index = parse_index_entry(spec, &path, &entry.fields)?;

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

    let mut next_entry = entry.clone();
    let mut next_declared = declared;
    next_declared.push(index.clone());
    next_entry.indexes = Some(next_declared);
    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());

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

    assert_records_hold_index(table_handle, &next_entry, &index)?;

    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_catalog() -> Catalog {
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "email": { "type": "string" },
            },
        }))
        .unwrap();
        create_table(&catalog, "users", &schema, 1_000)
    }

    fn open_users_table(dir: &tempfile::TempDir) -> Table {
        let schema = parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "email": { "type": "string" },
            },
        }))
        .unwrap();
        Table::open(schema, dir.path().join("users.jsonl")).unwrap()
    }

    #[test]
    fn declares_a_unique_index_when_live_records_already_hold_it() {
        let dir = tempdir().unwrap();
        let mut table = open_users_table(&dir);
        table
            .put(&json!({ "id": "u1", "email": "a@example.com" }), false)
            .unwrap();
        table
            .put(&json!({ "id": "u2", "email": "b@example.com" }), false)
            .unwrap();

        let catalog = users_catalog();
        let spec = json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true });

        let next = declare_index(&catalog, "users", &spec, &table, 2_000).unwrap();

        let indexes = next.tables["users"].indexes.clone().unwrap();
        assert_eq!(indexes.len(), 1);
        assert_eq!(indexes[0].name, "users_email_unique");
        assert_eq!(next.updated_at, 2_000);
    }

    #[test]
    fn refuses_a_unique_index_when_live_records_already_clash_and_changes_nothing() {
        let dir = tempdir().unwrap();
        let mut table = open_users_table(&dir);
        table
            .put(&json!({ "id": "u1", "email": "same@example.com" }), false)
            .unwrap();
        table
            .put(&json!({ "id": "u2", "email": "same@example.com" }), false)
            .unwrap();

        let catalog = users_catalog();
        let spec = json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true });

        let error = declare_index(&catalog, "users", &spec, &table, 2_000).unwrap_err();
        assert!(error.0.contains("users_email_unique"));
        assert!(error.0.contains("u1"));
        assert!(error.0.contains("u2"));
        assert!(catalog.tables["users"].indexes.is_none());
    }

    #[test]
    fn a_non_unique_index_is_declared_without_scanning_for_clashes() {
        let dir = tempdir().unwrap();
        let mut table = open_users_table(&dir);
        table
            .put(&json!({ "id": "u1", "email": "same@example.com" }), false)
            .unwrap();
        table
            .put(&json!({ "id": "u2", "email": "same@example.com" }), false)
            .unwrap();

        let catalog = users_catalog();
        let spec = json!({ "name": "users_email_exact", "fields": ["email"], "kind": "exact", "unique": false });

        let next = declare_index(&catalog, "users", &spec, &table, 2_000).unwrap();
        assert_eq!(next.tables["users"].indexes.clone().unwrap().len(), 1);
    }

    #[test]
    fn refuses_when_the_table_is_not_in_the_catalog() {
        let dir = tempdir().unwrap();
        let table = open_users_table(&dir);
        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
        let spec = json!({ "name": "x", "fields": ["email"], "kind": "exact", "unique": false });

        assert!(declare_index(&catalog, "users", &spec, &table, 2_000).is_err());
    }

    #[test]
    fn refuses_a_duplicate_index_name() {
        let dir = tempdir().unwrap();
        let table = open_users_table(&dir);
        let catalog = users_catalog();
        let spec = json!({ "name": "users_email_a", "fields": ["email"], "kind": "exact", "unique": false });
        let catalog = declare_index(&catalog, "users", &spec, &table, 2_000).unwrap();

        let spec2 = json!({ "name": "users_email_a", "fields": ["email"], "kind": "exact", "unique": false });
        let error = declare_index(&catalog, "users", &spec2, &table, 3_000).unwrap_err();
        assert!(error.0.contains("already declares an index named \"users_email_a\""));
    }
}