armdb 0.2.0

//! End-to-end integration tests for the armdb→armour handler layer.
//!
//! Covers: RPC error mapping, Take, apply_batch atomicity, Upsert
//! concurrency contracts, count(exact), contains, EntryLen, and Db::close
//! with mixed collection durability backends.
//!
//! All tests call the `RpcHandler` trait methods directly (in-process), which
//! is equivalent to what the network dispatch layer does after decoding a
//! frame.  No TCP/UDS server is spun up here.
#![cfg(all(
    feature = "typed-tree",
    feature = "armour",
    feature = "rpc",
    feature = "rapira-codec"
))]

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use armdb::armour::{Db, RpcHandler};
use armdb::{CollectionMeta, Config, DbError, FixedConfig, NoHook, RapiraCodec};
use armour_core::GetType;
use armour_rpc::protocol::UpsertKey;
use rapira::Rapira;
use tempfile::tempdir;

// ---------------------------------------------------------------------------
// Test fixtures
// ---------------------------------------------------------------------------

/// Simple domain type used across most tests.
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct TestItem {
    value: u64,
}

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

/// Zero-copy type stored in a ZeroTree (fixed 8-byte values = u64).
#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    GetType,
    zerocopy::FromBytes,
    zerocopy::IntoBytes,
    zerocopy::Immutable,
)]
#[repr(C)]
struct ZeroVal(u64);

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

/// Fixed-backend zero-copy type.
#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    GetType,
    zerocopy::FromBytes,
    zerocopy::IntoBytes,
    zerocopy::Immutable,
)]
#[repr(C)]
struct FixedVal(u64);

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

// ---------------------------------------------------------------------------
// Harness helpers
// ---------------------------------------------------------------------------

fn k(id: u64) -> [u8; 8] {
    id.to_be_bytes()
}

fn k_bytes(id: u64) -> Vec<u8> {
    k(id).to_vec()
}

/// Encode a TestItem via RapiraCodec for use as handler value bytes.
fn encode_item(item: &TestItem) -> Vec<u8> {
    let codec = RapiraCodec;
    let mut buf = Vec::new();
    armdb::Codec::encode_to(&codec, item, &mut buf).unwrap();
    buf
}

/// Returns the single registered handler for the named collection.
/// Panics if not found (programming error in the test setup).
fn get_handler(db: &Db, name: &str) -> Arc<dyn RpcHandler> {
    let hash = xxhash_rust::xxh3::xxh3_64(name.as_bytes());
    let tree_map = db.build_tree_map();
    tree_map
        .get(&hash)
        .cloned()
        .unwrap_or_else(|| panic!("handler for '{name}' not registered"))
}

// ---------------------------------------------------------------------------
// Task 6.2 — RPC error code mapping E2E
// ---------------------------------------------------------------------------

/// Exercises the semantic error codes produced by the RPC handler layer:
///
/// - Upsert update-only (Some(true)) on absent key  → 404 KeyNotFound
/// - Upsert insert-only (Some(false)) on existing   → 409 KeyExists
/// - Remove(soft=true)  → 501 NotImplemented
/// - Take(soft=true)    → 501 NotImplemented
/// - Wrong-length key   → 404 (check_key_len returns KeyNotFound)
///
/// Note: TypedTree::delete is idempotent — remove(soft=false) on an absent
/// key returns Ok(()) because the underlying tree returns Ok(None) and the
/// handler discards the Option.  This is explicitly verified at the end.
#[test]
fn rpc_error_codes_are_semantic() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open tree");

    let h = get_handler(&db, TestItem::NAME);

    // Upsert update-only (Some(true)) on absent key → 404
    let val = encode_item(&TestItem { value: 42 });
    let err = h
        .upsert(UpsertKey::Provided(k_bytes(1)), Some(true), val.clone())
        .expect_err("expected KeyNotFound");
    assert_eq!(err.status_code(), 404, "upsert update-only absent → 404");

    // Insert the key first (unconditional).
    h.upsert(UpsertKey::Provided(k_bytes(1)), None, val.clone())
        .expect("upsert unconditional");

    // Upsert insert-only (Some(false)) on existing key → 409
    let err = h
        .upsert(UpsertKey::Provided(k_bytes(1)), Some(false), val.clone())
        .expect_err("expected KeyExists");
    assert_eq!(err.status_code(), 409, "upsert insert-only existing → 409");

    // Remove(soft=true) → 501 NotImplemented
    let err = h
        .remove(&k_bytes(1), true)
        .expect_err("expected NotImplemented");
    assert_eq!(err.status_code(), 501, "remove soft → 501");

    // Take(soft=true) → 501 NotImplemented
    let err = h
        .take(&k_bytes(1), true)
        .expect_err("expected NotImplemented");
    assert_eq!(err.status_code(), 501, "take soft → 501");

    // Wrong-length key → 404 (check_key_len returns KeyNotFound)
    let short_key = vec![0u8; 4]; // [u8;8] expected, 4 given
    let err = h.contains(&short_key).expect_err("expected KeyNotFound");
    assert_eq!(err.status_code(), 404, "wrong-length key → 404");

    // remove(soft=false) on absent key is idempotent → Ok(())
    h.remove(&k_bytes(999), false)
        .expect("remove absent must be idempotent (Ok)");
}

// ---------------------------------------------------------------------------
// Task 6.3 — Take E2E
// ---------------------------------------------------------------------------

/// Put 1 entry; Take(soft=false) → returns the bytes, then Contains → false.
#[test]
fn take_returns_value_and_removes_key() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open tree");

    let h = get_handler(&db, TestItem::NAME);
    let val = encode_item(&TestItem { value: 99 });

    h.upsert(UpsertKey::Provided(k_bytes(10)), None, val.clone())
        .expect("insert");

    // take must return Some(bytes)
    let taken = h.take(&k_bytes(10), false).expect("take");
    assert!(taken.is_some(), "take should return the value bytes");

    // key must be gone
    let present = h.contains(&k_bytes(10)).expect("contains");
    assert!(!present, "key must be absent after take");

    // count must be 0
    let cnt = h.count(false).expect("count");
    assert_eq!(cnt, 0, "count must be 0 after take");
}

/// Take(soft=true) → Err with status 501.
#[test]
fn take_soft_returns_501() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open tree");

    let h = get_handler(&db, TestItem::NAME);

    let err = h
        .take(&k_bytes(1), true)
        .expect_err("expected NotImplemented");
    assert_eq!(err.status_code(), 501, "take soft=true → 501");
}

// ---------------------------------------------------------------------------
// Task 6.4 — apply_batch atomicity
// ---------------------------------------------------------------------------

/// Batch containing an item with wrong key length must be fully rejected —
/// no partial writes.
///
/// ZeroTree<[u8;8], 8, ZeroVal> is ideal for this: a value with wrong byte
/// length (V!=8) triggers the pre-validation abort path in apply_batch.
#[test]
fn apply_batch_rejects_bad_key_length_whole_batch() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_zero_tree::<ZeroVal, 8, _>(Config::test(), NoHook, &[])
        .expect("open zero tree");

    let h = get_handler(&db, ZeroVal::NAME);

    // k1 and k3 are valid 8-byte keys; k2 has 4-byte key (wrong length).
    let valid_val: Vec<u8> = vec![0u8; 8];
    let batch: Vec<(Vec<u8>, Option<Vec<u8>>)> = vec![
        (k_bytes(1), Some(valid_val.clone())),
        (vec![0u8; 4], Some(valid_val.clone())), // bad key length
        (k_bytes(3), Some(valid_val.clone())),
    ];

    let result = h.apply_batch(batch);
    assert!(
        result.is_err(),
        "batch with invalid key length must be rejected"
    );
    match result.unwrap_err() {
        DbError::Client(_) => {}
        other => panic!("expected Client error, got: {other:?}"),
    }

    // No partial writes: tree must be empty.
    assert_eq!(h.count(true).expect("count"), 0, "no partial writes");
}

/// Batch with invalid value length must be rejected atomically (ZeroTree V=8).
#[test]
fn apply_batch_rejects_bad_value_length_whole_batch() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    // Use a fresh named type to avoid sharing state with the above test.
    // We reuse ZeroVal NAME since the directory is fresh.
    let _tree = db
        .open_zero_tree::<ZeroVal, 8, _>(Config::test(), NoHook, &[])
        .expect("open zero tree");

    let h = get_handler(&db, ZeroVal::NAME);

    let valid_val: Vec<u8> = vec![0u8; 8];
    let bad_val: Vec<u8> = vec![0u8; 5]; // wrong length for V=8
    let batch: Vec<(Vec<u8>, Option<Vec<u8>>)> = vec![
        (k_bytes(1), Some(valid_val.clone())),
        (k_bytes(2), Some(bad_val)),
        (k_bytes(3), Some(valid_val.clone())),
    ];

    let result = h.apply_batch(batch);
    assert!(
        result.is_err(),
        "batch with invalid value length must be rejected"
    );
    match result.unwrap_err() {
        DbError::Client(_) => {}
        other => panic!("expected Client error, got: {other:?}"),
    }

    // No partial writes.
    assert_eq!(h.count(true).expect("count"), 0, "no partial writes");
}

// ---------------------------------------------------------------------------
// Task 6.5 — Upsert concurrent contracts
// ---------------------------------------------------------------------------

/// N threads all try to insert the same key with insert-only (Some(false)).
/// Exactly one must succeed (Ok), and the rest must get KeyExists (409).
#[test]
fn upsert_insert_only_concurrent_at_most_one_wins() {
    let dir = tempdir().expect("tmp");
    let db = Arc::new(Db::open(dir.path()).expect("open"));
    let _tree = db
        .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open tree");

    let ok_count = Arc::new(AtomicUsize::new(0));
    let conflict_count = Arc::new(AtomicUsize::new(0));

    let n_threads = 8;
    let val = encode_item(&TestItem { value: 1 });

    let handles: Vec<_> = (0..n_threads)
        .map(|_| {
            let db = Arc::clone(&db);
            let val = val.clone();
            let ok_count = Arc::clone(&ok_count);
            let conflict_count = Arc::clone(&conflict_count);
            std::thread::spawn(move || {
                let h = get_handler(&db, TestItem::NAME);
                let result = h.upsert(UpsertKey::Provided(k_bytes(42)), Some(false), val.clone());
                match result {
                    Ok(_) => {
                        ok_count.fetch_add(1, Ordering::Relaxed);
                    }
                    Err(DbError::KeyExists) => {
                        conflict_count.fetch_add(1, Ordering::Relaxed);
                    }
                    Err(e) => panic!("unexpected error: {e:?}"),
                }
            })
        })
        .collect();

    for h in handles {
        h.join().expect("thread panicked");
    }

    let oks = ok_count.load(Ordering::Relaxed);
    let conflicts = conflict_count.load(Ordering::Relaxed);

    assert_eq!(oks, 1, "exactly one insert must succeed; got {oks}");
    assert_eq!(
        conflicts,
        n_threads - 1,
        "remaining {} must get KeyExists",
        n_threads - 1
    );
}

/// Upsert update-only (Some(true)) on a key that does not exist → 404.
#[test]
fn upsert_update_only_missing_key_returns_404() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open tree");

    let h = get_handler(&db, TestItem::NAME);
    let val = encode_item(&TestItem { value: 7 });

    let err = h
        .upsert(UpsertKey::Provided(k_bytes(99)), Some(true), val)
        .expect_err("expected KeyNotFound");
    assert_eq!(err.status_code(), 404, "update-only on missing key → 404");
}

// ---------------------------------------------------------------------------
// Task 6.6 — count / contains / entry_len wire shape
// ---------------------------------------------------------------------------

/// Insert 50 items; count(false) and count(true) must both equal 50.
#[test]
fn count_exact_matches_iter_count_for_tree() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open tree");

    let h = get_handler(&db, TestItem::NAME);

    const N: u64 = 50;
    for i in 0..N {
        let val = encode_item(&TestItem { value: i });
        h.upsert(UpsertKey::Provided(k_bytes(i)), None, val)
            .expect("insert");
    }

    let approx = h.count(false).expect("count approx");
    let exact = h.count(true).expect("count exact");

    assert_eq!(approx, N, "count(false) must equal {N}");
    assert_eq!(exact, N, "count(true) must equal {N}");
}

/// contains returns true for a present key and false for an absent key.
#[test]
fn contains_returns_correct_bool() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open tree");

    let h = get_handler(&db, TestItem::NAME);

    let val = encode_item(&TestItem { value: 5 });
    h.upsert(UpsertKey::Provided(k_bytes(5)), None, val)
        .expect("insert");

    let present = h.contains(&k_bytes(5)).expect("contains present");
    assert!(present, "contains must be true for inserted key");

    let absent = h.contains(&k_bytes(99)).expect("contains absent");
    assert!(!absent, "contains must be false for missing key");
}

/// entry_len returns Some(expected_size) for a present key and None for absent.
///
/// For TypedTree, entry_len computes `codec.size(&v)`. A TestItem { value: u64 }
/// encoded by RapiraCodec is exactly 8 bytes (u64 LE).
#[test]
fn entry_len_returns_size_for_typed_tree() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open tree");

    let h = get_handler(&db, TestItem::NAME);

    let item = TestItem { value: 42 };
    let val = encode_item(&item);
    let expected_len = val.len() as u32;

    h.upsert(UpsertKey::Provided(k_bytes(1)), None, val)
        .expect("insert");

    let len = h.entry_len(&k_bytes(1)).expect("entry_len present");
    assert_eq!(len, Some(expected_len), "entry_len must equal encoded size");

    let absent = h.entry_len(&k_bytes(999)).expect("entry_len absent");
    assert_eq!(absent, None, "entry_len must be None for absent key");
}

/// ZeroTree entry_len returns Some(V) for present keys.
#[test]
fn entry_len_zero_tree_returns_v() {
    let dir = tempdir().expect("tmp");
    let db = Db::open(dir.path()).expect("open");
    let _tree = db
        .open_zero_tree::<ZeroVal, 8, _>(Config::test(), NoHook, &[])
        .expect("open zero tree");

    let h = get_handler(&db, ZeroVal::NAME);

    // Insert a zero value.
    let zero_val: Vec<u8> = vec![0xABu8; 8];
    h.upsert(UpsertKey::Provided(k_bytes(7)), None, zero_val)
        .expect("insert");

    let len = h.entry_len(&k_bytes(7)).expect("entry_len");
    assert_eq!(len, Some(8), "ZeroTree entry_len must be Some(V=8)");

    let absent = h.entry_len(&k_bytes(999)).expect("entry_len absent");
    assert_eq!(absent, None, "absent key → None");
}

// ---------------------------------------------------------------------------
// Task 6.7 — Db::close flushes mixed collection durability backends
// ---------------------------------------------------------------------------

/// Open a Db with:
///   - TypedTree (Bitcask)
///   - ZeroTree (Bitcask)
///   - ZeroTree (FixedStore)
///
/// Insert distinctive entries in each, close, drop, reopen, assert all present.
///
/// Note: VarTypedTree is not included here because the `var-collections`
/// feature may not be active in all configurations; the test is guarded only
/// by `typed-tree` + `armour` + `rapira-codec` + `rpc`.
#[test]
fn db_close_flushes_mixed_collection_types() {
    let dir = tempdir().expect("tmp");

    // Phase 1: open all collection types, write entries, close.
    {
        let db = Db::open(dir.path()).expect("open");

        let typed = db
            .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
            .expect("open typed tree");
        let zero = db
            .open_zero_tree::<ZeroVal, 8, _>(Config::test(), NoHook, &[])
            .expect("open zero tree");
        let fixed = db
            .open_zero_tree_fixed::<FixedVal, 8, _>(FixedConfig::test(), NoHook, &[])
            .expect("open fixed tree");

        typed
            .put(&k(1), TestItem { value: 111 })
            .expect("put typed");
        zero.put(&k(2), &ZeroVal(222)).expect("put zero");
        fixed.put(&k(3), &FixedVal(333)).expect("put fixed");

        db.close().expect("close");
    }

    // Phase 2: reopen and verify all entries survived.
    {
        let db = Db::open(dir.path()).expect("reopen");

        let typed = db
            .open_typed_tree::<TestItem, RapiraCodec, _>(Config::test(), NoHook, &[])
            .expect("reopen typed tree");
        let zero = db
            .open_zero_tree::<ZeroVal, 8, _>(Config::test(), NoHook, &[])
            .expect("reopen zero tree");
        let fixed = db
            .open_zero_tree_fixed::<FixedVal, 8, _>(FixedConfig::test(), NoHook, &[])
            .expect("reopen fixed tree");

        let t = typed.get(&k(1)).expect("typed key missing after reopen");
        assert_eq!(t.value, 111, "typed value mismatch");

        let z = zero.get(&k(2)).expect("zero key missing after reopen");
        assert_eq!(z.0, 222, "zero value mismatch");

        let f = fixed.get(&k(3)).expect("fixed key missing after reopen");
        assert_eq!(f.0, 333, "fixed value mismatch");
    }
}