hayson_engine/schema/

parse.rs

use std::collections::BTreeMap;

use serde_json::{Map, Value};

use crate::errors::HaysonSchemaError;

use super::types::{DefaultValue, FieldSpec, FieldType, IndexKind, TableSchema, SCHEMA_VERSION};

const FIELD_SPEC_KEYS: &[&str] = &[
    "type",
    "required",
    "unique",
    "index",
    "fulltext",
    "cypher",
    "passwordHash",
    "values",
    "default",
    "addressable",
];
const SCHEMA_KEYS: &[&str] = &[
    "schemaVersion",
    "table",
    "revision",
    "primaryKey",
    "partitionBy",
    "fields",
];

fn fail(message: impl Into<String>) -> HaysonSchemaError {
    HaysonSchemaError(message.into())
}

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

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

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

fn field_type_of(raw: &str) -> Option<FieldType> {
    Some(match raw {
        "string" => FieldType::String,
        "int" => FieldType::Int,
        "float" => FieldType::Float,
        "bool" => FieldType::Bool,
        "timestamp" => FieldType::Timestamp,
        "enum" => FieldType::Enum,
        "json" => FieldType::Json,
        "string[]" => FieldType::StringArray,
        "int[]" => FieldType::IntArray,
        "float[]" => FieldType::FloatArray,
        "bool[]" => FieldType::BoolArray,
        "markdown" => FieldType::Markdown,
        "image" => FieldType::Image,
        _ => return None,
    })
}

fn parse_flag(source: &Map<String, Value>, key: &str, what: &str) -> Result<bool, HaysonSchemaError> {
    match source.get(key) {
        None => Ok(false),
        Some(Value::Bool(b)) => Ok(*b),
        Some(_) => Err(fail(format!("{what} must be a boolean"))),
    }
}

fn parse_default(
    raw: Option<&Value>,
    field_type: FieldType,
    what: &str,
) -> Result<Option<DefaultValue>, HaysonSchemaError> {
    let Some(raw) = raw else { return Ok(None) };
    if let Value::String(s) = raw {
        if s == "now" {
            if field_type != FieldType::Timestamp {
                return Err(fail(format!("{what} may only be \"now\" on a timestamp field")));
            }
            return Ok(Some(DefaultValue::Now));
        }
    }
    if raw.is_null() {
        return Ok(Some(DefaultValue::Value(Value::Null)));
    }
    if raw.is_array() {
        if !field_type.is_array() {
            return Err(fail(format!(
                "{what} must not be an array on a {} field",
                field_type.as_str()
            )));
        }
        return Ok(Some(DefaultValue::Value(raw.clone())));
    }
    if !(raw.is_string() || raw.is_number() || raw.is_boolean()) {
        return Err(fail(format!("{what} must be a string, number, boolean, array or null")));
    }
    if field_type.is_array() {
        return Err(fail(format!(
            "{what} must be an array on a {} field",
            field_type.as_str()
        )));
    }
    Ok(Some(DefaultValue::Value(raw.clone())))
}

fn parse_field_spec(name: &str, raw: &Value) -> Result<FieldSpec, HaysonSchemaError> {
    let source = as_object(raw, &format!("field \"{name}\""))?;
    for key in source.keys() {
        if !FIELD_SPEC_KEYS.contains(&key.as_str()) {
            return Err(fail(format!("field \"{name}\" has an unknown option \"{key}\"")));
        }
    }

    let type_raw = source
        .get("type")
        .and_then(Value::as_str)
        .ok_or_else(|| fail(format!("field \"{name}\" has an unsupported type")))?;
    let field_type = field_type_of(type_raw)
        .ok_or_else(|| fail(format!("field \"{name}\" has an unsupported type {type_raw:?}")))?;

    let index = match source.get("index") {
        None => None,
        Some(Value::String(s)) if s == "exact" => Some(IndexKind::Exact),
        Some(Value::String(s)) if s == "sorted" => Some(IndexKind::Sorted),
        Some(other) => return Err(fail(format!("field \"{name}\" has an unsupported index {other:?}"))),
    };
    if index == Some(IndexKind::Sorted) && !field_type.is_sortable() {
        return Err(fail(format!(
            "field \"{name}\" of type {} cannot carry a sorted index",
            field_type.as_str()
        )));
    }

    let fulltext = parse_flag(source, "fulltext", &format!("field \"{name}\" option \"fulltext\""))?;
    if fulltext && field_type != FieldType::String && field_type != FieldType::StringArray {
        return Err(fail(format!(
            "field \"{name}\" of type {} cannot be indexed for full-text search",
            field_type.as_str()
        )));
    }

    let unique = parse_flag(source, "unique", &format!("field \"{name}\" option \"unique\""))?;
    if unique && field_type.is_array() {
        return Err(fail(format!(
            "field \"{name}\" of type {} cannot be unique",
            field_type.as_str()
        )));
    }

    let cypher = parse_flag(source, "cypher", &format!("field \"{name}\" option \"cypher\""))?;
    if cypher {
        // NOTE: a random IV makes ciphertext non-comparable, so it can't feed an index or unique.
        if index.is_some() {
            return Err(fail(format!(
                "encrypted field \"{name}\" cannot carry an index — a random IV makes two equal values differ"
            )));
        }
        if unique {
            return Err(fail(format!(
                "encrypted field \"{name}\" cannot be unique — equal ciphertext is not equal values"
            )));
        }
        if fulltext {
            return Err(fail(format!(
                "encrypted field \"{name}\" cannot be indexed for full-text search"
            )));
        }
        if source.contains_key("default") {
            return Err(fail(format!(
                "encrypted field \"{name}\" cannot carry a default — it would sit in the schema in clear"
            )));
        }
    }

    let password_hash = parse_flag(
        source,
        "passwordHash",
        &format!("field \"{name}\" option \"passwordHash\""),
    )?;
    if password_hash {
        if field_type != FieldType::String {
            return Err(fail(format!(
                "field \"{name}\" of type {} cannot hold a password hash — only a string can",
                field_type.as_str()
            )));
        }
        // NOTE: a random salt makes two hashes of one password differ, so nothing can be looked up by it.
        if index.is_some() || unique || fulltext {
            return Err(fail(format!(
                "password field \"{name}\" cannot be indexed, unique or searched — a random salt makes two equal passwords differ"
            )));
        }
        if source.contains_key("default") {
            return Err(fail(format!(
                "password field \"{name}\" cannot carry a default — it would sit in the schema in clear"
            )));
        }
    }

    let addressable = match source.get("addressable") {
        None => true,
        Some(_) if field_type != FieldType::Markdown => {
            return Err(fail(format!(
                "field \"{name}\" of type {} must not declare \"addressable\" — only a markdown field can opt out of a virtual path of its own",
                field_type.as_str()
            )))
        }
        Some(Value::Bool(flag)) => *flag,
        Some(_) => {
            return Err(fail(format!(
                "field \"{name}\" option \"addressable\" must be a boolean"
            )))
        }
    };

    if field_type.has_virtual_path() {
        if addressable {
            if index.is_some() {
                return Err(fail(format!(
                    "field \"{name}\" of type {} cannot carry an index — its virtual path is indexed by the engine",
                    field_type.as_str()
                )));
            }
            if unique {
                return Err(fail(format!(
                    "field \"{name}\" of type {} cannot declare \"unique\" — its virtual path is always unique",
                    field_type.as_str()
                )));
            }
        } else {
            if index.is_some() {
                return Err(fail(format!(
                    "field \"{name}\" of type {} cannot carry an index — a whole markdown document is not a lookup key",
                    field_type.as_str()
                )));
            }
            if unique {
                return Err(fail(format!(
                    "field \"{name}\" of type {} cannot declare \"unique\" — a whole markdown document is not a key to be unique on",
                    field_type.as_str()
                )));
            }
        }
        if fulltext {
            return Err(fail(format!(
                "field \"{name}\" of type {} cannot be indexed for full-text search",
                field_type.as_str()
            )));
        }
        if cypher {
            return Err(fail(format!(
                "field \"{name}\" of type {} cannot be encrypted",
                field_type.as_str()
            )));
        }
        if source.contains_key("default") {
            return Err(fail(format!(
                "field \"{name}\" of type {} cannot carry a default",
                field_type.as_str()
            )));
        }
    }

    let values = if field_type == FieldType::Enum {
        let raws = source
            .get("values")
            .and_then(Value::as_array)
            .filter(|v| !v.is_empty())
            .ok_or_else(|| fail(format!("enum field \"{name}\" must list its \"values\"")))?;
        let mut values = Vec::with_capacity(raws.len());
        for raw in raws {
            let s = raw
                .as_str()
                .ok_or_else(|| fail(format!("enum field \"{name}\" must list string \"values\"")))?;
            values.push(s.to_string());
        }
        let unique_count = values.iter().collect::<std::collections::HashSet<_>>().len();
        if unique_count != values.len() {
            return Err(fail(format!("enum field \"{name}\" has duplicate values")));
        }
        Some(values)
    } else {
        if source.contains_key("values") {
            return Err(fail(format!(
                "field \"{name}\" of type {} must not declare \"values\"",
                field_type.as_str()
            )));
        }
        None
    };

    let mut spec = FieldSpec::new(field_type);
    spec.required = parse_flag(source, "required", &format!("field \"{name}\" option \"required\""))?;
    spec.unique = unique;
    spec.fulltext = fulltext;
    spec.cypher = cypher;
    spec.password_hash = password_hash;
    spec.addressable = addressable;
    // NOTE: unique needs the exact registry to answer "is this value taken" without touching disk.
    spec.index = if unique {
        Some(index.unwrap_or(IndexKind::Exact))
    } else {
        index
    };
    spec.values = values;

    let default = parse_default(
        source.get("default"),
        field_type,
        &format!("field \"{name}\" option \"default\""),
    )?;
    if let Some(DefaultValue::Value(ref value)) = default {
        if let Some(values) = &spec.values {
            if !value.is_null() {
                let as_text = match value {
                    Value::String(s) => s.clone(),
                    other => other.to_string(),
                };
                if !values.contains(&as_text) {
                    return Err(fail(format!(
                        "enum field \"{name}\" has a default outside of its values"
                    )));
                }
            }
        }
    }
    spec.default = default;

    if field_type.is_array() {
        if let Some(DefaultValue::Value(Value::Array(items))) = &spec.default {
            let element = field_type.element_type().expect("array type has an element type");
            let bad = items.iter().any(|item| match element {
                FieldType::String => !item.is_string(),
                FieldType::Bool => !item.is_boolean(),
                _ => !item.is_number(),
            });
            if bad {
                return Err(fail(format!(
                    "field \"{name}\" has a default that does not match {}",
                    field_type.as_str()
                )));
            }
        }
    }

    Ok(spec)
}

fn parse_partition_by(
    raw: Option<&Value>,
    table: &str,
    primary_key: &str,
    fields: &BTreeMap<String, FieldSpec>,
) -> Result<Option<String>, HaysonSchemaError> {
    let Some(raw) = raw else { return Ok(None) };
    let name = raw
        .as_str()
        .filter(|s| !s.is_empty())
        .ok_or_else(|| fail(format!("\"partitionBy\" of table \"{table}\" must name a field")))?;
    let spec = fields.get(name).ok_or_else(|| {
        fail(format!(
            "\"partitionBy\" of table \"{table}\" names an unknown field \"{name}\""
        ))
    })?;
    if name == primary_key {
        return Err(fail(format!(
            "\"partitionBy\" of table \"{table}\" cannot be its primary key \"{name}\" — that is a file per record"
        )));
    }
    if spec.field_type != FieldType::String {
        return Err(fail(format!(
            "partition field \"{name}\" of table \"{table}\" must be a string, not {}",
            spec.field_type.as_str()
        )));
    }
    if !spec.required {
        return Err(fail(format!(
            "partition field \"{name}\" of table \"{table}\" must be required — every record needs a partition"
        )));
    }
    if spec.unique {
        return Err(fail(format!(
            "partition field \"{name}\" of table \"{table}\" cannot be unique — that is a file per record"
        )));
    }
    if spec.cypher || spec.password_hash {
        return Err(fail(format!(
            "partition field \"{name}\" of table \"{table}\" cannot be encrypted — its value is the file name"
        )));
    }
    Ok(Some(name.to_string()))
}

/// Moves the marker of a schema document written before the `"schemaVersion"` rename — the
/// historical `"hayson"` key — onto the current key, leaving an already-current document alone.
/// This is for readers of files someone else wrote in the old format (the format-1
/// `<table>.schema.json` the offline migration rewrites), never for an authoring path:
/// [`parse_schema`] itself rejects the old key.
pub(crate) fn with_current_schema_marker(mut document: Value) -> Value {
    if let Some(object) = document.as_object_mut() {
        if !object.contains_key("schemaVersion") {
            if let Some(marker) = object.remove("hayson") {
                object.insert("schemaVersion".to_string(), marker);
            }
        }
    }
    document
}

/// Parses and normalizes a table schema document (the contents of `<table>.schema.json`).
pub fn parse_schema(input: &Value) -> Result<TableSchema, HaysonSchemaError> {
    let source = as_object(input, "schema")?;

    if source.get("schemaVersion") != Some(&Value::from(SCHEMA_VERSION)) {
        return Err(fail(format!(
            "schema must declare \"schemaVersion\": {SCHEMA_VERSION}, got {:?}",
            source.get("schemaVersion")
        )));
    }

    for key in source.keys() {
        if !SCHEMA_KEYS.contains(&key.as_str()) {
            return Err(fail(format!(
                "schema of table {:?} has an unknown option \"{key}\" — it was most likely written by a newer version of hayson, and guessing what it means would read the table wrongly. Upgrade @theritual/hayson to open it.",
                source.get("table").and_then(Value::as_str)
            )));
        }
    }

    let table = source
        .get("table")
        .and_then(Value::as_str)
        .filter(|s| is_table_name(s))
        .ok_or_else(|| fail(format!("schema has an unusable table name {:?}", source.get("table"))))?
        .to_string();

    let revision = match source.get("revision") {
        None => 1,
        Some(Value::Number(n)) if n.is_u64() && n.as_u64().unwrap() >= 1 => n.as_u64().unwrap() as u32,
        _ => {
            return Err(fail(format!(
                "schema of table \"{table}\" must have a positive integer \"revision\""
            )))
        }
    };

    let primary_key = source
        .get("primaryKey")
        .and_then(Value::as_str)
        .filter(|s| !s.is_empty())
        .ok_or_else(|| fail(format!("schema of table \"{table}\" must name its \"primaryKey\"")))?
        .to_string();

    let raw_fields = as_object(
        source.get("fields").unwrap_or(&Value::Null),
        &format!("\"fields\" of table \"{table}\""),
    )?;
    if raw_fields.is_empty() {
        return Err(fail(format!("schema of table \"{table}\" declares no fields")));
    }

    let mut fields = BTreeMap::new();
    for (name, raw) in raw_fields {
        if !is_field_name(name) {
            return Err(fail(format!(
                "field name {name:?} in table \"{table}\" is not usable (reserved or invalid)"
            )));
        }
        fields.insert(name.clone(), parse_field_spec(name, raw)?);
    }

    {
        let key = fields.get_mut(&primary_key).ok_or_else(|| {
            fail(format!(
                "primary key \"{primary_key}\" of table \"{table}\" is not among its fields"
            ))
        })?;
        if key.field_type != FieldType::String && key.field_type != FieldType::Int {
            return Err(fail(format!(
                "primary key \"{primary_key}\" of table \"{table}\" must be a string or an int"
            )));
        }
        if key.cypher || key.password_hash {
            return Err(fail(format!(
                "primary key \"{primary_key}\" of table \"{table}\" cannot be encrypted — it is indexed by definition"
            )));
        }
        key.required = true;
        key.unique = true;
        key.index = Some(IndexKind::Exact);
    }

    let partition_by = parse_partition_by(source.get("partitionBy"), &table, &primary_key, &fields)?;

    Ok(TableSchema {
        table,
        revision,
        primary_key,
        partition_by,
        fields,
    })
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::path::PathBuf;

    use super::*;

    fn fixture(name: &str) -> Value {
        let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        path.push(format!("../../packages/hayson/test/fixtures/v1/{name}"));
        let text = fs::read_to_string(&path).unwrap_or_else(|e| panic!("{path:?}: {e}"));
        serde_json::from_str(&text).unwrap()
    }

    #[test]
    fn parses_the_ts_reference_fixture_and_normalizes_the_primary_key() {
        // NOTE: a v1 fixture is a frozen format-1 file — its marker still sits under "hayson".
        let schema = parse_schema(&with_current_schema_marker(fixture("unique-field/users.schema.json"))).unwrap();
        assert_eq!(schema.table, "users");
        assert_eq!(schema.primary_key, "id");
        let id = &schema.fields["id"];
        assert!(id.required);
        assert!(id.unique);
        assert_eq!(id.index, Some(IndexKind::Exact));

        let email = &schema.fields["email"];
        assert!(email.unique);
        assert_eq!(email.index, Some(IndexKind::Exact));

        let age = &schema.fields["age"];
        assert_eq!(age.field_type, FieldType::Int);
        assert_eq!(age.index, Some(IndexKind::Sorted));

        let tags = &schema.fields["tags"];
        assert_eq!(tags.field_type, FieldType::StringArray);
    }

    /// 5.0.0 renamed the marker: a caller still passing the old key gets told which key to use,
    /// not a schema parsed on a guess.
    #[test]
    fn rejects_a_schema_document_that_declares_the_pre_rename_marker() {
        let doc = serde_json::json!({
            "hayson": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": { "id": { "type": "string" } },
        });
        let failure = parse_schema(&doc).unwrap_err();
        assert!(failure.0.contains("schemaVersion"), "{}", failure.0);
    }

    #[test]
    fn moves_a_pre_rename_marker_onto_the_current_key_and_leaves_a_current_one_alone() {
        let old = with_current_schema_marker(serde_json::json!({ "hayson": 1, "table": "users" }));
        assert_eq!(old, serde_json::json!({ "table": "users", "schemaVersion": 1 }));

        let current = serde_json::json!({ "schemaVersion": 1, "table": "users" });
        assert_eq!(with_current_schema_marker(current.clone()), current);
    }

    #[test]
    fn rejects_an_unknown_schema_option() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": { "id": { "type": "string" } },
            "somethingNew": true,
        });
        assert!(parse_schema(&doc).is_err());
    }

    #[test]
    fn rejects_an_unknown_field_option() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": { "id": { "type": "string", "wat": true } },
        });
        assert!(parse_schema(&doc).is_err());
    }

    #[test]
    fn rejects_sorted_index_on_a_non_sortable_type() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "tags": { "type": "string[]", "index": "sorted" },
            },
        });
        assert!(parse_schema(&doc).is_err());
    }

    #[test]
    fn rejects_a_cypher_field_carrying_an_index() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "ssn": { "type": "string", "cypher": true, "index": "exact" },
            },
        });
        assert!(parse_schema(&doc).is_err());
    }

    #[test]
    fn rejects_partition_by_naming_the_primary_key() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "notes",
            "primaryKey": "id",
            "partitionBy": "id",
            "fields": {
                "id": { "type": "string" },
            },
        });
        assert!(parse_schema(&doc).is_err());
    }

    #[test]
    fn parses_a_partitioned_schema() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "notes",
            "primaryKey": "id",
            "partitionBy": "kind",
            "fields": {
                "id": { "type": "string" },
                "kind": { "type": "string", "required": true, "index": "exact" },
                "title": { "type": "string", "required": true, "fulltext": true },
            },
        });
        let schema = parse_schema(&doc).unwrap();
        assert_eq!(schema.partition_by, Some("kind".to_string()));
    }

    #[test]
    fn accepts_markdown_and_image_field_declarations() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown" },
                "cover": { "type": "image" },
            },
        });
        let schema = parse_schema(&doc).unwrap();
        assert_eq!(schema.fields["page"].field_type, FieldType::Markdown);
        assert_eq!(schema.fields["cover"].field_type, FieldType::Image);
    }

    #[test]
    fn rejects_unique_declared_directly_on_a_markdown_field() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown", "unique": true },
            },
        });
        assert!(parse_schema(&doc).is_err());
    }

    #[test]
    fn rejects_an_index_declared_directly_on_an_image_field() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "cover": { "type": "image", "index": "exact" },
            },
        });
        assert!(parse_schema(&doc).is_err());
    }

    #[test]
    fn rejects_fulltext_declared_directly_on_a_markdown_field() {
        let doc = serde_json::json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": { "type": "markdown", "fulltext": true },
            },
        });
        assert!(parse_schema(&doc).is_err());
    }

    fn docs_schema_with(page: serde_json::Value) -> Result<TableSchema, HaysonSchemaError> {
        parse_schema(&serde_json::json!({
            "schemaVersion": 1,
            "table": "docs",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "page": page,
            },
        }))
    }

    #[test]
    fn a_markdown_field_is_addressable_unless_it_says_otherwise() {
        let default = docs_schema_with(serde_json::json!({ "type": "markdown" })).unwrap();
        let opted_out = docs_schema_with(serde_json::json!({ "type": "markdown", "addressable": false })).unwrap();
        let opted_in = docs_schema_with(serde_json::json!({ "type": "markdown", "addressable": true })).unwrap();

        assert!(default.fields["page"].addressable);
        assert!(default.fields["page"].has_virtual_path());
        assert!(!opted_out.fields["page"].addressable);
        assert!(
            !opted_out.fields["page"].has_virtual_path(),
            "a field nothing can link to holds no virtual path of its own"
        );
        assert!(opted_in.fields["page"].has_virtual_path());
    }

    #[test]
    fn only_a_markdown_field_may_declare_addressable() {
        for field_type in ["image", "string", "json"] {
            let refused = docs_schema_with(serde_json::json!({ "type": field_type, "addressable": false }))
                .expect_err("only markdown opts out of a virtual path");

            assert!(refused.0.contains("addressable"), "{}", refused.0);
            assert!(refused.0.contains("markdown"), "{}", refused.0);
        }
    }

    #[test]
    fn addressable_has_to_be_a_boolean() {
        let refused = docs_schema_with(serde_json::json!({ "type": "markdown", "addressable": "no" }))
            .expect_err("\"no\" is not a boolean");

        assert!(refused.0.contains("must be a boolean"), "{}", refused.0);
    }

    #[test]
    fn a_non_addressable_markdown_field_is_refused_the_same_five_options_but_for_its_own_reason() {
        let index = docs_schema_with(serde_json::json!({ "type": "markdown", "addressable": false, "index": "exact" }))
            .expect_err("a whole markdown document is not a lookup key");
        let unique = docs_schema_with(serde_json::json!({ "type": "markdown", "addressable": false, "unique": true }))
            .expect_err("a whole markdown document is not a unique key");

        assert!(
            index.0.contains("not a lookup key"),
            "the refusal must not still claim the field has an indexed virtual path: {}",
            index.0
        );
        assert!(unique.0.contains("not a key to be unique on"), "{}", unique.0);
        for option in [
            serde_json::json!({ "type": "markdown", "addressable": false, "fulltext": true }),
            serde_json::json!({ "type": "markdown", "addressable": false, "cypher": true }),
            serde_json::json!({ "type": "markdown", "addressable": false, "default": "x" }),
        ] {
            assert!(
                docs_schema_with(option.clone()).is_err(),
                "{option} has to stay refused on a non-addressable field too"
            );
        }
    }

    #[test]
    fn opting_out_survives_the_schema_document_it_is_rendered_into() {
        let schema = docs_schema_with(serde_json::json!({ "type": "markdown", "addressable": false })).unwrap();

        let again = parse_schema(&super::super::serialize::schema_document(&schema)).unwrap();

        assert_eq!(again, schema);
        assert!(!again.fields["page"].addressable);
    }
}