hayson-binding-shared 0.17.0

use std::collections::HashMap;

use std::time::Duration;

use hayson_engine::catalog::CatalogCompaction;
use hayson_engine::crypto::PasswordHashParams;
use hayson_engine::crypto::{CipherOptions, MissingKeyPolicy};
use hayson_engine::database::{
    BackupOptions, FollowOption, LayoutKind, LayoutMigrationOptions, LayoutOptions, LockMode, OpenOptions, PackLimits,
    PackOptions, RestoreOptions, UnpackOptions, VerifyOptions,
};
use hayson_engine::lock::{GraceServed, LockOptions};
use serde_json::Value;

/// Rejection of an `open()` options document, before the engine is ever touched.
#[derive(Debug, PartialEq, Eq)]
pub enum OptionsError {
    /// A malformed option: wrong JSON type, unknown enum member, out-of-range number.
    Shape(String),
    /// Unusable key material — reported to JS as a cipher error, not a schema one.
    Key(String),
}

fn shape<T>(message: impl Into<String>) -> Result<T, OptionsError> {
    Err(OptionsError::Shape(message.into()))
}

fn as_object<'a>(value: &'a Value, what: &str) -> Result<&'a serde_json::Map<String, Value>, OptionsError> {
    match value.as_object() {
        Some(object) => Ok(object),
        None => shape(format!("\"{what}\" must be a JSON object")),
    }
}

fn optional_bool(object: &serde_json::Map<String, Value>, key: &str) -> Result<Option<bool>, OptionsError> {
    match object.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::Bool(value)) => Ok(Some(*value)),
        Some(_) => shape(format!("\"{key}\" must be a boolean")),
    }
}

fn optional_u64(object: &serde_json::Map<String, Value>, key: &str) -> Result<Option<u64>, OptionsError> {
    match object.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::Number(number)) => match number.as_u64() {
            Some(value) => Ok(Some(value)),
            None => shape(format!("\"{key}\" must be a non-negative whole number")),
        },
        Some(_) => shape(format!("\"{key}\" must be a number")),
    }
}

fn optional_i64(object: &serde_json::Map<String, Value>, key: &str) -> Result<Option<i64>, OptionsError> {
    match object.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::Number(number)) => match number.as_i64() {
            Some(value) => Ok(Some(value)),
            None => shape(format!("\"{key}\" must be a whole number")),
        },
        Some(_) => shape(format!("\"{key}\" must be a number")),
    }
}

fn optional_string(object: &serde_json::Map<String, Value>, key: &str) -> Result<Option<String>, OptionsError> {
    match object.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(value)) => Ok(Some(value.clone())),
        Some(_) => shape(format!("\"{key}\" must be a string")),
    }
}

/// `lock` in the shapes the TS `OpenOptions` takes (`boolean | LockOptions`): an object of lease
/// tuning, or `false` to open without taking the writer lock at all — mutual exclusion between
/// writers is then not enforced, which is the whole point and the whole danger of the option.
/// `true` is TS's "lock the usual way" and resolves to the engine's defaults, not to skipping.
fn parse_lock(value: &Value) -> Result<LockMode, OptionsError> {
    match value {
        Value::Bool(false) => Ok(LockMode::Skip),
        Value::Bool(true) => Ok(LockMode::Enabled(LockOptions::default())),
        Value::Object(object) => Ok(LockMode::Enabled(LockOptions {
            stale_after_ms: optional_u64(object, "staleAfterMs")?,
            heartbeat_ms: optional_u64(object, "heartbeatMs")?,
            timeout_ms: optional_u64(object, "timeoutMs")?,
            retry_interval_ms: optional_u64(object, "retryIntervalMs")?,
            grace_served: parse_grace_served(object)?,
        })),
        _ => shape("\"lock\" must be a boolean or an object of lock options"),
    }
}

/// `graceServed` — the lease a host has already challenged and waited the hand-over grace out on,
/// so that the open it precedes reclaims that lease without the engine sleeping inside a call the
/// host is waiting on. Not part of the public TS `LockOptions`: the binding layer synthesises it
/// between two attempts of its own, off the `holder` its previous refusal named.
///
/// A token naming a lease that has since moved is not an error — it simply buys nothing, and the
/// ordinary hand-over runs instead. Only its shape is judged here.
fn parse_grace_served(object: &serde_json::Map<String, Value>) -> Result<Option<GraceServed>, OptionsError> {
    let served = match object.get("graceServed") {
        None | Some(Value::Null) => return Ok(None),
        Some(value) => as_object(value, "graceServed")?,
    };
    let Some(owner) = optional_string(served, "owner")? else {
        return shape("\"graceServed\" must name the \"owner\" whose grace it waited out");
    };
    let Some(heartbeat_at) = optional_i64(served, "heartbeatAt")? else {
        return shape("\"graceServed\" must name the \"heartbeatAt\" whose grace it waited out");
    };
    Ok(GraceServed::parse(&owner, heartbeat_at))
}

/// `follow` in the three shapes the TS `OpenOptions` takes: `false` to never catch up unprompted,
/// `true` for the engine's default interval, or `{ intervalMs }` for a custom one. Absent is not
/// handled here — it means "resolve by `readonly`" and never reaches this function.
fn parse_follow(value: &Value) -> Result<FollowOption, OptionsError> {
    match value {
        Value::Bool(false) => Ok(FollowOption::Off),
        Value::Bool(true) => Ok(FollowOption::Every(Duration::ZERO)),
        Value::Object(object) => match optional_u64(object, "intervalMs")? {
            Some(interval_ms) => Ok(FollowOption::Every(Duration::from_millis(interval_ms))),
            None => Ok(FollowOption::Every(Duration::ZERO)),
        },
        _ => shape("\"follow\" must be a boolean or an object with \"intervalMs\""),
    }
}

/// `compaction` in the two shapes the TS `OpenOptions` takes: `false` to never rewrite a log
/// unprompted, or `{ deadBytesRatio, minDeadBytes }` to set both thresholds. Absent is not handled
/// here — it means "leave it to the catalog" and never reaches this function.
fn parse_compaction(value: &Value) -> Result<CatalogCompaction, OptionsError> {
    match value {
        Value::Bool(false) => Ok(CatalogCompaction::Disabled),
        Value::Object(object) => {
            let ratio = match object.get("deadBytesRatio") {
                None | Some(Value::Null) => None,
                Some(Value::Number(number)) => number.as_f64(),
                Some(_) => return shape("\"deadBytesRatio\" must be a number"),
            };
            let Some(dead_bytes_ratio) = ratio.filter(|ratio| *ratio > 0.0 && *ratio <= 1.0) else {
                return shape("\"deadBytesRatio\" must be a number greater than 0 and at most 1");
            };
            let min_dead_bytes = optional_u64(object, "minDeadBytes")?.unwrap_or(0);
            Ok(CatalogCompaction::Enabled {
                dead_bytes_ratio,
                min_dead_bytes,
            })
        }
        _ => shape("\"compaction\" must be false or an object of compaction thresholds"),
    }
}

fn parse_layout_kind(raw: &str) -> Result<LayoutKind, OptionsError> {
    match raw {
        "flat" => Ok(LayoutKind::Flat),
        "nested" => Ok(LayoutKind::Nested),
        other => shape(format!("\"layout\" must be \"flat\" or \"nested\" — got \"{other}\"")),
    }
}

/// The `layout`/`metadataFile`/`manifest` triple, passed on unresolved: whether the two file names
/// agree, and whether the requested arrangement matches the one on disk, is the engine's call, so
/// both refusals read the same however a caller reaches them.
fn parse_layout(object: &serde_json::Map<String, Value>) -> Result<LayoutOptions, OptionsError> {
    Ok(LayoutOptions {
        kind: match optional_string(object, "layout")? {
            Some(raw) => Some(parse_layout_kind(&raw)?),
            None => None,
        },
        metadata_file: optional_string(object, "metadataFile")?,
        manifest: optional_string(object, "manifest")?,
    })
}

fn parse_missing_key_policy(raw: &str) -> Result<MissingKeyPolicy, OptionsError> {
    match raw {
        "throw" => Ok(MissingKeyPolicy::Throw),
        "omit" => Ok(MissingKeyPolicy::Omit),
        "raw" => Ok(MissingKeyPolicy::Raw),
        other => shape(format!(
            "\"onMissingKey\" must be one of \"throw\", \"omit\", \"raw\" — got \"{other}\""
        )),
    }
}

/// Decodes one key of the `cypher.keys` map. `encoding` is the document-wide `keyEncoding`.
fn decode_key(id: &str, raw: &str, encoding: KeyEncoding) -> Result<Vec<u8>, OptionsError> {
    let decoded = match encoding {
        KeyEncoding::Base64 => decode_base64(raw),
        KeyEncoding::Hex => decode_hex(raw),
    };
    decoded.ok_or_else(|| OptionsError::Key(format!("key \"{id}\" is not valid {}", encoding.as_str())))
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum KeyEncoding {
    Base64,
    Hex,
}

impl KeyEncoding {
    fn as_str(self) -> &'static str {
        match self {
            KeyEncoding::Base64 => "base64",
            KeyEncoding::Hex => "hex",
        }
    }
}

fn base64_symbol(byte: u8) -> Option<u8> {
    match byte {
        b'A'..=b'Z' => Some(byte - b'A'),
        b'a'..=b'z' => Some(byte - b'a' + 26),
        b'0'..=b'9' => Some(byte - b'0' + 52),
        // NOTE: accepts the URL-safe alphabet too — Node emits it for `base64url` encodings.
        b'+' | b'-' => Some(62),
        b'/' | b'_' => Some(63),
        _ => None,
    }
}

/// Standard and URL-safe base64, padding optional. `None` on any symbol outside the alphabet or a
/// length that cannot be a whole number of bytes.
pub fn decode_base64(raw: &str) -> Option<Vec<u8>> {
    let symbols: &[u8] = raw.trim_end_matches('=').as_bytes();
    if symbols.len() % 4 == 1 {
        return None;
    }
    let mut out = Vec::with_capacity(symbols.len() * 3 / 4);
    let mut accumulator: u32 = 0;
    let mut bits = 0u32;
    for byte in symbols {
        accumulator = (accumulator << 6) | u32::from(base64_symbol(*byte)?);
        bits += 6;
        if bits >= 8 {
            bits -= 8;
            out.push((accumulator >> bits) as u8);
        }
    }
    // Leftover bits of a truncated group must be zero, or the input was not base64 of these bytes.
    if accumulator & ((1 << bits) - 1) != 0 {
        return None;
    }
    Some(out)
}

/// Lowercase or uppercase hex, no separators. `None` on an odd length or a non-hex digit.
pub fn decode_hex(raw: &str) -> Option<Vec<u8>> {
    let digits = raw.as_bytes();
    if !digits.len().is_multiple_of(2) {
        return None;
    }
    let mut out = Vec::with_capacity(digits.len() / 2);
    for pair in digits.chunks_exact(2) {
        let high = (pair[0] as char).to_digit(16)?;
        let low = (pair[1] as char).to_digit(16)?;
        out.push((high * 16 + low) as u8);
    }
    Some(out)
}

/// The `cypher` sub-document of the `open()` options document, on its own — `{ keys, keyId,
/// keyEncoding, onMissingKey }`. Public because a host that reads key material from a file or an
/// environment variable (`crate::keys`) has such a document and no options document around it.
pub fn parse_cypher(value: &Value) -> Result<CipherOptions, OptionsError> {
    let object = as_object(value, "cypher")?;
    let encoding = match optional_string(object, "keyEncoding")?.as_deref() {
        None | Some("base64") => KeyEncoding::Base64,
        Some("hex") => KeyEncoding::Hex,
        Some(other) => {
            return shape(format!(
                "\"keyEncoding\" must be \"base64\" or \"hex\" — got \"{other}\""
            ))
        }
    };

    let raw_keys = match object.get("keys") {
        Some(keys) => as_object(keys, "cypher.keys")?,
        None => return shape("\"cypher\" needs a \"keys\" object"),
    };

    let mut keys = HashMap::with_capacity(raw_keys.len());
    for (id, material) in raw_keys {
        let Some(material) = material.as_str() else {
            return shape(format!("key \"{id}\" must be a string"));
        };
        keys.insert(id.clone(), decode_key(id, material, encoding)?);
    }

    let on_missing_key = match optional_string(object, "onMissingKey")? {
        Some(raw) => Some(parse_missing_key_policy(&raw)?),
        None => None,
    };

    Ok(CipherOptions {
        keys,
        key_id: optional_string(object, "keyId")?,
        on_missing_key,
    })
}

/// Builds `OpenOptions` from the JSON document `open()` takes. Every member is optional; an absent
/// one keeps the engine's own default, so `{}` and `null` both mean "open it the usual way".
pub fn parse_open_options(value: &Value) -> Result<OpenOptions, OptionsError> {
    let mut options = OpenOptions::default();
    if value.is_null() {
        return Ok(options);
    }
    let object = as_object(value, "options")?;

    if let Some(readonly) = optional_bool(object, "readonly")? {
        options.readonly = readonly;
    }
    if let Some(snapshot_every) = optional_u64(object, "snapshotEvery")? {
        options.snapshot_every = snapshot_every;
    }
    if let Some(compaction) = object.get("compaction").filter(|value| !value.is_null()) {
        options.compaction = Some(parse_compaction(compaction)?);
    }
    if let Some(lock) = object.get("lock").filter(|value| !value.is_null()) {
        options.lock = parse_lock(lock)?;
    }
    if let Some(cypher) = object.get("cypher").filter(|value| !value.is_null()) {
        options.cypher = Some(parse_cypher(cypher)?);
    }
    if let Some(follow) = object.get("follow").filter(|value| !value.is_null()) {
        options.follow = parse_follow(follow)?;
    }
    if let Some(max) = optional_u64(object, "maxImageBytes")? {
        options.max_image_bytes = Some(max);
    }
    if let Some(cost) = object.get("passwordHash").filter(|value| !value.is_null()) {
        options.password_hash = parse_password_hash(cost)?;
    }
    options.layout = parse_layout(object)?;

    Ok(options)
}

/// Builds [`LayoutMigrationOptions`] from the JSON document `migrateLayout()` takes — the TS
/// `LayoutMigrationOptions`, member for member, plus the `lock` of `open()` so a host that has to
/// wait differently for the writer lock can say so. Every member is optional; `{}` and `null` both
/// mean "move it into the nested arrangement, keeping the document's name".
///
/// Neither pair of document names is resolved here, for the same reason `parse_layout` resolves
/// nothing: whether the two names of a pair agree is the engine's refusal to make, so it reads the
/// same however a caller reaches it.
pub fn parse_migrate_layout_options(value: &Value) -> Result<LayoutMigrationOptions, OptionsError> {
    let mut options = LayoutMigrationOptions::default();
    if value.is_null() {
        return Ok(options);
    }
    let object = as_object(value, "options")?;

    options.to = match optional_string(object, "to")? {
        Some(raw) => Some(parse_migration_target(&raw)?),
        None => None,
    };
    options.from_metadata_file = optional_string(object, "fromMetadataFile")?;
    options.from_manifest = optional_string(object, "fromManifest")?;
    options.metadata_file = optional_string(object, "metadataFile")?;
    options.manifest_file = optional_string(object, "manifestFile")?;
    options.dry_run = optional_bool(object, "dryRun")?.unwrap_or(false);
    if let Some(lock) = object.get("lock").filter(|value| !value.is_null()) {
        options.lock = parse_lock(lock)?;
    }

    Ok(options)
}

/// Builds [`VerifyOptions`] from the JSON document `verify()` takes:
/// `{ relations?: boolean, images?: boolean, maxImageBytes?: number }`.
/// `null` and `{}` mean the defaults.
pub fn parse_verify_options(value: &Value) -> Result<VerifyOptions, OptionsError> {
    let mut options = VerifyOptions::default();
    if value.is_null() {
        return Ok(options);
    }
    let object = as_object(value, "options")?;
    if let Some(relations) = optional_bool(object, "relations")? {
        options.relations = relations;
    }
    if let Some(images) = optional_bool(object, "images")? {
        options.images = images;
    }
    options.max_image_bytes = match object.get("maxImageBytes") {
        None | Some(Value::Null) => None,
        Some(value) => match value.as_u64() {
            Some(bytes) => Some(bytes),
            None => return shape("\"maxImageBytes\" must be a non-negative integer".to_string()),
        },
    };
    Ok(options)
}

/// Builds [`BackupOptions`] from the JSON document `backup()` takes: `{ indexes?: boolean, excludeTables?: string[], lock? }`.
pub fn parse_backup_options(value: &Value) -> Result<BackupOptions, OptionsError> {
    let mut options = BackupOptions::default();
    if value.is_null() {
        return Ok(options);
    }
    let object = as_object(value, "options")?;
    options.indexes = optional_bool(object, "indexes")?.unwrap_or(false);
    options.exclude_tables = parse_exclude_tables(object)?;
    if let Some(lock) = object.get("lock").filter(|value| !value.is_null()) {
        options.lock = parse_lock(lock)?;
    }
    Ok(options)
}

/// Builds [`RestoreOptions`] from the JSON document `restore()` takes: `{ maxImageBytes?, lock? }`.
pub fn parse_restore_options(value: &Value) -> Result<RestoreOptions, OptionsError> {
    let mut options = RestoreOptions::default();
    if value.is_null() {
        return Ok(options);
    }
    let object = as_object(value, "options")?;
    if let Some(lock) = object.get("lock").filter(|value| !value.is_null()) {
        options.lock = parse_lock(lock)?;
    }
    Ok(options)
}

/// `{ memoryKiB?, iterations?, parallelism? }` — the Argon2id cost of new `passwordHash` values;
/// whatever is left out keeps the engine's default.
fn parse_password_hash(value: &Value) -> Result<PasswordHashParams, OptionsError> {
    let object = as_object(value, "passwordHash")?;
    let mut params = PasswordHashParams::default();
    for (key, target) in [
        ("memoryKiB", &mut params.memory_kib),
        ("iterations", &mut params.iterations),
        ("parallelism", &mut params.parallelism),
    ] {
        if let Some(number) = optional_u64(object, key)? {
            *target = match u32::try_from(number) {
                Ok(number) => number,
                Err(_) => return shape(format!("\"passwordHash.{key}\" is too large")),
            };
        }
    }
    params.validate().map_err(|error| OptionsError::Shape(error.0))?;
    Ok(params)
}

/// `excludeTables?: string[]` of a `backup()`/`pack()` options document.
fn parse_exclude_tables(object: &serde_json::Map<String, Value>) -> Result<Vec<String>, OptionsError> {
    let Some(raw) = object.get("excludeTables").filter(|value| !value.is_null()) else {
        return Ok(Vec::new());
    };
    let Some(items) = raw.as_array() else {
        return shape("\"excludeTables\" must be an array of table names");
    };
    items
        .iter()
        .map(|item| match item.as_str() {
            Some(name) => Ok(name.to_string()),
            None => shape("\"excludeTables\" must be an array of table names"),
        })
        .collect()
}

fn parse_pack_limits(object: &serde_json::Map<String, Value>) -> Result<PackLimits, OptionsError> {
    let mut limits = PackLimits::default();
    let Some(raw) = object.get("limits").filter(|value| !value.is_null()) else {
        return Ok(limits);
    };
    let raw = as_object(raw, "limits")?;
    let number = |key: &str| -> Result<Option<u64>, OptionsError> {
        match raw.get(key).filter(|value| !value.is_null()) {
            None => Ok(None),
            Some(value) => match value.as_u64() {
                Some(number) => Ok(Some(number)),
                None => shape(format!("\"limits.{key}\" must be a non-negative integer")),
            },
        }
    };
    if let Some(value) = number("maxArchiveBytes")? {
        limits.max_archive_bytes = value;
    }
    if let Some(value) = number("maxTotalBytes")? {
        limits.max_total_bytes = value;
    }
    if let Some(value) = number("maxEntryBytes")? {
        limits.max_entry_bytes = value;
    }
    if let Some(value) = number("maxEntries")? {
        limits.max_entries = value as usize;
    }
    Ok(limits)
}

/// Builds [`PackLimits`] from `{ limits?: { maxArchiveBytes?, maxTotalBytes?, maxEntryBytes?, maxEntries? } }`.
pub fn parse_pack_limits_options(value: &Value) -> Result<PackLimits, OptionsError> {
    if value.is_null() {
        return Ok(PackLimits::default());
    }
    parse_pack_limits(as_object(value, "options")?)
}

/// Builds [`PackOptions`] from the JSON document `pack()` takes:
/// `{ indexes?: boolean, excludeTables?: string[], extra?: { [entryName]: string }, limits?, lock? }`.
pub fn parse_pack_options(value: &Value) -> Result<PackOptions, OptionsError> {
    let mut options = PackOptions::default();
    if value.is_null() {
        return Ok(options);
    }
    let object = as_object(value, "options")?;
    options.indexes = optional_bool(object, "indexes")?.unwrap_or(false);
    options.exclude_tables = parse_exclude_tables(object)?;
    if let Some(extra) = object.get("extra").filter(|value| !value.is_null()) {
        for (name, content) in as_object(extra, "extra")? {
            match content.as_str() {
                Some(text) => options.extra.push((name.clone(), text.as_bytes().to_vec())),
                None => return shape(format!("\"extra.{name}\" must be a string")),
            }
        }
    }
    options.limits = parse_pack_limits(object)?;
    if let Some(lock) = object.get("lock").filter(|value| !value.is_null()) {
        options.lock = parse_lock(lock)?;
    }
    Ok(options)
}

/// Builds [`UnpackOptions`] from the JSON document `unpack()` takes: `{ limits?, maxImageBytes?, lock? }`.
pub fn parse_unpack_options(value: &Value) -> Result<UnpackOptions, OptionsError> {
    let mut options = UnpackOptions::default();
    if value.is_null() {
        return Ok(options);
    }
    let object = as_object(value, "options")?;
    options.max_image_bytes = optional_u64(object, "maxImageBytes")?;
    options.limits = parse_pack_limits(object)?;
    if let Some(lock) = object.get("lock").filter(|value| !value.is_null()) {
        options.lock = parse_lock(lock)?;
    }
    Ok(options)
}

fn parse_migration_target(raw: &str) -> Result<LayoutKind, OptionsError> {
    match raw {
        "flat" => Ok(LayoutKind::Flat),
        "nested" => Ok(LayoutKind::Nested),
        other => shape(format!("\"to\" must be \"flat\" or \"nested\" — got \"{other}\"")),
    }
}

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

    use super::*;

    #[test]
    fn an_absent_document_keeps_every_engine_default() {
        let options = parse_open_options(&Value::Null).unwrap();

        assert!(!options.readonly);
        assert_eq!(options.snapshot_every, OpenOptions::default().snapshot_every);
        assert!(options.cypher.is_none());
    }

    #[test]
    fn readonly_snapshot_every_and_lock_reach_the_engine_options() {
        let options = parse_open_options(&json!({
            "readonly": true,
            "snapshotEvery": 5,
            "lock": { "timeoutMs": 0, "retryIntervalMs": 10 },
        }))
        .unwrap();

        assert!(options.readonly);
        assert_eq!(options.snapshot_every, 5);
        let lock = options
            .lock
            .options()
            .expect("an object shape asks for a lock, not for skipping");
        assert_eq!(lock.timeout_ms, Some(0));
        assert_eq!(lock.retry_interval_ms, Some(10));
        assert_eq!(lock.heartbeat_ms, None);
    }

    #[test]
    fn compaction_reaches_the_engine_in_both_the_thresholds_and_the_false_shape() {
        let named = parse_open_options(&json!({
            "compaction": { "deadBytesRatio": 0.5, "minDeadBytes": 200 },
        }))
        .unwrap();
        assert_eq!(
            named.compaction,
            Some(CatalogCompaction::Enabled {
                dead_bytes_ratio: 0.5,
                min_dead_bytes: 200
            })
        );

        let off = parse_open_options(&json!({ "compaction": false })).unwrap();
        assert_eq!(off.compaction, Some(CatalogCompaction::Disabled));

        // NOTE: absent and null both mean "leave it to the catalog", which is not the same as false.
        for document in [json!({}), json!({ "compaction": null })] {
            assert_eq!(parse_open_options(&document).unwrap().compaction, None, "{document}");
        }
    }

    #[test]
    fn a_compaction_ratio_outside_zero_to_one_is_refused_before_the_engine_is_touched() {
        for ratio in [json!(0), json!(1.5), json!("half")] {
            let document = json!({ "compaction": { "deadBytesRatio": ratio, "minDeadBytes": 1 } });
            assert!(parse_open_options(&document).is_err(), "{document}");
        }
        assert!(parse_open_options(&json!({ "compaction": true })).is_err());
    }

    #[test]
    fn lock_false_asks_the_engine_to_skip_the_writer_lock_and_true_asks_for_the_defaults() {
        assert!(matches!(
            parse_open_options(&json!({ "lock": false })).unwrap().lock,
            LockMode::Skip
        ));

        // NOTE: absent and null must stay the locked default — skipping is opt-in only.
        for document in [
            json!({}),
            json!({ "lock": null }),
            json!({ "lock": true }),
            json!({ "lock": {} }),
        ] {
            let lock = parse_open_options(&document).unwrap().lock;
            assert!(matches!(lock, LockMode::Enabled(_)), "{document} must keep the lock");
            assert_eq!(lock.options().unwrap().timeout_ms, None);
        }
    }

    #[test]
    fn a_lock_that_is_neither_a_flag_nor_an_options_object_is_refused_as_a_shape_error() {
        assert_eq!(
            parse_open_options(&json!({ "lock": "off" })).unwrap_err(),
            OptionsError::Shape("\"lock\" must be a boolean or an object of lock options".to_string())
        );
        assert_eq!(
            parse_open_options(&json!({ "lock": { "timeoutMs": -1 } })).unwrap_err(),
            OptionsError::Shape("\"timeoutMs\" must be a non-negative whole number".to_string())
        );
    }

    #[test]
    fn a_base64_keyring_decodes_to_the_exact_bytes_and_keeps_its_write_key() {
        let options = parse_open_options(&json!({
            "cypher": {
                "keys": { "k1": "AAECAw==" },
                "keyId": "k1",
                "onMissingKey": "omit",
            },
        }))
        .unwrap();

        let cypher = options.cypher.unwrap();
        assert_eq!(cypher.keys["k1"], vec![0, 1, 2, 3]);
        assert_eq!(cypher.key_id.as_deref(), Some("k1"));
        assert_eq!(cypher.on_missing_key, Some(MissingKeyPolicy::Omit));
    }

    #[test]
    fn a_hex_keyring_decodes_only_when_key_encoding_says_so() {
        let hex = json!({ "cypher": { "keys": { "k1": "00010203" }, "keyEncoding": "hex" } });
        let options = parse_open_options(&hex).unwrap();
        assert_eq!(options.cypher.unwrap().keys["k1"], vec![0, 1, 2, 3]);

        // NOTE: "00010203" is also valid base64 — the same text decodes to different bytes.
        let as_base64 = json!({ "cypher": { "keys": { "k1": "00010203" } } });
        assert_ne!(
            parse_open_options(&as_base64).unwrap().cypher.unwrap().keys["k1"],
            vec![0, 1, 2, 3]
        );
    }

    #[test]
    fn unusable_key_material_is_reported_as_a_key_error_naming_the_key() {
        let error = parse_open_options(&json!({ "cypher": { "keys": { "k1": "!!!!" } } })).unwrap_err();

        assert_eq!(error, OptionsError::Key("key \"k1\" is not valid base64".to_string()));
    }

    #[test]
    fn a_wrong_typed_option_is_refused_as_a_shape_error_naming_the_member() {
        assert_eq!(
            parse_open_options(&json!({ "readonly": "yes" })).unwrap_err(),
            OptionsError::Shape("\"readonly\" must be a boolean".to_string())
        );
        assert_eq!(
            parse_open_options(&json!({ "snapshotEvery": -1 })).unwrap_err(),
            OptionsError::Shape("\"snapshotEvery\" must be a non-negative whole number".to_string())
        );
        assert!(matches!(
            parse_open_options(&json!({ "cypher": { "keys": {}, "onMissingKey": "shrug" } })).unwrap_err(),
            OptionsError::Shape(_)
        ));
    }

    #[test]
    fn follow_reaches_the_engine_in_all_three_shapes_the_option_takes() {
        assert_eq!(parse_open_options(&json!({})).unwrap().follow, FollowOption::Default);
        assert_eq!(
            parse_open_options(&json!({ "follow": null })).unwrap().follow,
            FollowOption::Default
        );
        assert_eq!(
            parse_open_options(&json!({ "follow": false })).unwrap().follow,
            FollowOption::Off
        );
        assert_eq!(
            parse_open_options(&json!({ "follow": true })).unwrap().follow,
            FollowOption::Every(Duration::ZERO)
        );
        assert_eq!(
            parse_open_options(&json!({ "follow": { "intervalMs": 250 } }))
                .unwrap()
                .follow,
            FollowOption::Every(Duration::from_millis(250))
        );
        assert_eq!(
            parse_open_options(&json!({ "follow": {} })).unwrap().follow,
            FollowOption::Every(Duration::ZERO)
        );
    }

    #[test]
    fn a_follow_that_is_neither_a_flag_nor_an_interval_is_refused_as_a_shape_error() {
        assert!(matches!(
            parse_open_options(&json!({ "follow": "always" })).unwrap_err(),
            OptionsError::Shape(_)
        ));
        assert!(matches!(
            parse_open_options(&json!({ "follow": { "intervalMs": -1 } })).unwrap_err(),
            OptionsError::Shape(_)
        ));
    }

    #[test]
    fn the_layout_triple_reaches_the_engine_unresolved() {
        assert_eq!(parse_open_options(&json!({})).unwrap().layout, LayoutOptions::default());

        let named = parse_open_options(&json!({
            "layout": "nested",
            "metadataFile": "cerebro.json",
        }))
        .unwrap();
        assert_eq!(named.layout.kind, Some(LayoutKind::Nested));
        assert_eq!(named.layout.metadata_file.as_deref(), Some("cerebro.json"));
        assert_eq!(named.layout.manifest, None);

        // NOTE: the two names are passed on as given — the engine, not this, decides they conflict.
        let both = parse_open_options(&json!({ "manifest": "a.json", "metadataFile": "b.json" })).unwrap();
        assert_eq!(both.layout.manifest.as_deref(), Some("a.json"));
        assert_eq!(both.layout.metadata_file.as_deref(), Some("b.json"));
    }

    #[test]
    fn an_unknown_layout_kind_and_a_non_string_file_name_are_refused_as_shape_errors() {
        assert_eq!(
            parse_open_options(&json!({ "layout": "sideways" })).unwrap_err(),
            OptionsError::Shape("\"layout\" must be \"flat\" or \"nested\" — got \"sideways\"".to_string())
        );
        assert_eq!(
            parse_open_options(&json!({ "metadataFile": 7 })).unwrap_err(),
            OptionsError::Shape("\"metadataFile\" must be a string".to_string())
        );
    }

    #[test]
    fn a_migration_with_no_document_moves_into_the_nested_arrangement_under_the_engine_lock() {
        let options = parse_migrate_layout_options(&Value::Null).unwrap();

        assert_eq!(options.to, None, "absent is the engine's own default, not a second one");
        assert!(!options.dry_run);
        assert!(options.metadata_file.is_none() && options.from_metadata_file.is_none());
        assert!(matches!(options.lock, LockMode::Enabled(_)));
    }

    #[test]
    fn both_document_name_pairs_reach_the_engine_unresolved_so_it_makes_the_refusal() {
        let options = parse_migrate_layout_options(&json!({
            "to": "flat",
            "fromMetadataFile": "cerebro.json",
            "fromManifest": "legacy.json",
            "metadataFile": "hayson.json",
            "manifestFile": "old.json",
            "dryRun": true,
        }))
        .unwrap();

        assert_eq!(options.to, Some(LayoutKind::Flat));
        assert_eq!(options.from_metadata_file.as_deref(), Some("cerebro.json"));
        assert_eq!(options.from_manifest.as_deref(), Some("legacy.json"));
        assert_eq!(options.metadata_file.as_deref(), Some("hayson.json"));
        assert_eq!(options.manifest_file.as_deref(), Some("old.json"));
        assert!(options.dry_run);
    }

    #[test]
    fn a_migration_takes_the_same_lock_document_an_open_takes() {
        let options = parse_migrate_layout_options(&json!({ "lock": { "timeoutMs": 0 } })).unwrap();

        assert_eq!(
            options.lock.options().and_then(|lock| lock.timeout_ms),
            Some(0),
            "a host that owns the wait has to be able to ask for a fail-fast attempt"
        );
        let skipped = parse_migrate_layout_options(&json!({ "lock": false })).unwrap();
        assert!(matches!(skipped.lock, LockMode::Skip));
    }

    #[test]
    fn an_unknown_migration_target_is_refused_as_a_shape_error() {
        assert_eq!(
            parse_migrate_layout_options(&json!({ "to": "sideways" })).unwrap_err(),
            OptionsError::Shape("\"to\" must be \"flat\" or \"nested\" — got \"sideways\"".to_string())
        );
        assert_eq!(
            parse_migrate_layout_options(&json!({ "dryRun": "yes" })).unwrap_err(),
            OptionsError::Shape("\"dryRun\" must be a boolean".to_string())
        );
    }

    #[test]
    fn base64_accepts_the_url_safe_alphabet_and_optional_padding() {
        assert_eq!(decode_base64("+/8="), decode_base64("-_8"));
        assert_eq!(decode_base64(""), Some(Vec::new()));
        assert_eq!(decode_base64("QQ"), Some(vec![b'A']));
        assert_eq!(decode_base64("QQ=="), Some(vec![b'A']));
    }

    #[test]
    fn base64_refuses_a_length_and_a_symbol_that_cannot_be_base64() {
        assert_eq!(decode_base64("A"), None);
        assert_eq!(decode_base64("QQ*="), None);
    }

    #[test]
    fn hex_round_trips_both_cases_and_refuses_odd_or_non_hex_input() {
        assert_eq!(decode_hex("DEADbeef"), Some(vec![0xde, 0xad, 0xbe, 0xef]));
        assert_eq!(decode_hex("abc"), None);
        assert_eq!(decode_hex("zz"), None);
    }

    #[test]
    fn pack_options_carry_extra_entries_limits_and_the_lock() {
        let options = parse_pack_options(&json!({
            "indexes": true,
            "excludeTables": ["users"],
            "extra": { "gly.json": "{}" },
            "limits": { "maxEntries": 5, "maxEntryBytes": 10 },
        }))
        .unwrap();

        assert!(options.indexes);
        assert_eq!(options.exclude_tables, vec!["users".to_string()]);
        assert_eq!(options.extra, vec![("gly.json".to_string(), b"{}".to_vec())]);
        assert_eq!(options.limits.max_entries, 5);
        assert_eq!(options.limits.max_entry_bytes, 10);
        assert_eq!(
            options.limits.max_archive_bytes,
            PackLimits::default().max_archive_bytes
        );
        assert!(parse_pack_options(&json!({ "extra": { "a": 1 } })).is_err());
        assert!(parse_unpack_options(&json!({ "limits": { "maxEntries": -1 } })).is_err());
        assert_eq!(
            parse_unpack_options(&Value::Null).unwrap().limits,
            PackLimits::default()
        );
    }
}