haematite 0.7.0

//! Shared fixtures for the migration test suites (§9): build a v1-shaped
//! on-disk directory, read a shard's committed root, and compute the fresh-build
//! canonical root the "migrate ≡ fresh-build" witness compares against.
//!
//! The fixture writes v1 trees with the SAME low-level primitives the engine
//! uses (`DiskStore` + `batch_mutate_owned` under `V1_DEFAULT` + `DurableWal`),
//! the analogue of `format_version_tests` forging on-disk config: it constructs
//! genuine v1 on-disk state to migrate.

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

use crate::branch::{
    BranchRefStore, BranchRegistry, CommitDurability, CommitRequest, commit_branch, create_branch,
};
use crate::shard::router::{SHARD_STORE_DIR, SHARD_WAL_FILE};
use crate::store::{DiskStore, MemoryStore, NodeStore};
use crate::tree::{Hash, LeafNode, Node, TreePolicy, batch_mutate_owned};
use crate::wal::{DurableWal, FsyncPolicy, WalRecovery};

use super::super::DatabaseConfig;
use super::super::config::{DEFAULT_INTERNAL_TARGET_BYTES, DEFAULT_LEAF_TARGET_BYTES, finalize_v1};

pub(super) type Entries = Vec<(Vec<u8>, Vec<u8>)>;
pub(super) type Res<T> = Result<T, Box<dyn Error>>;

/// An oversized (≥ 64 KiB leaf target) value — a DETERMINISTIC v2 singleton
/// leaf (§2.1 oversized isolation). A shard of ≥ 2 such entries therefore has a
/// v2 root (a singleton-leaf spine) that DIFFERS from its v1 root (one packed
/// leaf), which is what makes the migration observable. The leading byte is an
/// uppercase ASCII letter, so a value never collides with a TTL/stamp envelope
/// magic and re-inserts as never-expiring under a normal open.
pub(super) fn oversized(fill: u8) -> Vec<u8> {
    vec![b'A' + (fill % 26); DEFAULT_LEAF_TARGET_BYTES as usize]
}

/// `shards` shards, each holding `per_shard` oversized singleton entries with
/// distinct keys.
pub(super) fn oversized_shards(shards: usize, per_shard: usize) -> Vec<Entries> {
    (0..shards)
        .map(|shard| {
            (0..per_shard)
                .map(|index| {
                    let fill = (shard * per_shard + index) as u8;
                    (
                        format!("k{shard:02}-{index:02}").into_bytes(),
                        oversized(fill),
                    )
                })
                .collect()
        })
        .collect()
}

pub(super) fn config_for(data_dir: &Path, shard_count: usize) -> DatabaseConfig {
    DatabaseConfig {
        data_dir: data_dir.to_path_buf(),
        shard_count,
        distributed: None,
        executor_threads: None,
    }
}

fn shard_store_dir(data_dir: &Path, shard_id: usize) -> std::path::PathBuf {
    data_dir
        .join(format!("shard-{shard_id}"))
        .join(SHARD_STORE_DIR)
}

fn shard_wal_path(data_dir: &Path, shard_id: usize) -> std::path::PathBuf {
    data_dir
        .join(format!("shard-{shard_id}"))
        .join(SHARD_WAL_FILE)
}

fn build_root<S: NodeStore + ?Sized>(
    store: &mut S,
    entries: &Entries,
    policy: TreePolicy,
) -> Res<Hash>
where
    S::Error: 'static,
{
    let empty = store.put(&Node::Leaf(LeafNode::new(Vec::new())?))?;
    if entries.is_empty() {
        return Ok(empty);
    }
    let batch: Vec<(Vec<u8>, Option<Vec<u8>>)> = entries
        .iter()
        .cloned()
        .map(|(key, value)| (key, Some(value)))
        .collect();
    Ok(batch_mutate_owned(store, empty, batch, policy)?)
}

fn write_shard_tree(
    data_dir: &Path,
    shard_id: usize,
    entries: &Entries,
    policy: TreePolicy,
) -> Res<Hash> {
    let shard_dir = data_dir.join(format!("shard-{shard_id}"));
    std::fs::create_dir_all(&shard_dir)?;
    let mut store = DiskStore::new(shard_store_dir(data_dir, shard_id))?;
    let root = build_root(&mut store, entries, policy)?;
    store.sync_dirty_dirs()?;
    let mut wal = DurableWal::new(shard_wal_path(data_dir, shard_id), FsyncPolicy::CommitOnly)?;
    wal.commit(root)?;
    Ok(root)
}

/// Build a v1-shaped directory: a v1 `config.json` plus one v1 tree per shard.
/// Returns each shard's v1 committed root.
pub(super) fn build_v1_fixture(data_dir: &Path, shard_entries: &[Entries]) -> Res<Vec<Hash>> {
    std::fs::create_dir_all(data_dir)?;
    finalize_v1(&config_for(data_dir, shard_entries.len()))?;
    let mut roots = Vec::with_capacity(shard_entries.len());
    for (shard_id, entries) in shard_entries.iter().enumerate() {
        roots.push(write_shard_tree(
            data_dir,
            shard_id,
            entries,
            TreePolicy::V1_DEFAULT,
        )?);
    }
    Ok(roots)
}

/// The fresh-build canonical root of `entries` under `policy`, in an isolated
/// `MemoryStore` — the reference the "migrate ≡ fresh-build" witness compares.
pub(super) fn fresh_build(entries: &Entries, policy: TreePolicy) -> Res<Hash> {
    let mut store = MemoryStore::new();
    build_root(&mut store, entries, policy)
}

/// The v2 policy the forward migration stamps (the ratified defaults, §2.3).
pub(super) fn v2_default() -> TreePolicy {
    TreePolicy::v2(DEFAULT_LEAF_TARGET_BYTES, DEFAULT_INTERNAL_TARGET_BYTES)
}

/// Read a shard's current committed root from disk, or `None` if empty.
pub(super) fn read_shard_root(data_dir: &Path, shard_id: usize) -> Res<Option<Hash>> {
    let store = DiskStore::new(shard_store_dir(data_dir, shard_id))?;
    let recovered = WalRecovery::recover_path(shard_wal_path(data_dir, shard_id), &store)?;
    Ok(recovered.committed_root())
}

/// Every shard's current committed root, index-aligned with `0..shard_count`.
pub(super) fn read_all_roots(data_dir: &Path, shard_count: usize) -> Res<Vec<Option<Hash>>> {
    (0..shard_count)
        .map(|shard_id| read_shard_root(data_dir, shard_id))
        .collect()
}

/// Raw `config.json` as a JSON value (for asserting the on-disk envelope).
fn config_json(data_dir: &Path) -> Res<serde_json::Value> {
    let bytes = std::fs::read(data_dir.join("config.json"))?;
    Ok(serde_json::from_slice(&bytes)?)
}

/// The raw `format_version` stamp (an old binary reads exactly this).
pub(super) fn format_version(data_dir: &Path) -> Res<Option<u64>> {
    Ok(config_json(data_dir)?
        .get("format_version")
        .and_then(serde_json::Value::as_u64))
}

/// True when `config.json` carries a (non-null) migration fence.
pub(super) fn is_fenced(data_dir: &Path) -> Res<bool> {
    Ok(config_json(data_dir)?
        .get("migration_fence")
        .is_some_and(|fence| !fence.is_null()))
}

/// The canonical in-dir branch refstore directory.
fn branches_dir(data_dir: &Path) -> std::path::PathBuf {
    data_dir.join("branches")
}

/// Create a durable HBR1 branch `name` on `shard_id` from an EMPTY anchor and
/// commit `puts` under v1, writing its record into `data_dir/branches` and its
/// nodes into the shard store. The committed head is therefore
/// `fresh_build(puts, V1_DEFAULT)`. The shard's store dir must already exist
/// (build the v1 fixture first).
pub(super) fn build_v1_branch(data_dir: &Path, shard_id: usize, puts: &Entries) -> Res<()> {
    let mut store = DiskStore::new(shard_store_dir(data_dir, shard_id))?;
    let anchor = store.put(&Node::Leaf(LeafNode::new(Vec::new())?))?;
    std::fs::create_dir_all(branches_dir(data_dir))?;
    let mut refs = BranchRefStore::open(branches_dir(data_dir))?;
    let registry = BranchRegistry::new();
    let branch = create_branch("main", [(shard_id, anchor)], &mut refs, &registry, 10)?;
    for (key, value) in puts {
        branch.put(shard_id, key.clone(), value.clone())?;
    }
    commit_branch(
        &branch,
        &mut store,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: &[],
            timestamp: 20,
        },
        TreePolicy::V1_DEFAULT,
    )?;
    Ok(())
}

/// Read the committed head of branch `name` for `shard_id` from the on-disk
/// refstore.
pub(super) fn read_branch_head(data_dir: &Path, name: &str, shard_id: usize) -> Res<Hash> {
    let refs = BranchRefStore::open(branches_dir(data_dir))?;
    let record = refs.get(name).ok_or("branch record not found")?;
    let shard = record
        .shards
        .iter()
        .find(|shard| shard.shard_id == shard_id)
        .ok_or("shard not present in branch record")?;
    Ok(shard.head)
}