haematite 0.7.0

//! Stamp-dispatch matrix (CHUNKING-POLICY.md §4.1 Phase 2).
//!
//! The five-cell contract this pins:
//! - fresh `create` = v2 (stamps `format_version: 2` + the ratified
//!   `chunking_policy` block, and derives the v2 policy);
//! - a v1-stamped directory opens under v1 FOREVER;
//! - an unstamped (legacy) directory opens under v1 FOREVER;
//! - a zero (or otherwise invalid) v2 target refuses with a typed
//!   [`DatabaseError::InvalidChunkingStamp`] — the engine never guesses a target;
//! - a fenced (mid-migration) directory refuses every normal open — writer AND
//!   observer — with [`DatabaseError::MigrationInProgress`].
//!
//! The private-derivation half unit-tests [`derive_policy`] /
//! [`validate_format_version`] directly; the black-box half drives real
//! `Database::create`/`open` and `ReadOnlyDatabase::open` against forged
//! `config.json` files.

use std::error::Error;
use std::fs;
use std::path::Path;

use serde_json::Value;

use super::{
    CHUNKING_V2_ID, ChunkingStamp, DEFAULT_INTERNAL_TARGET_BYTES, DEFAULT_LEAF_TARGET_BYTES,
    DatabaseConfig, DatabaseError, ON_DISK_FORMAT_VERSION, V1_FORMAT_VERSION,
    default_create_policy, derive_policy, read_config, validate_format_version, write_config,
};
use crate::db::{Database, ObserverError, ReadOnlyDatabase};
use crate::tree::TreePolicy;

fn v2_stamp(leaf: u64, internal: u64) -> ChunkingStamp {
    ChunkingStamp {
        chunking: CHUNKING_V2_ID.to_owned(),
        leaf_target_bytes: leaf,
        internal_target_bytes: internal,
    }
}

// ---- private derivation: the policy source of truth ------------------------

#[test]
fn absent_and_v1_stamps_derive_v1_default() -> Result<(), Box<dyn Error>> {
    // Unstamped (legacy) and explicit v1 both open under the count-target rule,
    // regardless of any stray policy block — v2 targets are never consulted.
    let stray = v2_stamp(1, 1);
    for (version, stamp) in [
        (None, None),
        (None, Some(&stray)),
        (Some(V1_FORMAT_VERSION), None),
        (Some(V1_FORMAT_VERSION), Some(&stray)),
    ] {
        let policy = derive_policy(version, stamp)?;
        assert_eq!(
            policy,
            TreePolicy::V1_DEFAULT,
            "format {version:?} must derive V1_DEFAULT"
        );
    }
    Ok(())
}

#[test]
fn v2_stamp_derives_the_stamped_targets() -> Result<(), Box<dyn Error>> {
    let stamp = v2_stamp(4096, 2048);
    let policy = derive_policy(Some(ON_DISK_FORMAT_VERSION), Some(&stamp))?;
    assert_eq!(
        policy,
        TreePolicy::v2(4096, 2048),
        "a v2 stamp must derive exactly its stamped byte targets"
    );
    Ok(())
}

#[test]
fn v2_without_policy_block_refuses() {
    assert!(matches!(
        derive_policy(Some(ON_DISK_FORMAT_VERSION), None),
        Err(DatabaseError::InvalidChunkingStamp(_))
    ));
}

#[test]
fn v2_zero_target_refuses() {
    for stamp in [v2_stamp(0, 48), v2_stamp(64, 0), v2_stamp(0, 0)] {
        assert!(
            matches!(
                derive_policy(Some(ON_DISK_FORMAT_VERSION), Some(&stamp)),
                Err(DatabaseError::InvalidChunkingStamp(_))
            ),
            "zero target must be a typed InvalidChunkingStamp, not a silent default"
        );
    }
}

#[test]
fn v2_unknown_policy_id_refuses() {
    let stamp = ChunkingStamp {
        chunking: "bytes-v9".to_owned(),
        leaf_target_bytes: 64,
        internal_target_bytes: 48,
    };
    assert!(matches!(
        derive_policy(Some(ON_DISK_FORMAT_VERSION), Some(&stamp)),
        Err(DatabaseError::InvalidChunkingStamp(_))
    ));
}

#[test]
fn validate_accepts_v1_and_current_refuses_future_and_zero() {
    assert!(validate_format_version(None).is_ok());
    assert!(validate_format_version(Some(V1_FORMAT_VERSION)).is_ok());
    assert!(validate_format_version(Some(ON_DISK_FORMAT_VERSION)).is_ok());
    assert!(matches!(
        validate_format_version(Some(ON_DISK_FORMAT_VERSION + 1)),
        Err(DatabaseError::FormatVersionTooNew { .. })
    ));
    assert!(matches!(
        validate_format_version(Some(0)),
        Err(DatabaseError::ConfigParse(_))
    ));
}

#[test]
fn default_create_policy_is_the_ratified_v2() {
    assert_eq!(
        default_create_policy(),
        TreePolicy::v2(DEFAULT_LEAF_TARGET_BYTES, DEFAULT_INTERNAL_TARGET_BYTES),
    );
    assert_eq!(DEFAULT_LEAF_TARGET_BYTES, 64 * 1024);
    assert_eq!(DEFAULT_INTERNAL_TARGET_BYTES, 48 * 1024);
}

#[test]
fn write_then_read_round_trips_to_v2_default() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let config = DatabaseConfig {
        data_dir: dir.path().to_path_buf(),
        shard_count: 3,
        distributed: None,
        executor_threads: None,
    };
    write_config(&config)?;
    let opened = read_config(dir.path())?;
    assert_eq!(
        opened.config.shard_count, 3,
        "the caller config round-trips"
    );
    assert_eq!(
        opened.policy,
        default_create_policy(),
        "a freshly stamped directory reads back as the ratified v2 policy"
    );
    Ok(())
}

// ---- black-box: real create/open against forged config.json ----------------

fn create_db(data_dir: &Path, shard_count: usize) -> Result<(), Box<dyn Error>> {
    let db = Database::create(DatabaseConfig {
        data_dir: data_dir.to_path_buf(),
        shard_count,
        distributed: None,
        executor_threads: None,
    })?;
    drop(db);
    Ok(())
}

fn read_config_json(data_dir: &Path) -> Result<Value, Box<dyn Error>> {
    Ok(serde_json::from_slice(&fs::read(
        data_dir.join("config.json"),
    )?)?)
}

/// Forge `config.json` as a foreign binary would have written it — a raw
/// `fs::write`, never the engine's stamping path.
fn forge_config(
    data_dir: &Path,
    mutate: impl FnOnce(&mut serde_json::Map<String, Value>) -> Result<(), Box<dyn Error>>,
) -> Result<(), Box<dyn Error>> {
    let mut value = read_config_json(data_dir)?;
    let object = value
        .as_object_mut()
        .ok_or("config.json must be an object")?;
    mutate(object)?;
    fs::write(
        data_dir.join("config.json"),
        serde_json::to_vec_pretty(&value)?,
    )?;
    Ok(())
}

#[test]
fn fresh_create_stamps_v2_natively() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    create_db(&data_dir, 2)?;

    let value = read_config_json(&data_dir)?;
    assert_eq!(
        value.get("format_version"),
        Some(&Value::from(ON_DISK_FORMAT_VERSION)),
        "create stamps the current (v2) format version"
    );
    let policy = value
        .get("chunking_policy")
        .and_then(Value::as_object)
        .ok_or("create must stamp a chunking_policy block")?;
    assert_eq!(policy.get("chunking"), Some(&Value::from(CHUNKING_V2_ID)));
    assert_eq!(
        policy.get("leaf_target_bytes"),
        Some(&Value::from(DEFAULT_LEAF_TARGET_BYTES))
    );
    assert_eq!(
        policy.get("internal_target_bytes"),
        Some(&Value::from(DEFAULT_INTERNAL_TARGET_BYTES))
    );
    // And it reopens cleanly under the derived v2 policy.
    drop(Database::open(&data_dir)?);
    Ok(())
}

#[test]
fn v1_stamped_directory_opens_forever() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    create_db(&data_dir, 2)?;
    // Forge an explicit v1 directory (as a pre-v2 binary would have written).
    forge_config(&data_dir, |object| {
        object.insert("format_version".to_owned(), Value::from(V1_FORMAT_VERSION));
        object.remove("chunking_policy");
        Ok(())
    })?;
    // Opens under the v1 detector; the derivation is unit-pinned above.
    drop(Database::open(&data_dir)?);
    Ok(())
}

#[test]
fn unstamped_legacy_directory_opens_forever() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    create_db(&data_dir, 2)?;
    forge_config(&data_dir, |object| {
        object.remove("format_version");
        object.remove("chunking_policy");
        Ok(())
    })?;
    drop(Database::open(&data_dir)?);
    Ok(())
}

#[test]
fn open_surfaces_stamp_derived_tree_policy() -> Result<(), Box<dyn Error>> {
    // F1 (Fable key-turn): `Database::tree_policy()` is THE sanctioned policy
    // source for downstream branch-commit/merge consumers. A v2-stamped
    // directory MUST surface its stamped v2 policy (never V1_DEFAULT), and a
    // v1/unstamped directory MUST surface V1_DEFAULT — the exact value each
    // consumer then passes to the public tree/branch APIs.
    let dir = tempfile::tempdir()?;

    let v2_dir = dir.path().join("v2");
    create_db(&v2_dir, 2)?;
    let v2_db = Database::open(&v2_dir)?;
    assert_eq!(
        v2_db.tree_policy(),
        default_create_policy(),
        "a v2-stamped directory surfaces its stamped v2 policy"
    );
    assert!(v2_db.tree_policy().is_v2());
    drop(v2_db);

    let v1_dir = dir.path().join("v1");
    create_db(&v1_dir, 2)?;
    forge_config(&v1_dir, |object| {
        object.insert("format_version".to_owned(), Value::from(V1_FORMAT_VERSION));
        object.remove("chunking_policy");
        Ok(())
    })?;
    let v1_db = Database::open(&v1_dir)?;
    assert_eq!(
        v1_db.tree_policy(),
        TreePolicy::V1_DEFAULT,
        "a v1 directory surfaces V1_DEFAULT"
    );
    drop(v1_db);

    let legacy_dir = dir.path().join("legacy");
    create_db(&legacy_dir, 2)?;
    forge_config(&legacy_dir, |object| {
        object.remove("format_version");
        object.remove("chunking_policy");
        Ok(())
    })?;
    let legacy_db = Database::open(&legacy_dir)?;
    assert_eq!(
        legacy_db.tree_policy(),
        TreePolicy::V1_DEFAULT,
        "an unstamped legacy directory surfaces V1_DEFAULT"
    );
    drop(legacy_db);
    Ok(())
}

#[test]
fn zero_v2_target_refuses_open() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    create_db(&data_dir, 2)?;
    forge_config(&data_dir, |object| {
        object.insert(
            "chunking_policy".to_owned(),
            serde_json::json!({
                "chunking": CHUNKING_V2_ID,
                "leaf_target_bytes": 0,
                "internal_target_bytes": DEFAULT_INTERNAL_TARGET_BYTES,
            }),
        );
        Ok(())
    })?;
    assert!(
        matches!(
            Database::open(&data_dir),
            Err(DatabaseError::InvalidChunkingStamp(_))
        ),
        "a zero leaf target must refuse the open with a typed error"
    );
    Ok(())
}

#[test]
fn migration_fence_refuses_writer_and_observer() -> Result<(), Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let data_dir = dir.path().join("db");
    create_db(&data_dir, 2)?;
    forge_config(&data_dir, |object| {
        object.insert(
            "migration_fence".to_owned(),
            serde_json::json!({
                "source_policy": "count-v1",
                "target_policy": CHUNKING_V2_ID,
                "phase": "forward",
            }),
        );
        Ok(())
    })?;
    assert!(
        matches!(
            Database::open(&data_dir),
            Err(DatabaseError::MigrationInProgress { .. })
        ),
        "a fenced directory must refuse the writer open"
    );
    assert!(
        matches!(
            ReadOnlyDatabase::open(&data_dir),
            Err(ObserverError::Config(
                DatabaseError::MigrationInProgress { .. }
            ))
        ),
        "a fenced directory must refuse the observer open too (§4.2 step 8)"
    );
    Ok(())
}