armdb 0.14.4

use std::fs;
use std::path::Path;

use armdb::armour::Db;
use armdb::{
    CollectionMeta, DbError, FixedConfig, FormatAction, FormatMigration, FormatMigrationPolicy,
    NoHook,
};
use armour_core::GetType;
use tempfile::tempdir;
use zerocopy::{FromBytes, Immutable, IntoBytes};

const NAME: &str = "format_fixed_rows";
const HEADER_SIZE: usize = 4096;
const SLOT_HEADER_SIZE: usize = 8;

#[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
#[repr(C)]
struct V1 {
    value: u32,
}

impl CollectionMeta for V1 {
    type SelfId = [u8; 8];
    const NAME: &'static str = NAME;
    const VERSION: u16 = 1;
}

#[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
#[repr(C)]
struct V2 {
    value: u64,
}

fn widen(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
    let value = u32::from_ne_bytes(old.try_into().expect("V1 width"));
    out.extend_from_slice(&(value as u64).to_ne_bytes());
    Ok(FormatAction::Write)
}

fn drop_even(key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
    if u64::from_be_bytes(key.try_into().expect("key width")).is_multiple_of(2) {
        return Ok(FormatAction::Drop);
    }
    widen(key, old, out)
}

const WIDEN: FormatMigration = FormatMigration::raw_checked(1, &V1::TYPE, &V2::TYPE, widen);
const DROP_EVEN: FormatMigration = FormatMigration::raw(1, drop_even);

impl CollectionMeta for V2 {
    type SelfId = [u8; 8];
    const NAME: &'static str = NAME;
    const VERSION: u16 = 2;
    const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN];
}

#[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
#[repr(C)]
struct V2Drop {
    value: u64,
}

impl CollectionMeta for V2Drop {
    type SelfId = [u8; 8];
    const NAME: &'static str = NAME;
    const VERSION: u16 = 2;
    const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[DROP_EVEN];
}

fn fixed_slot_size(key_len: usize, value_len: usize) -> usize {
    (SLOT_HEADER_SIZE + key_len + value_len + 7) & !7
}

fn slot_for(root: &std::path::Path, key: [u8; 8], value_len: usize) -> (u32, u32) {
    for entry in fs::read_dir(root.join(NAME)).expect("collection directory") {
        let entry = entry.expect("shard entry");
        if !entry.file_name().to_string_lossy().starts_with("shard_") {
            continue;
        }
        let bytes = fs::read(entry.path().join("fixed.data")).expect("fixed data");
        let count = u32::from_le_bytes(bytes[8..12].try_into().expect("header count"));
        let slot_size = fixed_slot_size(8, value_len);
        for slot_id in 0..count {
            let offset = HEADER_SIZE + slot_id as usize * slot_size;
            let meta = u32::from_le_bytes(bytes[offset..offset + 4].try_into().expect("meta"));
            if meta & 3 != 1 {
                continue;
            }
            if bytes[offset + SLOT_HEADER_SIZE..offset + SLOT_HEADER_SIZE + 8] == key {
                return (slot_id, meta);
            }
        }
    }
    panic!("key {key:?} was not found in a fixed slot");
}

fn header_value_len(root: &std::path::Path) -> u16 {
    let shard = fs::read_dir(root.join(NAME))
        .expect("collection directory")
        .filter_map(Result::ok)
        .find(|entry| entry.file_name().to_string_lossy().starts_with("shard_"))
        .expect("fixed shard");
    let bytes = fs::read(shard.path().join("fixed.data")).expect("fixed data");
    u16::from_le_bytes(bytes[14..16].try_into().expect("header value_len"))
}

fn seed(root: &std::path::Path, keys: &[u64]) {
    let db = Db::open_test(root).expect("open seed db");
    let tree = db
        .open_zero_tree_fixed::<V1, { std::mem::size_of::<V1>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("open V1 fixed tree");
    for key in keys {
        tree.put(&key.to_be_bytes(), &V1 { value: *key as u32 })
            .expect("seed row");
    }
    db.close().expect("close seed db");
}

fn seed_empty(root: &std::path::Path) {
    seed(root, &[]);
}

fn snapshot_collection(root: &Path) -> Vec<(String, bool, u64, Vec<u8>)> {
    fn visit(base: &Path, path: &Path, entries: &mut Vec<(String, bool, u64, Vec<u8>)>) {
        let mut children = fs::read_dir(path)
            .expect("snapshot directory")
            .collect::<Result<Vec<_>, _>>()
            .expect("snapshot entries");
        children.sort_by_key(fs::DirEntry::file_name);
        for child in children {
            let child_path = child.path();
            let relative = child_path
                .strip_prefix(base)
                .expect("snapshot relative path")
                .display()
                .to_string();
            let file_type = child.file_type().expect("snapshot file type");
            if file_type.is_dir() {
                entries.push((relative, true, 0, Vec::new()));
                visit(base, &child_path, entries);
            } else {
                let bytes = fs::read(&child_path).expect("snapshot file bytes");
                entries.push((relative, false, bytes.len() as u64, bytes));
            }
        }
    }

    let collection = root.join(NAME);
    let mut entries = Vec::new();
    visit(&collection, &collection, &mut entries);
    entries
}

#[test]
fn format_migration_fixed_zero_tree_changes_size_and_preserves_slots() {
    let dir = tempdir().expect("tempdir");
    let key = 1u64.to_be_bytes();
    seed(dir.path(), &[1]);
    let before = slot_for(dir.path(), key, std::mem::size_of::<V1>());

    let db = Db::open_test(dir.path()).expect("reopen db");
    let tree = db
        .open_zero_tree_fixed::<V2, { std::mem::size_of::<V2>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("format migration opens V2 fixed tree");
    assert_eq!(tree.get(&key), Some(V2 { value: 1 }));
    db.close().expect("close migrated db");

    assert_eq!(
        header_value_len(dir.path()),
        std::mem::size_of::<V2>() as u16
    );
    assert_eq!(slot_for(dir.path(), key, std::mem::size_of::<V2>()), before);
}

#[test]
fn format_migration_fixed_zero_map_changes_size_and_bitmap() {
    // The FixedStore physical layout is shared by maps and trees; this test
    // exercises the map opener once Task 5 dispatches the same rewrite pass.
    let dir = tempdir().expect("tempdir");
    seed(dir.path(), &[2]);
    let key = 2u64.to_be_bytes();
    let before = slot_for(dir.path(), key, std::mem::size_of::<V1>());

    let db = Db::open_test(dir.path()).expect("reopen db");
    let map = db
        .open_zero_map_fixed::<V2, { std::mem::size_of::<V2>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("format migration opens V2 fixed map");
    assert_eq!(map.get(&key), Some(V2 { value: 2 }));
    db.close().expect("close migrated db");

    assert_eq!(slot_for(dir.path(), key, std::mem::size_of::<V2>()), before);
}

#[test]
fn format_migration_fixed_drop_leaves_slot_free() {
    let dir = tempdir().expect("tempdir");
    let dropped = 2u64.to_be_bytes();
    let survivor = 3u64.to_be_bytes();
    seed(dir.path(), &[2, 3]);
    let dropped_slot = slot_for(dir.path(), dropped, std::mem::size_of::<V1>()).0;
    let survivor_before = slot_for(dir.path(), survivor, std::mem::size_of::<V1>());

    let db = Db::open_test(dir.path()).expect("reopen db");
    let tree = db
        .open_zero_tree_fixed::<V2Drop, { std::mem::size_of::<V2>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("format migration opens V2 fixed tree");
    assert!(tree.get(&dropped).is_none());
    assert_eq!(tree.get(&survivor), Some(V2Drop { value: 3 }));
    tree.put(&4u64.to_be_bytes(), &V2Drop { value: 4 })
        .expect("reuse freed slot");
    db.close().expect("close migrated db");

    assert_eq!(
        slot_for(dir.path(), 4u64.to_be_bytes(), std::mem::size_of::<V2>()).0,
        dropped_slot
    );
    assert_eq!(
        slot_for(dir.path(), survivor, std::mem::size_of::<V2>()),
        survivor_before
    );
}

#[test]
fn format_migration_fixed_keeps_meta_version() {
    let dir = tempdir().expect("tempdir");
    let key = 9u64.to_be_bytes();
    seed(dir.path(), &[9]);
    let before = slot_for(dir.path(), key, std::mem::size_of::<V1>());

    let db = Db::open_test(dir.path()).expect("reopen db");
    let _tree = db
        .open_zero_tree_fixed::<V2, { std::mem::size_of::<V2>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("format migration opens V2 fixed tree");
    db.close().expect("close migrated db");

    assert_eq!(
        slot_for(dir.path(), key, std::mem::size_of::<V2>()).1,
        before.1
    );
}

#[test]
fn format_migration_refuse_fixed_writes_nothing() {
    let dir = tempdir().unwrap();
    seed(dir.path(), &[1]);
    let before = snapshot_collection(dir.path());
    let db = Db::open_test(dir.path()).unwrap();
    let mut config = FixedConfig::test();
    config.format_migration = FormatMigrationPolicy::Refuse;
    let result =
        db.open_zero_tree_fixed::<V2, { std::mem::size_of::<V2>() }, _>(config, NoHook, &[]);
    assert!(matches!(
        result,
        Err(armdb::DbError::FormatMigrationRequired {
            collection: NAME,
            from: 1,
            to: 2
        })
    ));
    assert_eq!(snapshot_collection(dir.path()), before);
}

#[test]
fn explicit_format_migration_fixed_bypasses_refuse_policy() {
    let dir = tempdir().unwrap();
    seed(dir.path(), &[1]);
    let db = Db::open_test(dir.path()).unwrap();
    let mut config = FixedConfig::test();
    config.format_migration = FormatMigrationPolicy::Refuse;

    let report = db
        .migrate_format_fixed::<V2, { std::mem::size_of::<V2>() }>(&config)
        .expect("explicit executor must bypass the managed-open Refuse policy");

    assert_eq!((report.from, report.to), (1, 2));
}

#[test]
fn format_migration_explicit_fixed_empty_plan_invalid_config_writes_nothing() {
    let dir = tempdir().expect("tempdir");
    seed(dir.path(), &[1]);
    let before = snapshot_collection(dir.path());
    let db = Db::open_test(dir.path()).expect("reopen db");
    let mut invalid = FixedConfig::test();
    invalid.shard_count = 0;
    assert!(matches!(
        db.migrate_format_fixed::<V1, { std::mem::size_of::<V1>() }>(&invalid),
        Err(DbError::Config("shard_count must be 1..=255"))
    ));
    assert_eq!(
        snapshot_collection(dir.path()),
        before,
        "invalid config must be rejected before the empty-plan return"
    );
}

#[test]
fn format_migration_fixed_auto_wrong_shard_count_writes_nothing() {
    let dir = tempdir().expect("tempdir");
    seed(dir.path(), &[1]);
    let db = Db::open_test(dir.path()).expect("reopen db");
    let before = snapshot_collection(dir.path());
    let mut wrong = FixedConfig::test();
    wrong.shard_count = 1;
    let err =
        match db.open_zero_tree_fixed::<V2, { std::mem::size_of::<V2>() }, _>(wrong, NoHook, &[]) {
            Err(err) => err,
            Ok(_) => panic!("persisted shard count must be validated before migration"),
        };
    assert!(
        matches!(err, DbError::FormatMismatch(ref message) if message.contains("shard_count mismatch")),
        "{err}"
    );
    assert_eq!(
        snapshot_collection(dir.path()),
        before,
        "automatic migration mutated a collection before immutable-config validation"
    );
    assert!(!db.db_info().collections[NAME].projections_dirty);
}

#[test]
fn format_migration_fixed_explicit_wrong_shard_count_writes_nothing() {
    let dir = tempdir().expect("tempdir");
    seed(dir.path(), &[1]);
    let db = Db::open_test(dir.path()).expect("reopen db");
    let before = snapshot_collection(dir.path());
    let mut wrong = FixedConfig::test();
    wrong.shard_count = 1;
    let err = db
        .migrate_format_fixed::<V2, { std::mem::size_of::<V2>() }>(&wrong)
        .expect_err("persisted shard count must be validated before migration");
    assert!(
        matches!(err, DbError::FormatMismatch(ref message) if message.contains("shard_count mismatch")),
        "{err}"
    );
    assert_eq!(
        snapshot_collection(dir.path()),
        before,
        "explicit migration mutated a collection before immutable-config validation"
    );
    assert!(!db.db_info().collections[NAME].projections_dirty);
}

#[test]
fn format_migration_fixed_auto_rejects_a_missing_configured_shard_without_writes() {
    let dir = tempdir().expect("tempdir");
    seed_empty(dir.path());
    fs::remove_dir_all(dir.path().join(NAME).join("shard_001")).expect("remove shard 1");
    let before = snapshot_collection(dir.path());

    let db = Db::open_test(dir.path()).expect("reopen db");
    let error = match db.open_zero_tree_fixed::<V2, { std::mem::size_of::<V2>() }, _>(
        FixedConfig::test(),
        NoHook,
        &[],
    ) {
        Err(error) => error,
        Ok(_) => panic!("a missing configured FixedStore shard must fail closed"),
    };

    assert!(matches!(error, DbError::FormatMismatch(_)), "{error}");
    assert_eq!(snapshot_collection(dir.path()), before);
    assert!(!db.db_info().collections[NAME].projections_dirty);
}

#[test]
fn format_migration_fixed_explicit_rejects_a_missing_configured_shard_without_writes() {
    let dir = tempdir().expect("tempdir");
    seed_empty(dir.path());
    fs::remove_dir_all(dir.path().join(NAME).join("shard_001")).expect("remove shard 1");
    let before = snapshot_collection(dir.path());

    let db = Db::open_test(dir.path()).expect("reopen db");
    let error = db
        .migrate_format_fixed::<V2, { std::mem::size_of::<V2>() }>(&FixedConfig::test())
        .expect_err("a missing configured FixedStore shard must fail closed");

    assert!(matches!(error, DbError::FormatMismatch(_)), "{error}");
    assert_eq!(snapshot_collection(dir.path()), before);
    assert!(!db.db_info().collections[NAME].projections_dirty);
}

#[cfg(feature = "test-faults")]
#[test]
fn format_migration_replay_fixed_put_preserves_the_fixed_version_tag() {
    let dir = tempdir().expect("tempdir");
    let key = 11u64.to_be_bytes();
    {
        let db = Db::open_test(dir.path()).expect("open V1 db");
        let tree = db
            .open_zero_tree_fixed::<V1, { std::mem::size_of::<V1>() }, _>(
                FixedConfig::test(),
                NoHook,
                &[],
            )
            .expect("open V1 fixed tree");
        Db::tx_debug_skip_apply(true);
        db.tx()
            .put(&tree, &key, &V1 { value: 11 })
            .expect("stage fixed put")
            .durable()
            .commit()
            .expect("commit into txlog");
        Db::tx_debug_skip_apply(false);
        db.close().expect("close V1 db");
    }

    let db = Db::open_test(dir.path()).expect("reopen V2 db");
    let tree = db
        .open_zero_tree_fixed::<V2, { std::mem::size_of::<V2>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("fixed replay transforms the pending V1 Put");
    assert_eq!(tree.get(&key), Some(V2 { value: 11 }));
}

#[cfg(feature = "test-faults")]
#[test]
fn explicit_format_migration_fixed_replays_pending_old_put_and_delete() {
    let dir = tempdir().expect("tempdir");
    let put_key = 12u64.to_be_bytes();
    let delete_key = 13u64.to_be_bytes();
    {
        let db = Db::open_test(dir.path()).expect("open V1 db");
        let tree = db
            .open_zero_tree_fixed::<V1, { std::mem::size_of::<V1>() }, _>(
                FixedConfig::test(),
                NoHook,
                &[],
            )
            .expect("open V1 fixed tree");
        tree.put(&delete_key, &V1 { value: 13 })
            .expect("seed row to delete");
        Db::tx_debug_skip_apply(true);
        db.tx()
            .put(&tree, &put_key, &V1 { value: 12 })
            .expect("stage fixed Put")
            .delete(&tree, &delete_key)
            .expect("stage fixed Delete")
            .durable()
            .commit()
            .expect("commit fixed old operations");
        Db::tx_debug_skip_apply(false);
        db.close().expect("close V1 db");
    }

    let db = Db::open_test(dir.path()).expect("reopen for explicit fixed migration");
    db.migrate_format_fixed::<V2, { std::mem::size_of::<V2>() }>(&FixedConfig::test())
        .expect("install V2 fixed physical format without replay");
    let tree = db
        .open_zero_tree_fixed::<V2, { std::mem::size_of::<V2>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("reconstruct fixed replay-only V1 to V2 plan");
    assert_eq!(tree.get(&put_key), Some(V2 { value: 12 }));
    assert!(tree.get(&delete_key).is_none());
}

// Rewrite → value → rewrite on FixedStore: the value step leaves `format.v` at
// 2 while `db.info` reaches 3.
const LAG_NAME: &str = "format_fixed_value_lag";

mod lag_v1 {
    use super::*;
    #[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
    #[repr(C)]
    pub struct Row {
        pub value: u32,
    }
    impl CollectionMeta for Row {
        type SelfId = [u8; 8];
        const NAME: &'static str = LAG_NAME;
        const VERSION: u16 = 1;
    }
}

mod lag_v2 {
    use super::*;
    #[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
    #[repr(C)]
    pub struct Row {
        pub value: u64,
    }
    pub const WIDEN: FormatMigration =
        FormatMigration::raw_checked(1, &super::lag_v1::Row::TYPE, &Row::TYPE, super::widen);
    impl CollectionMeta for Row {
        type SelfId = [u8; 8];
        const NAME: &'static str = LAG_NAME;
        const VERSION: u16 = 2;
        const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[WIDEN];
    }
}

mod lag_v3 {
    use super::*;
    #[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
    #[repr(C)]
    pub struct Row {
        pub value: u64,
    }
    impl CollectionMeta for Row {
        type SelfId = [u8; 8];
        const NAME: &'static str = LAG_NAME;
        const VERSION: u16 = 3;
        const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[super::lag_v2::WIDEN];
    }
    pub fn bump(_key: &[u8; 8], old: &Row) -> armdb::MigrateAction<Row> {
        armdb::MigrateAction::Update(Row {
            value: old.value + 1000,
        })
    }
}

mod lag_v4 {
    use super::*;
    #[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
    #[repr(C)]
    pub struct Row {
        pub value: u64,
        pub tag: u64,
    }
    fn extend(_key: &[u8], old: &[u8], out: &mut Vec<u8>) -> armdb::DbResult<FormatAction> {
        out.extend_from_slice(old);
        out.extend_from_slice(&7u64.to_ne_bytes());
        Ok(FormatAction::Write)
    }
    pub const EXTEND: FormatMigration =
        FormatMigration::raw_checked(3, &super::lag_v3::Row::TYPE, &Row::TYPE, extend);
    impl CollectionMeta for Row {
        type SelfId = [u8; 8];
        const NAME: &'static str = LAG_NAME;
        const VERSION: u16 = 4;
        const FORMAT_MIGRATIONS: &'static [FormatMigration] = &[EXTEND];
    }
}

/// The 1 → 2 rewrite is still declared next to the new one.
mod lag_v4_kept {
    use super::*;
    #[derive(Clone, Copy, Debug, Eq, FromBytes, GetType, Immutable, IntoBytes, PartialEq)]
    #[repr(C)]
    pub struct Row {
        pub value: u64,
        pub tag: u64,
    }
    impl CollectionMeta for Row {
        type SelfId = [u8; 8];
        const NAME: &'static str = LAG_NAME;
        const VERSION: u16 = 4;
        const FORMAT_MIGRATIONS: &'static [FormatMigration] =
            &[super::lag_v2::WIDEN, super::lag_v4::EXTEND];
    }
}

#[test]
fn fixed_retained_chain_reopens_a_collection_created_at_the_current_version() {
    let dir = tempdir().expect("tempdir");
    let key = 4u64.to_be_bytes();
    for open in ["create", "reopen"] {
        let db = Db::open_test(dir.path()).expect("open v4 db");
        let tree = db
            .open_zero_tree_fixed::<lag_v4_kept::Row, { std::mem::size_of::<lag_v4_kept::Row>() }, _>(
                FixedConfig::test(),
                NoHook,
                &[],
            )
            .unwrap_or_else(|error| panic!("{open}: {error}"));
        if open == "create" {
            tree.put(&key, &lag_v4_kept::Row { value: 4, tag: 1 })
                .expect("seed v4 row");
        }
        assert_eq!(
            tree.get(&key),
            Some(lag_v4_kept::Row { value: 4, tag: 1 }),
            "{open}"
        );
        drop(tree);
        db.close().expect("close v4 db");
    }
}

#[test]
fn fixed_rewrite_after_value_migration_accepts_a_lagging_format_marker() {
    let dir = tempdir().expect("tempdir");
    let key = 3u64.to_be_bytes();
    {
        let db = Db::open_test(dir.path()).expect("open v1 db");
        let tree = db
            .open_zero_tree_fixed::<lag_v1::Row, { std::mem::size_of::<lag_v1::Row>() }, _>(
                FixedConfig::test(),
                NoHook,
                &[],
            )
            .expect("open v1 fixed tree");
        tree.put(&key, &lag_v1::Row { value: 3 }).expect("seed row");
        db.close().expect("close v1 db");
    }
    {
        let db = Db::open_test(dir.path()).expect("open v2 db");
        db.open_zero_tree_fixed::<lag_v2::Row, { std::mem::size_of::<lag_v2::Row>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("checked rewrite to v2");
        db.close().expect("close v2 db");
    }
    {
        let db = Db::open_test(dir.path()).expect("open v3 db");
        let migrations: &[armdb::armour::TypedMigration<[u8; 8], lag_v3::Row>] =
            &[(2, lag_v3::bump)];
        db.open_zero_tree_fixed::<lag_v3::Row, { std::mem::size_of::<lag_v3::Row>() }, _>(
            FixedConfig::test(),
            NoHook,
            migrations,
        )
        .expect("value migration to v3");
        db.close().expect("close v3 db");
    }

    let db = Db::open_test(dir.path()).expect("open v4 db");
    let tree = db
        .open_zero_tree_fixed::<lag_v4::Row, { std::mem::size_of::<lag_v4::Row>() }, _>(
            FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("format.v 2 is equivalent to version 3 through the value step");
    assert_eq!(
        tree.get(&key),
        Some(lag_v4::Row {
            value: 1003,
            tag: 7
        })
    );
}