hayson-engine 0.25.0

use serde_json::Value;

use crate::errors::HaysonPartitionError;

use super::types::TableSchema;
use super::validate::HaysonRecord;

/// The single partition of a table that declares no `partitionBy` — the empty name every
/// unpartitioned slot, tally and log carries, so both layouts share one code path.
pub const UNPARTITIONED: &str = "";

const MAX_PARTITION_LEN: usize = 64;

pub fn is_partitioned(schema: &TableSchema) -> bool {
    schema.partition_by.as_ref().is_some_and(|field| !field.is_empty())
}

/// Mirrors the TS `PARTITION_VALUE` (`^[a-z0-9_-]{1,64}$`) — a partition value becomes a file name
/// verbatim, so anything that could escape the directory or collide case-insensitively is out.
fn names_a_file(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= MAX_PARTITION_LEN
        && value
            .chars()
            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '-')
}

/// The partition `record` belongs to: [`UNPARTITIONED`] for a table with no `partitionBy`, and the
/// value of the named field otherwise.
pub fn partition_of(schema: &TableSchema, record: &HaysonRecord) -> Result<String, HaysonPartitionError> {
    let Some(field) = schema.partition_by.as_ref().filter(|field| !field.is_empty()) else {
        return Ok(UNPARTITIONED.to_string());
    };
    assert_partition_value(schema, record.get(field))
}

pub fn assert_partition_value(schema: &TableSchema, value: Option<&Value>) -> Result<String, HaysonPartitionError> {
    let field = schema.partition_by.clone().unwrap_or_default();
    match value.and_then(Value::as_str) {
        Some(text) if names_a_file(text) => Ok(text.to_string()),
        _ => Err(HaysonPartitionError::new(
            schema.table.clone(),
            field,
            format!(
                "{} cannot name a partition file — expected 1 to 64 characters of a-z, 0-9, \"_\" or \"-\"",
                match value {
                    Some(value) => value.to_string(),
                    None => "undefined".to_string(),
                }
            ),
        )),
    }
}

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

    use super::super::parse::parse_schema;
    use super::*;

    fn notes_schema(partitioned: bool) -> TableSchema {
        let mut spec = json!({
            "schemaVersion": 1,
            "table": "notes",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "kind": { "type": "string", "required": true },
            },
        });
        if partitioned {
            spec.as_object_mut()
                .unwrap()
                .insert("partitionBy".into(), json!("kind"));
        }
        parse_schema(&spec).unwrap()
    }

    fn record(kind: Value) -> HaysonRecord {
        let mut record = HaysonRecord::new();
        record.insert("id".into(), json!("n1"));
        record.insert("kind".into(), kind);
        record
    }

    #[test]
    fn an_unpartitioned_table_puts_every_record_in_the_empty_partition() {
        let schema = notes_schema(false);
        assert!(!is_partitioned(&schema));
        assert_eq!(partition_of(&schema, &record(json!("note"))).unwrap(), UNPARTITIONED);
    }

    #[test]
    fn a_partitioned_table_reads_the_partition_off_the_named_field() {
        let schema = notes_schema(true);
        assert!(is_partitioned(&schema));
        assert_eq!(partition_of(&schema, &record(json!("adr"))).unwrap(), "adr");
    }

    #[test]
    fn refuses_a_value_that_cannot_name_a_file() {
        let schema = notes_schema(true);
        for kind in [
            json!("../escape"),
            json!("Upper"),
            json!("has space"),
            json!(""),
            json!(7),
        ] {
            let error = partition_of(&schema, &record(kind)).unwrap_err();
            assert_eq!(error.field, "kind");
            assert!(error.to_string().contains("cannot name a partition file"));
        }
        assert!(partition_of(&schema, &HaysonRecord::new()).is_err());
    }

    #[test]
    fn refuses_a_value_longer_than_sixty_four_characters_and_takes_one_of_exactly_that_length() {
        let schema = notes_schema(true);
        assert_eq!(
            partition_of(&schema, &record(json!("a".repeat(64)))).unwrap(),
            "a".repeat(64)
        );
        assert!(partition_of(&schema, &record(json!("a".repeat(65)))).is_err());
    }
}