hayson-engine 0.27.0

use std::borrow::Cow;
use std::collections::BTreeMap;

use serde_json::{Number, Value};

use crate::errors::{HaysonValidationError, ValidationIssue};

use super::content_hash::{is_physical_path, physical_path_of};
use super::timestamp::parse_iso_timestamp;
use super::types::{DefaultValue, FieldSpec, FieldType, TableSchema, META_PREFIX};

fn virtual_path_of(source: &serde_json::Map<String, Value>) -> Result<String, String> {
    match source.get("virtualPath") {
        Some(Value::String(s)) if s.starts_with('/') && s.len() > 1 => Ok(s.clone()),
        _ => Err("expected \"virtualPath\" to be an absolute path string".into()),
    }
}

/// A markdown field value. `addressable` is the field's own declaration: an addressable field
/// must carry a `virtualPath`, a non-addressable one may leave it out — and keeps it, still shape
/// checked, when it is there, so flipping an existing field to `addressable: false` does not make
/// the records already on disk unreadable.
fn coerce_markdown(value: &Value, addressable: bool) -> Result<Value, String> {
    let source = value
        .as_object()
        .ok_or("expected an object with \"content\" and \"virtualPath\"")?;
    for key in source.keys() {
        if key != "content" && key != "virtualPath" {
            return Err(format!("has an unknown key \"{key}\""));
        }
    }
    let content = match source.get("content") {
        Some(Value::String(s)) => s.clone(),
        _ => return Err("expected \"content\" to be a string".into()),
    };
    let mut coerced = serde_json::Map::new();
    coerced.insert("content".into(), Value::String(content));
    if addressable || source.contains_key("virtualPath") {
        coerced.insert("virtualPath".into(), Value::String(virtual_path_of(source)?));
    }
    Ok(Value::Object(coerced))
}

fn coerce_image(value: &Value, image_key: Option<&[u8; 32]>) -> Result<Value, String> {
    let source = value
        .as_object()
        .ok_or("expected an object with \"content\" and \"virtualPath\"")?;
    // NOTE: also accepts this field's own coerced shape {physicalPath, virtualPath} unchanged, so
    // re-validating an untouched image field (e.g. during patch's merge-then-revalidate) round-trips.
    if source.contains_key("physicalPath") {
        for key in source.keys() {
            if key != "physicalPath" && key != "virtualPath" {
                return Err(format!("has an unknown key \"{key}\""));
            }
        }
        let physical_path = match source.get("physicalPath") {
            Some(Value::String(s)) if is_physical_path(s) => s.clone(),
            Some(Value::String(_)) => {
                return Err(
                    "expected \"physicalPath\" to be a content-hash file name this engine wrote (\"<16 hex digits>.bin\") — pass the image bytes as \"content\" instead of naming a file"
                        .into(),
                )
            }
            _ => return Err("expected \"physicalPath\" to be a string".into()),
        };
        let virtual_path = virtual_path_of(source)?;
        let mut coerced = serde_json::Map::new();
        coerced.insert("physicalPath".into(), Value::String(physical_path));
        coerced.insert("virtualPath".into(), Value::String(virtual_path));
        return Ok(Value::Object(coerced));
    }
    for key in source.keys() {
        if key != "content" && key != "virtualPath" {
            return Err(format!("has an unknown key \"{key}\""));
        }
    }
    let content = match source.get("content") {
        Some(Value::String(s)) => s.clone(),
        _ => return Err("expected \"content\" to be a string".into()),
    };
    let virtual_path = virtual_path_of(source)?;
    let decoded = super::content_hash::decode_image_content(&content)?;
    let physical_path = match image_key {
        Some(key) => crate::crypto::keyed_physical_path(key, &decoded),
        None => physical_path_of(&decoded),
    };
    let mut coerced = serde_json::Map::new();
    coerced.insert("physicalPath".into(), Value::String(physical_path));
    coerced.insert("virtualPath".into(), Value::String(virtual_path));
    Ok(Value::Object(coerced))
}

pub type HaysonRecord = BTreeMap<String, Value>;

#[derive(Debug, Clone, Copy, Default)]
pub struct ValidateOptions {
    pub partial: bool,
    pub now: Option<i64>,
    /// Set for a table of an encrypted database: image file names are keyed with it instead of being a
    /// plain content hash.
    pub image_key: Option<[u8; 32]>,
}

fn is_json_safe(value: &Value, depth: u32) -> bool {
    if depth > 32 {
        return false;
    }
    match value {
        Value::Null | Value::String(_) | Value::Bool(_) => true,
        Value::Number(n) => n.as_f64().is_some_and(f64::is_finite),
        Value::Array(items) => items.iter().all(|item| is_json_safe(item, depth + 1)),
        Value::Object(map) => map.values().all(|item| is_json_safe(item, depth + 1)),
    }
}

fn coerce_scalar(
    field_type: FieldType,
    values: Option<&[String]>,
    value: &Value,
    image_key: Option<&[u8; 32]>,
) -> Result<Value, String> {
    match field_type {
        FieldType::String => match value {
            Value::String(_) => Ok(value.clone()),
            _ => Err("expected a string".into()),
        },
        FieldType::Int => match value {
            Value::Number(n) if n.as_f64().is_some_and(|f| f.is_finite() && f.fract() == 0.0) => Ok(value.clone()),
            _ => Err("expected an integer".into()),
        },
        FieldType::Float => match value.as_f64() {
            Some(n) if n.is_finite() => Ok(value.clone()),
            _ => Err("expected a finite number".into()),
        },
        FieldType::Bool => match value {
            Value::Bool(_) => Ok(value.clone()),
            _ => Err("expected a boolean".into()),
        },
        FieldType::Timestamp => match value {
            Value::Number(n) if n.as_f64().is_some_and(f64::is_finite) => {
                Ok(Value::Number(Number::from(n.as_f64().unwrap().trunc() as i64)))
            }
            Value::String(s) => match parse_iso_timestamp(s) {
                Some(ms) => Ok(Value::Number(Number::from(ms))),
                None => Err("expected a valid date".into()),
            },
            _ => Err("expected epoch milliseconds or an ISO date string".into()),
        },
        FieldType::Enum => match value {
            Value::String(s) if values.is_some_and(|v| v.iter().any(|item| item == s)) => Ok(value.clone()),
            _ => Err(format!(
                "expected one of {}",
                values.map(|v| v.join(", ")).unwrap_or_default()
            )),
        },
        FieldType::Json => {
            if is_json_safe(value, 0) {
                Ok(value.clone())
            } else {
                Err("expected JSON data".into())
            }
        }
        FieldType::Image => coerce_image(value, image_key),
        _ => Err(format!("unsupported type {}", field_type.as_str())),
    }
}

fn coerce_field(spec: &FieldSpec, value: &Value, image_key: Option<&[u8; 32]>) -> Result<Value, String> {
    if value.is_null() {
        return if spec.required {
            Err("is required and must not be null".into())
        } else {
            Ok(Value::Null)
        };
    }
    // NOTE: markdown is coerced here, not in `coerce_scalar` — only the spec knows `addressable`.
    if spec.field_type == FieldType::Markdown {
        return coerce_markdown(value, spec.addressable);
    }
    if !spec.field_type.is_array() {
        return coerce_scalar(spec.field_type, spec.values.as_deref(), value, image_key);
    }

    let items = value.as_array().ok_or_else(|| {
        format!(
            "expected an array of {}",
            spec.field_type.element_type().unwrap().as_str()
        )
    })?;
    let element = spec.field_type.element_type().expect("array type has an element");
    let mut coerced = Vec::with_capacity(items.len());
    for (position, item) in items.iter().enumerate() {
        let value =
            coerce_scalar(element, None, item, image_key).map_err(|reason| format!("item {position} {reason}"))?;
        coerced.push(value);
    }
    Ok(Value::Array(coerced))
}

fn resolve_default(spec: &FieldSpec, now: i64) -> Value {
    match &spec.default {
        Some(DefaultValue::Now) => Value::Number(Number::from(now)),
        Some(DefaultValue::Value(value)) => value.clone(),
        None => Value::Null,
    }
}

pub fn validate_record(
    schema: &TableSchema,
    input: &Value,
    options: ValidateOptions,
) -> Result<HaysonRecord, HaysonValidationError> {
    let Some(source) = input.as_object() else {
        return Err(HaysonValidationError {
            table: schema.table.clone(),
            issues: vec![ValidationIssue {
                field: "(record)".into(),
                reason: "expected an object".into(),
            }],
        });
    };

    let mut issues = Vec::new();
    for name in source.keys() {
        if name.starts_with(META_PREFIX) {
            issues.push(ValidationIssue {
                field: name.clone(),
                reason: "uses the reserved \"_\" prefix".into(),
            });
        } else if !schema.fields.contains_key(name) {
            issues.push(ValidationIssue {
                field: name.clone(),
                reason: "is not declared in the schema".into(),
            });
        }
    }

    let now = options.now.unwrap_or_else(|| chrono::Utc::now().timestamp_millis());
    let mut record = HaysonRecord::new();
    for (name, spec) in &schema.fields {
        match source.get(name) {
            None => {
                if options.partial {
                    continue;
                }
                if spec.default.is_some() {
                    record.insert(name.clone(), resolve_default(spec, now));
                } else if spec.required {
                    issues.push(ValidationIssue {
                        field: name.clone(),
                        reason: "is required".into(),
                    });
                }
            }
            Some(value) => match coerce_field(spec, value, options.image_key.as_ref()) {
                Ok(coerced) => {
                    record.insert(name.clone(), coerced);
                }
                Err(reason) => issues.push(ValidationIssue {
                    field: name.clone(),
                    reason,
                }),
            },
        }
    }

    if !issues.is_empty() {
        return Err(HaysonValidationError {
            table: schema.table.clone(),
            issues,
        });
    }
    Ok(record)
}

/// Whether `name` is a field of `schema` or a reserved `_` key. A reserved key is kept here on
/// purpose: `validate_record` refuses it whatever the table's `validation` says, so it is never
/// one of the fields a `warn` table drops.
fn is_writable_field(schema: &TableSchema, name: &str) -> bool {
    name.starts_with(META_PREFIX) || schema.fields.contains_key(name)
}

/// The fields `input` names that `schema` does not declare — what a `validation: "warn"` table
/// drops from a write and what a `strict` one refuses it for.
pub fn undeclared_fields(schema: &TableSchema, input: &Value) -> Vec<String> {
    let Some(source) = input.as_object() else {
        return Vec::new();
    };
    source
        .keys()
        .filter(|name| !is_writable_field(schema, name))
        .cloned()
        .collect()
}

/// `input` without the fields [`undeclared_fields`] names — the whole of what `validation: "warn"`
/// does to a write, which is then validated exactly as a `strict` one. Borrowed unchanged when
/// there is nothing to drop, so a table nobody writes an undeclared field to copies no document.
pub fn without_undeclared<'a>(schema: &TableSchema, input: &'a Value) -> Cow<'a, Value> {
    let Some(source) = input.as_object() else {
        return Cow::Borrowed(input);
    };
    if source.keys().all(|name| is_writable_field(schema, name)) {
        return Cow::Borrowed(input);
    }
    let kept: serde_json::Map<String, Value> = source
        .iter()
        .filter(|(name, _)| is_writable_field(schema, name))
        .map(|(name, value)| (name.clone(), value.clone()))
        .collect();
    Cow::Owned(Value::Object(kept))
}

#[allow(dead_code)]
pub fn project_to_schema(schema: &TableSchema, record: &HaysonRecord) -> HaysonRecord {
    record
        .iter()
        .filter(|(name, _)| schema.fields.contains_key(*name))
        .map(|(name, value)| (name.clone(), value.clone()))
        .collect()
}

pub fn primary_key_of(schema: &TableSchema, record: &HaysonRecord) -> Result<String, HaysonValidationError> {
    match record.get(&schema.primary_key) {
        Some(Value::String(s)) => Ok(s.clone()),
        Some(Value::Number(n)) => Ok(n.to_string()),
        _ => Err(HaysonValidationError {
            table: schema.table.clone(),
            issues: vec![ValidationIssue {
                field: schema.primary_key.clone(),
                reason: "is required as the primary key".into(),
            }],
        }),
    }
}

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

    use super::*;
    use crate::schema::parse::parse_schema;

    fn users_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string", "required": true },
                "age": { "type": "int" },
                "createdAt": { "type": "timestamp", "default": "now" },
                "role": { "type": "enum", "values": ["admin", "user"] },
                "tags": { "type": "string[]" },
            },
        }))
        .unwrap()
    }

    #[test]
    fn fills_in_defaults_and_the_now_sentinel() {
        let schema = users_schema();
        let record = validate_record(
            &schema,
            &json!({ "id": "u1", "name": "Ania" }),
            ValidateOptions {
                partial: false,
                now: Some(1000),
                ..ValidateOptions::default()
            },
        )
        .unwrap();
        assert_eq!(record.get("createdAt"), Some(&Value::from(1000)));
    }

    #[test]
    fn rejects_a_required_field_missing_without_a_default() {
        let schema = users_schema();
        let err = validate_record(&schema, &json!({ "id": "u1" }), ValidateOptions::default()).unwrap_err();
        assert!(err.issues.iter().any(|i| i.field == "name"));
    }

    #[test]
    fn rejects_a_field_not_declared_in_the_schema() {
        let schema = users_schema();
        let err = validate_record(
            &schema,
            &json!({ "id": "u1", "name": "Ania", "unknown": 1 }),
            ValidateOptions::default(),
        )
        .unwrap_err();
        assert!(err.issues.iter().any(|i| i.field == "unknown"));
    }

    #[test]
    fn rejects_a_field_using_the_reserved_meta_prefix() {
        let schema = users_schema();
        let err = validate_record(
            &schema,
            &json!({ "id": "u1", "name": "Ania", "_seq": 1 }),
            ValidateOptions::default(),
        )
        .unwrap_err();
        assert!(err.issues.iter().any(|i| i.field == "_seq"));
    }

    #[test]
    fn rejects_a_non_integer_for_an_int_field() {
        let schema = users_schema();
        let err = validate_record(
            &schema,
            &json!({ "id": "u1", "name": "Ania", "age": 1.5 }),
            ValidateOptions::default(),
        )
        .unwrap_err();
        assert!(err.issues.iter().any(|i| i.field == "age"));
    }

    #[test]
    fn accepts_an_iso_timestamp_and_converts_it_to_epoch_millis() {
        let schema = users_schema();
        let record = validate_record(
            &schema,
            &json!({ "id": "u1", "name": "Ania", "createdAt": "2025-01-01T00:00:00Z" }),
            ValidateOptions::default(),
        )
        .unwrap();
        assert_eq!(record.get("createdAt"), Some(&Value::from(1735689600000i64)));
    }

    #[test]
    fn rejects_an_enum_value_outside_its_declared_values() {
        let schema = users_schema();
        let err = validate_record(
            &schema,
            &json!({ "id": "u1", "name": "Ania", "role": "superadmin" }),
            ValidateOptions::default(),
        )
        .unwrap_err();
        assert!(err.issues.iter().any(|i| i.field == "role"));
    }

    #[test]
    fn accepts_a_typed_array_field() {
        let schema = users_schema();
        let record = validate_record(
            &schema,
            &json!({ "id": "u1", "name": "Ania", "tags": ["a", "b"] }),
            ValidateOptions::default(),
        )
        .unwrap();
        assert_eq!(record.get("tags"), Some(&json!(["a", "b"])));
    }

    #[test]
    fn partial_mode_skips_absent_fields_instead_of_defaulting_or_requiring() {
        let schema = users_schema();
        let record = validate_record(
            &schema,
            &json!({ "age": 5 }),
            ValidateOptions {
                partial: true,
                now: Some(1),
                ..ValidateOptions::default()
            },
        )
        .unwrap();
        assert!(!record.contains_key("name"));
        assert!(!record.contains_key("createdAt"));
        assert_eq!(record.get("age"), Some(&Value::from(5)));
    }

    #[test]
    fn projects_out_fields_the_schema_no_longer_declares() {
        let schema = users_schema();
        let mut record = HaysonRecord::new();
        record.insert("id".into(), Value::String("u1".into()));
        record.insert("dropped".into(), Value::from(1));
        let projected = project_to_schema(&schema, &record);
        assert!(!projected.contains_key("dropped"));
        assert!(projected.contains_key("id"));
    }

    #[test]
    fn undeclared_fields_names_the_stray_keys_and_leaves_a_reserved_one_alone() {
        let schema = users_schema();
        let input = json!({ "id": "u1", "nickname": "Anka", "_seq": 7, "shoeSize": 39 });

        assert_eq!(
            undeclared_fields(&schema, &input),
            vec!["nickname".to_string(), "shoeSize".to_string()]
        );
    }

    #[test]
    fn without_undeclared_keeps_a_clean_document_borrowed_and_copies_only_a_dirty_one() {
        let schema = users_schema();
        let clean = json!({ "id": "u1", "name": "Ania" });
        let dirty = json!({ "id": "u1", "name": "Ania", "nickname": "Anka", "_seq": 7 });

        assert!(matches!(without_undeclared(&schema, &clean), Cow::Borrowed(_)));
        let kept = without_undeclared(&schema, &dirty);
        assert!(matches!(kept, Cow::Owned(_)));
        assert_eq!(kept.get("name"), Some(&json!("Ania")));
        assert_eq!(kept.get("nickname"), None);
        assert_eq!(kept.get("_seq"), Some(&json!(7)));
    }

    #[test]
    fn a_document_stripped_of_its_stray_key_validates_while_the_original_does_not() {
        let schema = users_schema();
        let dirty = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });

        assert!(validate_record(&schema, &dirty, ValidateOptions::default()).is_err());
        let record = validate_record(
            &schema,
            without_undeclared(&schema, &dirty).as_ref(),
            ValidateOptions::default(),
        )
        .unwrap();
        assert_eq!(record.get("name"), Some(&json!("Ania")));
        assert!(!record.contains_key("nickname"));
    }

    fn docs_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
                "cover": { "type": "image" },
            },
        }))
        .unwrap()
    }

    #[test]
    fn coerces_a_markdown_field_into_content_and_virtual_path() {
        let schema = docs_schema();
        let record = validate_record(
            &schema,
            &json!({
                "id": "d1",
                "page": { "content": "# Hi", "virtualPath": "/docs/hi.md" },
            }),
            ValidateOptions::default(),
        )
        .unwrap();
        assert_eq!(
            record.get("page"),
            Some(&json!({ "content": "# Hi", "virtualPath": "/docs/hi.md" }))
        );
    }

    #[test]
    fn rejects_a_markdown_value_missing_its_virtual_path() {
        let schema = docs_schema();
        let err = validate_record(
            &schema,
            &json!({ "id": "d1", "page": { "content": "# Hi" } }),
            ValidateOptions::default(),
        )
        .unwrap_err();
        assert!(err.issues.iter().any(|i| i.field == "page"));
    }

    fn notes_schema() -> TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "notes",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "body": { "type": "markdown", "addressable": false },
            },
        }))
        .unwrap()
    }

    #[test]
    fn a_non_addressable_markdown_field_validates_on_content_alone() {
        let record = validate_record(
            &notes_schema(),
            &json!({ "id": "n1", "body": { "content": "[a](/docs/a.md)" } }),
            ValidateOptions::default(),
        )
        .unwrap();

        assert_eq!(record.get("body"), Some(&json!({ "content": "[a](/docs/a.md)" })));
    }

    #[test]
    fn a_non_addressable_markdown_field_still_shape_checks_a_path_it_was_handed() {
        let kept = validate_record(
            &notes_schema(),
            &json!({ "id": "n1", "body": { "content": "x", "virtualPath": "/notes/n1.md" } }),
            ValidateOptions::default(),
        )
        .unwrap();
        let refused = validate_record(
            &notes_schema(),
            &json!({ "id": "n1", "body": { "content": "x", "virtualPath": "notes/n1.md" } }),
            ValidateOptions::default(),
        )
        .unwrap_err();

        assert_eq!(
            kept.get("body"),
            Some(&json!({ "content": "x", "virtualPath": "/notes/n1.md" })),
            "a record written before the field opted out has to keep reading back"
        );
        assert!(refused.issues.iter().any(|issue| issue.field == "body"));
    }

    #[test]
    fn coerces_an_image_field_into_a_content_addressed_physical_path() {
        let schema = docs_schema();
        let record = validate_record(
            &schema,
            &json!({
                "id": "d1",
                "cover": { "content": "Ynl0ZXMtb2YtdGhlLWxvZ28=", "virtualPath": "/images/logo.png" },
            }),
            ValidateOptions::default(),
        )
        .unwrap();
        let cover = record.get("cover").unwrap().as_object().unwrap();
        assert_eq!(cover.get("virtualPath"), Some(&json!("/images/logo.png")));
        assert!(cover.get("physicalPath").unwrap().as_str().unwrap().ends_with(".bin"));
        assert!(!cover.contains_key("content"));
    }

    #[test]
    fn rejects_an_image_content_that_is_not_valid_base64() {
        let schema = docs_schema();
        let err = validate_record(
            &schema,
            &json!({ "id": "d1", "cover": { "content": "not base64!!", "virtualPath": "/images/a.png" } }),
            ValidateOptions::default(),
        )
        .unwrap_err();
        assert!(err.issues.iter().any(|issue| issue.reason.contains("base64")));
    }

    #[test]
    fn refuses_a_caller_supplied_physical_path_that_is_not_a_content_hash_file_name() {
        let schema = docs_schema();
        let err = validate_record(
            &schema,
            &json!({
                "id": "d1",
                "cover": { "physicalPath": "../../victim/secret.txt", "virtualPath": "/images/a.png" },
            }),
            ValidateOptions::default(),
        )
        .unwrap_err();
        assert!(
            err.issues
                .iter()
                .any(|issue| issue.field == "cover" && issue.reason.contains("content-hash file name")),
            "{:?}",
            err.issues
        );
    }

    #[test]
    fn round_trips_an_image_field_it_coerced_itself() {
        let schema = docs_schema();
        let coerced = validate_record(
            &schema,
            &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
            ValidateOptions::default(),
        )
        .unwrap();

        let again = validate_record(
            &schema,
            &json!({ "id": "d1", "cover": coerced.get("cover").unwrap() }),
            ValidateOptions::default(),
        )
        .unwrap();

        assert_eq!(again.get("cover"), coerced.get("cover"));
    }

    #[test]
    fn hashes_identical_image_content_to_the_same_physical_path() {
        let schema = docs_schema();
        let one = validate_record(
            &schema,
            &json!({ "id": "d1", "cover": { "content": "c2FtZS1ieXRlcw==", "virtualPath": "/images/a.png" } }),
            ValidateOptions::default(),
        )
        .unwrap();
        let two = validate_record(
            &schema,
            &json!({ "id": "d2", "cover": { "content": "c2FtZS1ieXRlcw==", "virtualPath": "/images/b.png" } }),
            ValidateOptions::default(),
        )
        .unwrap();
        assert_eq!(
            one.get("cover").unwrap().get("physicalPath"),
            two.get("cover").unwrap().get("physicalPath")
        );
    }
}