armdb 0.11.2

use std::mem::size_of;
use std::ops::Bound;
#[cfg(feature = "rpc")]
use std::ops::ControlFlow;

#[cfg(feature = "rpc")]
use armour_rpc::protocol::WriteKey;

use crate::{DbError, DbResult, Key};

#[cfg_attr(not(feature = "rpc"), allow(dead_code))]
pub type KeyBytes = Vec<u8>;
#[cfg_attr(not(feature = "rpc"), allow(dead_code))]
pub type ValueBytes = Vec<u8>;

#[cfg(feature = "rpc")]
pub trait RpcHandler: Send + Sync {
    fn name(&self) -> &str;
    fn info(&self) -> (u64, u16);
    fn schema(&self) -> armour_rpc::SchemaResponse;
    /// Reports whether this handler supports the complete ordered-scan contract.
    ///
    /// Returning `true` promises support for [`Self::first`], [`Self::last`],
    /// [`Self::range`], and [`Self::range_keys`]. The conservative default is
    /// `false`, so existing custom handlers do not advertise scan support
    /// unless they explicitly opt in.
    fn scannable(&self) -> bool {
        false
    }
    fn get(&self, key: &[u8]) -> DbResult<Option<ValueBytes>>;
    fn contains(&self, key: &[u8]) -> DbResult<bool>;
    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>>;
    fn first(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>>;
    fn last(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>>;
    fn range(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<(KeyBytes, ValueBytes)>>;
    fn range_keys(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<ValueBytes>>;
    fn write(&self, key: WriteKey, flag: Option<bool>, value: ValueBytes) -> DbResult<ValueBytes>;
    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()>;
    fn take(&self, key: &[u8], soft: bool) -> DbResult<Option<ValueBytes>>;
    fn count(&self, exact: bool) -> DbResult<u64>;
    fn apply_batch(&self, items: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()>;
}

// ─── helpers ─────────────────────────────────────────────────────────────────

#[cfg_attr(not(feature = "rpc"), allow(dead_code))]
pub(crate) fn check_key_len<K: Key>(key: &[u8]) -> DbResult<K> {
    if key.len() != size_of::<K>() {
        return Err(DbError::KeyNotFound);
    }
    Ok(K::from_bytes(key))
}

#[cfg(feature = "rpc")]
pub(crate) fn resolve_key<K: Key>(
    key: WriteKey,
    seq: &super::seq::SeqGen,
    name: &str,
) -> DbResult<(K, KeyBytes)> {
    let key_bytes = match key {
        WriteKey::Sequence => {
            let id = seq.next_id(name)?;
            // Big-endian: every 8-byte `Key` type orders by its BE byte
            // representation, so BE keeps iteration order == assignment order.
            id.to_be_bytes().to_vec()
        }
        WriteKey::Provided(k) => k,
    };
    if key_bytes.len() != size_of::<K>() {
        return Err(DbError::Client("invalid key bytes"));
    }
    Ok((K::from_bytes(&key_bytes), key_bytes))
}

#[cfg_attr(not(feature = "rpc"), allow(dead_code))]
pub(crate) fn bound_to_key_bound<K: Key>(bound: &Bound<KeyBytes>) -> DbResult<Bound<K>> {
    match bound {
        Bound::Included(b) => {
            if b.len() != size_of::<K>() {
                return Err(DbError::Client("invalid bound key bytes"));
            }
            Ok(Bound::Included(K::from_bytes(b)))
        }
        Bound::Excluded(b) => {
            if b.len() != size_of::<K>() {
                return Err(DbError::Client("invalid bound key bytes"));
            }
            Ok(Bound::Excluded(K::from_bytes(b)))
        }
        Bound::Unbounded => Ok(Bound::Unbounded),
    }
}

#[cfg(feature = "rpc")]
pub(crate) const UNLIMITED_RANGE_CAP: usize = 1_000_000;

/// Accumulates one RPC `range` / `range_keys` response under the wire's limit rule.
///
/// `limit == 0` is the wire spelling of "no limit" — but it is also what an unset
/// default field looks like, and the result is materialised into a `Vec` before
/// being sent, so it is capped at [`UNLIMITED_RANGE_CAP`] rather than allowed to
/// stream an unbounded collection into RAM.
///
/// Exists because the RPC adapter drives a visitor (`Access::scan_range`); the
/// hand-written handler bodies it replaced drove a plain iterator through a
/// `take`-based limit instead. Both shapes must produce byte-identical
/// responses, so the rule lives in one place.
#[cfg(feature = "rpc")]
pub(crate) struct RangeCollector {
    take: usize,
    rows: Vec<(KeyBytes, ValueBytes)>,
    keys: Vec<KeyBytes>,
}

#[cfg(feature = "rpc")]
impl RangeCollector {
    pub(crate) fn new(limit: u32) -> Self {
        let take = if limit == 0 {
            UNLIMITED_RANGE_CAP
        } else {
            limit as usize
        };
        Self {
            take,
            rows: Vec::new(),
            keys: Vec::new(),
        }
    }

    /// `Break` once the response is full — `encode` is not called for that row.
    pub(crate) fn push_with(
        &mut self,
        encode: impl FnOnce() -> DbResult<(KeyBytes, ValueBytes)>,
    ) -> DbResult<ControlFlow<()>> {
        if self.rows.len() >= self.take {
            return Ok(ControlFlow::Break(()));
        }
        self.rows.push(encode()?);
        if self.rows.len() >= self.take {
            return Ok(ControlFlow::Break(()));
        }
        Ok(ControlFlow::Continue(()))
    }

    pub(crate) fn push_key(&mut self, key: KeyBytes) -> ControlFlow<()> {
        if self.keys.len() >= self.take {
            return ControlFlow::Break(());
        }
        self.keys.push(key);
        if self.keys.len() >= self.take {
            return ControlFlow::Break(());
        }
        ControlFlow::Continue(())
    }

    pub(crate) fn rows(self) -> Vec<(KeyBytes, ValueBytes)> {
        self.rows
    }

    pub(crate) fn keys(self) -> Vec<KeyBytes> {
        self.keys
    }
}

#[cfg(all(test, feature = "rpc"))]
mod range_helper_tests {
    use super::{DbError, DbResult, KeyBytes, RangeCollector, UNLIMITED_RANGE_CAP, ValueBytes};
    use crate::Key;

    /// Drives `RangeCollector` the way the RPC adapter does: push until it asks
    /// to stop.
    fn rows_from<K: Key, V>(
        items: Vec<(K, V)>,
        limit: u32,
        mut encode: impl FnMut(&V, &mut Vec<u8>) -> DbResult<()>,
    ) -> DbResult<Vec<(KeyBytes, ValueBytes)>> {
        let mut out = RangeCollector::new(limit);
        for (key, value) in items {
            let step = out.push_with(|| {
                let mut bytes = Vec::new();
                encode(&value, &mut bytes)?;
                Ok((key.as_bytes().to_vec(), bytes))
            })?;
            if step.is_break() {
                break;
            }
        }
        Ok(out.rows())
    }

    /// Drives `RangeCollector::push_key` the way the RPC adapter does for
    /// `range_keys`: push until it asks to stop.
    fn keys_from<K: Key>(items: impl Iterator<Item = K>, limit: u32) -> Vec<KeyBytes> {
        let mut out = RangeCollector::new(limit);
        for key in items {
            if out.push_key(key.as_bytes().to_vec()).is_break() {
                break;
            }
        }
        out.keys()
    }

    #[test]
    fn explicit_limit_is_honored_exactly() {
        let mut out = RangeCollector::new(3);
        for n in 0u32.. {
            if out.push_key(n.to_be_bytes().to_vec()).is_break() {
                break;
            }
        }
        assert_eq!(out.keys().len(), 3);

        let mut out = RangeCollector::new(10);
        for n in 0u32..2 {
            if out.push_key(n.to_be_bytes().to_vec()).is_break() {
                break;
            }
        }
        assert_eq!(out.keys().len(), 2, "short iterator ends early");
    }

    #[test]
    fn zero_limit_means_everything_but_is_capped() {
        // `limit == 0` is the wire spelling of "no limit" and must keep meaning
        // "everything" — but it is also what an unset default field looks like,
        // and the result is materialized into a Vec before being sent. Cap it so
        // an accidental 0 cannot stream an unbounded collection into RAM.
        let mut out = RangeCollector::new(0);
        for _ in std::iter::repeat(0u8) {
            if out.push_key(vec![0u8]).is_break() {
                break;
            }
        }
        assert_eq!(
            out.keys().len(),
            UNLIMITED_RANGE_CAP,
            "an infinite source must terminate at the cap, not hang"
        );
        // Anything smaller than the cap is still returned in full.
        let mut out = RangeCollector::new(0);
        for n in 0u32..500 {
            if out.push_key(n.to_be_bytes().to_vec()).is_break() {
                break;
            }
        }
        assert_eq!(out.keys().len(), 500);
    }

    #[test]
    fn collect_range_honors_limit_and_propagates_encode_error() {
        let rows: Vec<_> = (1_u64..=4).map(|n| (n.to_be_bytes(), n)).collect();
        let got = rows_from(rows, 2, |value, buf| {
            buf.extend_from_slice(&value.to_be_bytes());
            Ok(())
        })
        .unwrap();
        assert_eq!(got.len(), 2);
        assert_eq!(
            got[0],
            (1_u64.to_be_bytes().to_vec(), 1_u64.to_be_bytes().to_vec())
        );

        let err = rows_from(vec![(1_u64.to_be_bytes(), 7_u64)], 1, |_value, _buf| {
            Err(DbError::Client("encode failed"))
        })
        .unwrap_err();
        assert!(matches!(err, DbError::Client("encode failed")));
    }

    #[test]
    fn collect_keys_applies_explicit_and_zero_limits() {
        assert_eq!(
            keys_from((1_u64..=4).map(u64::to_be_bytes), 2),
            vec![1_u64.to_be_bytes().to_vec(), 2_u64.to_be_bytes().to_vec()],
        );
        assert_eq!(
            keys_from(std::iter::repeat(1_u64.to_be_bytes()), 0).len(),
            UNLIMITED_RANGE_CAP,
        );
    }
}

#[cfg(test)]
mod inspection_handler_tests {
    #[cfg(feature = "rapira-codec")]
    use std::collections::HashSet;
    #[cfg(feature = "rapira-codec")]
    use std::num::NonZeroU32;
    use std::ops::Bound;
    use std::sync::Arc;

    #[cfg(feature = "rapira-codec")]
    use armour_core::CollectionShape;
    use armour_core::{GetType, KeyScheme, KeyType, ScalarTyp, StorageBackend, Typ, ValueModel};
    #[cfg(feature = "rapira-codec")]
    use rapira::Rapira;
    use tempfile::tempdir;

    use crate::access::{Adapter, CodecWire, ZeroWire};
    use crate::armour::inspection::{
        InspectionHandler, IterationDirection, NormalizedBounds, ScanCapability,
    };
    use crate::armour::seq::SeqGen;
    #[cfg(feature = "rapira-codec")]
    use crate::codec::RapiraCodec;
    use crate::codec::{Codec, InspectionEncoding};
    use crate::durability::{Bitcask, Fixed};
    use crate::tree_meta::TreeMeta;
    use crate::{CollectionMeta, Config, DbError, DbResult, FixedConfig, Key, NoHook};

    type Key8 = [u8; 8];

    #[cfg(feature = "rapira-codec")]
    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct Row {
        n: u64,
    }

    #[cfg(feature = "rapira-codec")]
    impl CollectionMeta for Row {
        type SelfId = Key8;
        const NAME: &'static str = "inspect_row";
        const VERSION: u16 = 7;
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        zerocopy::FromBytes,
        zerocopy::IntoBytes,
        zerocopy::Immutable,
    )]
    #[repr(C)]
    struct ZeroRow {
        n: u32,
    }

    impl CollectionMeta for ZeroRow {
        type SelfId = Key8;
        const NAME: &'static str = "inspect_zero";
        const VERSION: u16 = 3;
    }

    #[derive(Default, Clone)]
    struct OpaqueCodec;

    impl Codec<u64> for OpaqueCodec {
        fn encode_to(&self, value: &u64, buf: &mut Vec<u8>) -> DbResult<()> {
            buf.clear();
            buf.extend_from_slice(&value.to_le_bytes());
            Ok(())
        }
        fn decode_from(&self, bytes: &[u8]) -> DbResult<u64> {
            let arr: [u8; 8] = bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset: 0 })?;
            Ok(u64::from_le_bytes(arr))
        }
    }

    struct TestEnv {
        _dir: tempfile::TempDir,
        seq: Arc<SeqGen>,
    }
    impl TestEnv {
        fn new() -> Self {
            let dir = tempdir().unwrap();
            let seq = SeqGen::open(dir.path().join("seq")).unwrap();
            Self { _dir: dir, seq }
        }
        fn path(&self, name: &str) -> std::path::PathBuf {
            self._dir.path().join(name)
        }
    }

    fn key(id: u64) -> Key8 {
        id.to_be_bytes()
    }
    fn full_bounds() -> NormalizedBounds {
        NormalizedBounds {
            start: Bound::Unbounded,
            end: Bound::Unbounded,
            empty: false,
        }
    }

    #[cfg(feature = "rapira-codec")]
    fn open_typed_tree(
        env: &TestEnv,
        reversed: bool,
    ) -> Arc<crate::TypedTree<Key8, Row, RapiraCodec, NoHook>> {
        let mut cfg = Config::test();
        cfg.reversed = reversed;
        Arc::new(crate::TypedTree::open(env.path("typed_tree"), cfg, RapiraCodec).unwrap())
    }

    #[cfg(feature = "rapira-codec")]
    fn open_typed_map(
        env: &TestEnv,
        iterable: bool,
        reversed: bool,
    ) -> Arc<crate::TypedMap<Key8, Row, RapiraCodec, NoHook>> {
        let mut cfg = Config::test();
        cfg.iterable = iterable;
        cfg.reversed = reversed;
        Arc::new(
            crate::TypedMap::open(
                env.path(if iterable {
                    "typed_map_iter"
                } else {
                    "typed_map_snap"
                }),
                cfg,
                RapiraCodec,
            )
            .unwrap(),
        )
    }

    fn open_opaque_tree(env: &TestEnv) -> Arc<crate::TypedTree<Key8, u64, OpaqueCodec, NoHook>> {
        Arc::new(
            crate::TypedTree::open(env.path("opaque_tree"), Config::test(), OpaqueCodec).unwrap(),
        )
    }

    fn open_zero_bitcask_tree(
        env: &TestEnv,
        name: &str,
        reversed: bool,
    ) -> Arc<crate::ZeroTree<Key8, 4, ZeroRow, NoHook, Bitcask>> {
        let mut cfg = Config::test();
        cfg.reversed = reversed;
        Arc::new(
            crate::ZeroTree::<Key8, 4, ZeroRow, NoHook, Bitcask>::open(env.path(name), cfg)
                .unwrap(),
        )
    }

    fn open_zero_fixed_tree(
        env: &TestEnv,
        name: &str,
        reversed: bool,
    ) -> Arc<crate::ZeroTree<Key8, 4, ZeroRow, NoHook, Fixed>> {
        let cfg = FixedConfig {
            reversed,
            ..FixedConfig::test()
        };
        Arc::new(
            crate::ZeroTree::<Key8, 4, ZeroRow, NoHook, Fixed>::open(env.path(name), cfg).unwrap(),
        )
    }

    fn open_zero_bitcask_map(
        env: &TestEnv,
        name: &str,
        iterable: bool,
        reversed: bool,
    ) -> Arc<crate::ZeroMap<Key8, 4, ZeroRow, NoHook, Bitcask>> {
        let mut cfg = Config::test();
        cfg.iterable = iterable;
        cfg.reversed = reversed;
        Arc::new(
            crate::ZeroMap::<Key8, 4, ZeroRow, NoHook, Bitcask>::open(env.path(name), cfg).unwrap(),
        )
    }

    fn open_zero_fixed_map(
        env: &TestEnv,
        name: &str,
        iterable: bool,
        reversed: bool,
    ) -> Arc<crate::ZeroMap<Key8, 4, ZeroRow, NoHook, Fixed>> {
        let cfg = FixedConfig {
            iterable,
            reversed,
            ..FixedConfig::test()
        };
        Arc::new(
            crate::ZeroMap::<Key8, 4, ZeroRow, NoHook, Fixed>::open(env.path(name), cfg).unwrap(),
        )
    }

    #[cfg(feature = "rapira-codec")]
    fn encode_row(row: &Row) -> Vec<u8> {
        let mut buf = Vec::new();
        RapiraCodec.encode_to(row, &mut buf).unwrap();
        buf
    }

    /// Builds the same `Adapter` `register_typed_tree_handler` publishes as the
    /// live Inspection handler (see `armour/db.rs`), so these tests exercise the
    /// production dispatch path rather than the deleted direct impl.
    #[cfg(feature = "rapira-codec")]
    fn adapter_for_typed_tree(
        env: &TestEnv,
        tree: &Arc<crate::TypedTree<Key8, Row, RapiraCodec, NoHook>>,
    ) -> Adapter<crate::TypedTree<Key8, Row, RapiraCodec, NoHook>, CodecWire<RapiraCodec>> {
        Adapter::new(
            &TreeMeta::of::<Row>(),
            Arc::clone(tree),
            CodecWire(RapiraCodec),
            Arc::clone(&env.seq),
        )
    }

    #[cfg(feature = "rapira-codec")]
    fn adapter_for_typed_map(
        env: &TestEnv,
        map: &Arc<crate::TypedMap<Key8, Row, RapiraCodec, NoHook>>,
    ) -> Adapter<crate::TypedMap<Key8, Row, RapiraCodec, NoHook>, CodecWire<RapiraCodec>> {
        Adapter::new(
            &TreeMeta::of::<Row>(),
            Arc::clone(map),
            CodecWire(RapiraCodec),
            Arc::clone(&env.seq),
        )
    }

    /// `adapter_for_opaque_tree` hand-rolls a mismatched `ty`/`key_scheme`
    /// (opaque codec over a `Row`-shaped `typ_hash`), so its `TreeMeta` is
    /// built by hand to mirror those exact fields instead of
    /// `TreeMeta::of::<Row>()`.
    fn adapter_for_opaque_tree(
        env: &TestEnv,
        tree: &Arc<crate::TypedTree<Key8, u64, OpaqueCodec, NoHook>>,
    ) -> Adapter<crate::TypedTree<Key8, u64, OpaqueCodec, NoHook>, CodecWire<OpaqueCodec>> {
        let meta = TreeMeta {
            name: "opaque_tree",
            key_scheme: KeyScheme::Typed(&[KeyType::U64]),
            ty: Typ::Scalar(ScalarTyp::U64),
            group_bits: <Key8 as Key>::GROUP_BITS,
            version: 1,
        };
        Adapter::new(
            &meta,
            Arc::clone(tree),
            CodecWire(OpaqueCodec),
            Arc::clone(&env.seq),
        )
    }

    fn adapter_for_zero_tree<D>(
        env: &TestEnv,
        tree: &Arc<crate::ZeroTree<Key8, 4, ZeroRow, NoHook, D>>,
    ) -> Adapter<crate::ZeroTree<Key8, 4, ZeroRow, NoHook, D>, ZeroWire<4>>
    where
        D: crate::durability::Durability + crate::access::DurBackend,
    {
        Adapter::new(
            &TreeMeta::of::<ZeroRow>(),
            Arc::clone(tree),
            ZeroWire::<4>,
            Arc::clone(&env.seq),
        )
    }

    fn adapter_for_zero_map<D>(
        env: &TestEnv,
        map: &Arc<crate::ZeroMap<Key8, 4, ZeroRow, NoHook, D>>,
    ) -> Adapter<crate::ZeroMap<Key8, 4, ZeroRow, NoHook, D>, ZeroWire<4>>
    where
        D: crate::durability::Durability + crate::access::DurBackend,
    {
        Adapter::new(
            &TreeMeta::of::<ZeroRow>(),
            Arc::clone(map),
            ZeroWire::<4>,
            Arc::clone(&env.seq),
        )
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_metadata_typed_tree() {
        let env = TestEnv::new();
        let tree = open_typed_tree(&env, false);
        tree.put(&key(1), Row { n: 1 }).unwrap();
        let meta = adapter_for_typed_tree(&env, &tree).metadata().unwrap();
        assert_eq!(meta.name, Row::NAME);
        assert_eq!(meta.schema.version, Row::VERSION);
        assert_eq!(meta.schema.typ, Row::TYPE);
        assert_eq!(meta.schema.key_scheme, <Key8 as Key>::KEY_SCHEME);
        assert_eq!(meta.schema.key_size, 8);
        assert_eq!(meta.schema.encoding, InspectionEncoding::Rapira);
        assert_eq!(meta.storage.model, ValueModel::Typed);
        assert_eq!(meta.storage.shape, CollectionShape::Tree);
        assert_eq!(meta.storage.backend, StorageBackend::Bitcask);
        assert_eq!(meta.count, Some(1));
        assert_eq!(meta.scan, ScanCapability::Ordered);
        assert_eq!(meta.direction, Some(IterationDirection::Ascending));
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_metadata_iterable_typed_map() {
        let env = TestEnv::new();
        let map = open_typed_map(&env, true, true);
        map.put(&key(1), Row { n: 1 }).unwrap();
        let meta = adapter_for_typed_map(&env, &map).metadata().unwrap();
        assert_eq!(meta.scan, ScanCapability::Ordered);
        assert_eq!(meta.direction, Some(IterationDirection::Descending));
        assert_eq!(meta.storage.shape, CollectionShape::Map);
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_metadata_non_iterable_typed_map() {
        let env = TestEnv::new();
        let map = open_typed_map(&env, false, false);
        map.put(&key(1), Row { n: 1 }).unwrap();
        let meta = adapter_for_typed_map(&env, &map).metadata().unwrap();
        assert_eq!(meta.scan, ScanCapability::UnorderedSnapshot);
        assert!(meta.direction.is_none());
    }

    #[test]
    fn inspection_handler_metadata_zero_tree_bitcask() {
        let env = TestEnv::new();
        let tree = open_zero_bitcask_tree(&env, "ztree_bc", false);
        tree.put(&key(1), &ZeroRow { n: 1 }).unwrap();
        let meta = adapter_for_zero_tree(&env, &tree).metadata().unwrap();
        assert_eq!(meta.storage.model, ValueModel::Zero);
        assert_eq!(meta.storage.backend, StorageBackend::Bitcask);
        if cfg!(target_endian = "little") {
            assert_eq!(meta.schema.encoding, InspectionEncoding::Rapira);
        } else {
            assert_eq!(meta.schema.encoding, InspectionEncoding::RawLayout);
        }
        assert_eq!(meta.direction, Some(IterationDirection::Ascending));
    }

    #[test]
    fn inspection_handler_metadata_zero_tree_fixed() {
        let env = TestEnv::new();
        let tree = open_zero_fixed_tree(&env, "ztree_fx", true);
        tree.put(&key(1), &ZeroRow { n: 1 }).unwrap();
        let meta = adapter_for_zero_tree(&env, &tree).metadata().unwrap();
        assert_eq!(meta.storage.backend, StorageBackend::Fixed);
        assert_eq!(meta.direction, Some(IterationDirection::Descending));
    }

    #[test]
    fn inspection_handler_metadata_zero_map_backends() {
        let env = TestEnv::new();
        let bitcask = open_zero_bitcask_map(&env, "zmap_bc", true, false);
        bitcask.put(&key(1), &ZeroRow { n: 1 }).unwrap();
        let meta = adapter_for_zero_map(&env, &bitcask).metadata().unwrap();
        assert_eq!(meta.storage.backend, StorageBackend::Bitcask);
        assert_eq!(meta.scan, ScanCapability::Ordered);
        let fixed = open_zero_fixed_map(&env, "zmap_fx", false, false);
        fixed.put(&key(1), &ZeroRow { n: 1 }).unwrap();
        let meta = adapter_for_zero_map(&env, &fixed).metadata().unwrap();
        assert_eq!(meta.storage.backend, StorageBackend::Fixed);
        assert_eq!(meta.scan, ScanCapability::UnorderedSnapshot);
        assert!(meta.direction.is_none());
    }

    #[test]
    fn inspection_handler_metadata_opaque_codec() {
        let env = TestEnv::new();
        let tree = open_opaque_tree(&env);
        tree.put(&key(1), 42).unwrap();
        assert_eq!(
            adapter_for_opaque_tree(&env, &tree)
                .metadata()
                .unwrap()
                .schema
                .encoding,
            InspectionEncoding::Opaque
        );
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_exact_typed_tree() {
        let env = TestEnv::new();
        let tree = open_typed_tree(&env, false);
        let row = Row { n: 9 };
        tree.put(&key(1), row.clone()).unwrap();
        let adapter = adapter_for_typed_tree(&env, &tree);
        assert_eq!(adapter.get_raw(&key(1)).unwrap().unwrap(), encode_row(&row));
        assert!(adapter.get_raw(&key(2)).unwrap().is_none());
        assert_eq!(adapter.exact_count().unwrap(), 1);
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_exact_non_iterable_map_count_without_view() {
        let env = TestEnv::new();
        let map = open_typed_map(&env, false, false);
        map.put(&key(1), Row { n: 1 }).unwrap();
        map.put(&key(2), Row { n: 2 }).unwrap();
        assert!(map.iter_view().is_none());
        let adapter = adapter_for_typed_map(&env, &map);
        assert_eq!(adapter.exact_count().unwrap(), 2);
        assert!(adapter.get_raw(&key(9)).unwrap().is_none());
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_page_ascending_three_rows_cursor() {
        let env = TestEnv::new();
        let tree = open_typed_tree(&env, false);
        for id in 1..=3 {
            tree.put(&key(id), Row { n: id }).unwrap();
        }
        let page = adapter_for_typed_tree(&env, &tree)
            .ordered_page(&full_bounds(), NonZeroU32::new(2).unwrap())
            .unwrap();
        assert_eq!(page.entries.len(), 2);
        assert_eq!(page.entries[0].key, key(1).to_vec());
        assert_eq!(page.entries[1].key, key(2).to_vec());
        assert_eq!(page.next_key, Some(key(2).to_vec()));
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_page_ascending_two_rows_no_cursor() {
        let env = TestEnv::new();
        let tree = open_typed_tree(&env, false);
        for id in 1..=2 {
            tree.put(&key(id), Row { n: id }).unwrap();
        }
        let page = adapter_for_typed_tree(&env, &tree)
            .ordered_page(&full_bounds(), NonZeroU32::new(2).unwrap())
            .unwrap();
        assert_eq!(page.entries.len(), 2);
        assert!(page.next_key.is_none());
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_page_descending_three_rows_cursor() {
        let env = TestEnv::new();
        let tree = open_typed_tree(&env, true);
        for id in 1..=3 {
            tree.put(&key(id), Row { n: id }).unwrap();
        }
        let page = adapter_for_typed_tree(&env, &tree)
            .ordered_page(&full_bounds(), NonZeroU32::new(2).unwrap())
            .unwrap();
        assert_eq!(page.entries[0].key, key(3).to_vec());
        assert_eq!(page.entries[1].key, key(2).to_vec());
        assert_eq!(page.next_key, Some(key(2).to_vec()));
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_page_iterable_map_ascending() {
        let env = TestEnv::new();
        let map = open_typed_map(&env, true, false);
        for id in 1..=3 {
            map.put(&key(id), Row { n: id }).unwrap();
        }
        let page = adapter_for_typed_map(&env, &map)
            .ordered_page(&full_bounds(), NonZeroU32::new(2).unwrap())
            .unwrap();
        assert_eq!(page.entries.len(), 2);
        assert_eq!(page.next_key, Some(key(2).to_vec()));
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_snapshot_exact_set_unordered() {
        let env = TestEnv::new();
        let map = open_typed_map(&env, false, false);
        map.put(&key(1), Row { n: 10 }).unwrap();
        map.put(&key(2), Row { n: 20 }).unwrap();
        let entries = adapter_for_typed_map(&env, &map)
            .unordered_snapshot(NonZeroU32::new(2).unwrap())
            .unwrap();
        let keys: HashSet<_> = entries.iter().map(|e| e.key.clone()).collect();
        assert_eq!(keys, HashSet::from([key(1).to_vec(), key(2).to_vec()]));
    }

    #[cfg(feature = "rapira-codec")]
    #[test]
    fn inspection_handler_snapshot_preflight_and_visitor_stop() {
        let env = TestEnv::new();
        let map = open_typed_map(&env, false, false);
        for id in 1..=3 {
            map.put(&key(id), Row { n: id }).unwrap();
        }
        let err = adapter_for_typed_map(&env, &map)
            .unordered_snapshot(NonZeroU32::new(2).unwrap())
            .unwrap_err();
        assert!(matches!(
            err,
            crate::armour::inspection::InspectionError::SnapshotTooLarge {
                limit: 2,
                observed: Some(3)
            }
        ));
        let mut visits = 0u32;
        let mut acc =
            crate::armour::inspection::SnapshotAccumulator::new(NonZeroU32::new(2).unwrap());
        let _ = map.for_each_while(|k, v| {
            visits += 1;
            let mut value = Vec::new();
            RapiraCodec.encode_to(v, &mut value).unwrap();
            acc.push(crate::armour::inspection::InspectionEntry {
                key: k.as_bytes().to_vec(),
                value,
            })
        });
        assert_eq!(visits, 3);
        assert!(acc.finish().is_err());
    }

    #[cfg(feature = "rpc")]
    #[cfg(feature = "rapira-codec")]
    #[test]
    fn rpc_malformed_key_returns_key_not_found() {
        use super::RpcHandler;
        let env = TestEnv::new();
        let tree = open_typed_tree(&env, false);
        let err = RpcHandler::get(&adapter_for_typed_tree(&env, &tree), &[1, 2, 3]).unwrap_err();
        assert!(matches!(err, DbError::KeyNotFound));
    }
}

#[cfg(all(test, feature = "var-collections"))]
mod inspection_var_handler_tests {
    use std::num::NonZeroU32;
    use std::ops::Bound;
    use std::sync::Arc;

    use armour_core::GetType;
    use rapira::Rapira;
    use tempfile::tempdir;

    use crate::access::{Adapter, CodecWire};
    use crate::armour::inspection::{InspectionHandler, NormalizedBounds};
    use crate::armour::seq::SeqGen;
    use crate::test_faults::POISON;
    use crate::tree_meta::TreeMeta;
    use crate::{CollectionMeta, Config, DbError, DbResult, NoHook};

    type Key8 = [u8; 8];

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct Row {
        n: u64,
    }

    impl CollectionMeta for Row {
        type SelfId = Key8;
        const NAME: &'static str = "inspect_var_row";
        const VERSION: u16 = 9;
    }

    /// Mirror of `test_faults::PoisonCodec` for a `CollectionMeta` value type:
    /// decodes any 8-byte input except the `POISON` sentinel. `PoisonCodec`
    /// itself is `Codec<u64>` and can't serve `VarTypedTree`/`VarTypedMap`
    /// here — `Access` requires the value type to implement `CollectionMeta`,
    /// which bare `u64` does not.
    #[derive(Clone, Default)]
    struct PoisonRowCodec;

    impl crate::Codec<Row> for PoisonRowCodec {
        fn encode_to(&self, value: &Row, buf: &mut Vec<u8>) -> DbResult<()> {
            buf.clear();
            buf.extend_from_slice(&value.n.to_le_bytes());
            Ok(())
        }

        fn decode_from(&self, bytes: &[u8]) -> DbResult<Row> {
            let arr: [u8; 8] = bytes
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset: 0 })?;
            let n = u64::from_le_bytes(arr);
            if n == POISON {
                return Err(DbError::CorruptedEntry { offset: 0 });
            }
            Ok(Row { n })
        }

        fn size(&self, _value: &Row) -> usize {
            8
        }
    }

    fn key(id: u64) -> Key8 {
        id.to_be_bytes()
    }

    fn full_bounds() -> NormalizedBounds {
        NormalizedBounds {
            start: Bound::Unbounded,
            end: Bound::Unbounded,
            empty: false,
        }
    }

    fn open_var_tree() -> (
        tempfile::TempDir,
        Arc<SeqGen>,
        Arc<crate::VarTypedTree<Key8, Row, PoisonRowCodec, NoHook>>,
    ) {
        let dir = tempdir().unwrap();
        let seq = SeqGen::open(dir.path().join("seq")).unwrap();
        let tree = Arc::new(
            crate::VarTypedTree::open(dir.path().join("tree"), Config::test(), PoisonRowCodec)
                .unwrap(),
        );
        (dir, seq, tree)
    }

    fn open_var_map(
        iterable: bool,
    ) -> (
        tempfile::TempDir,
        Arc<SeqGen>,
        Arc<crate::VarTypedMap<Key8, Row, PoisonRowCodec, NoHook>>,
    ) {
        let dir = tempdir().unwrap();
        let seq = SeqGen::open(dir.path().join("seq")).unwrap();
        let mut cfg = Config::test();
        cfg.iterable = iterable;
        let map = Arc::new(
            crate::VarTypedMap::open(
                dir.path()
                    .join(if iterable { "map_iter" } else { "map_snap" }),
                cfg,
                PoisonRowCodec,
            )
            .unwrap(),
        );
        (dir, seq, map)
    }

    /// Builds the same `Adapter` `register_var_typed_tree_handler` /
    /// `register_var_typed_map_handler` publish as the live Inspection handler
    /// (see `armour/db.rs`), so these tests exercise the production dispatch
    /// path rather than the deleted direct impl.
    fn adapter_for_var_tree(
        tree: &Arc<crate::VarTypedTree<Key8, Row, PoisonRowCodec, NoHook>>,
        seq: &Arc<SeqGen>,
    ) -> Adapter<crate::VarTypedTree<Key8, Row, PoisonRowCodec, NoHook>, CodecWire<PoisonRowCodec>>
    {
        Adapter::new(
            &TreeMeta::of::<Row>(),
            Arc::clone(tree),
            CodecWire(PoisonRowCodec),
            Arc::clone(seq),
        )
    }

    fn adapter_for_var_map(
        map: &Arc<crate::VarTypedMap<Key8, Row, PoisonRowCodec, NoHook>>,
        seq: &Arc<SeqGen>,
    ) -> Adapter<crate::VarTypedMap<Key8, Row, PoisonRowCodec, NoHook>, CodecWire<PoisonRowCodec>>
    {
        Adapter::new(
            &TreeMeta::of::<Row>(),
            Arc::clone(map),
            CodecWire(PoisonRowCodec),
            Arc::clone(seq),
        )
    }

    #[test]
    fn inspection_var_tree_get_propagates_decode_error() {
        let (_dir, seq, tree) = open_var_tree();
        tree.put(&key(1), &Row { n: POISON }).unwrap();
        let adapter = adapter_for_var_tree(&tree, &seq);
        let err = adapter.get_raw(&key(1)).unwrap_err();
        assert!(matches!(err, DbError::CorruptedEntry { .. }), "{err:?}");
        assert!(adapter.get_raw(&key(9)).unwrap().is_none());
    }

    #[test]
    fn inspection_var_map_get_propagates_decode_error() {
        let (_dir, seq, map) = open_var_map(false);
        map.put(&key(1), &Row { n: POISON }).unwrap();
        let err = adapter_for_var_map(&map, &seq)
            .get_raw(&key(1))
            .unwrap_err();
        assert!(matches!(err, DbError::CorruptedEntry { .. }), "{err:?}");
    }

    #[test]
    fn inspection_var_tree_page_fails_on_poison_without_partial_page() {
        let (_dir, seq, tree) = open_var_tree();
        tree.put(&key(1), &Row { n: 1 }).unwrap();
        tree.put(&key(2), &Row { n: POISON }).unwrap();
        tree.put(&key(3), &Row { n: 3 }).unwrap();
        let err = adapter_for_var_tree(&tree, &seq)
            .ordered_page(&full_bounds(), NonZeroU32::new(10).unwrap())
            .unwrap_err();
        assert!(matches!(err, DbError::CorruptedEntry { .. }), "{err:?}");
    }

    #[test]
    fn inspection_var_map_page_fails_on_poison_without_partial_page() {
        let (_dir, seq, map) = open_var_map(true);
        map.put(&key(1), &Row { n: 1 }).unwrap();
        map.put(&key(2), &Row { n: POISON }).unwrap();
        map.put(&key(3), &Row { n: 3 }).unwrap();
        let err = adapter_for_var_map(&map, &seq)
            .ordered_page(&full_bounds(), NonZeroU32::new(10).unwrap())
            .unwrap_err();
        assert!(matches!(err, DbError::CorruptedEntry { .. }), "{err:?}");
    }

    #[test]
    fn inspection_var_map_snapshot_fails_on_poison_without_partial_snapshot() {
        let (_dir, seq, map) = open_var_map(false);
        map.put(&key(1), &Row { n: 1 }).unwrap();
        map.put(&key(2), &Row { n: POISON }).unwrap();
        map.put(&key(3), &Row { n: 3 }).unwrap();
        let err = adapter_for_var_map(&map, &seq)
            .unordered_snapshot(NonZeroU32::new(10).unwrap())
            .unwrap_err();
        assert!(
            matches!(
                err,
                crate::armour::inspection::InspectionError::Database(
                    DbError::CorruptedEntry { .. }
                )
            ),
            "{err:?}"
        );
    }

    #[test]
    fn inspection_var_tree_count_includes_poison_key() {
        let (_dir, seq, tree) = open_var_tree();
        tree.put(&key(1), &Row { n: 1 }).unwrap();
        tree.put(&key(2), &Row { n: POISON }).unwrap();
        tree.put(&key(3), &Row { n: 3 }).unwrap();
        assert_eq!(adapter_for_var_tree(&tree, &seq).exact_count().unwrap(), 3);
        assert_eq!(tree.iter().count(), 2, "lossy iter skips poison row");
    }

    #[test]
    fn inspection_var_tree_count_propagates_traversal_storage_error() {
        use std::io;

        let (_dir, seq, tree) = open_var_tree();
        tree.put(&key(1), &Row { n: 1 }).unwrap();
        tree.put(&key(2), &Row { n: 2 }).unwrap();
        tree.put(&key(3), &Row { n: 3 }).unwrap();
        let _fault = crate::var_tree::fail_count_live_keys_after(
            1,
            io::Error::new(io::ErrorKind::NotFound, "injected key-only traversal fault"),
        );
        let err = adapter_for_var_tree(&tree, &seq).exact_count().unwrap_err();
        assert!(
            matches!(err, DbError::Io(_)),
            "storage fault must propagate as Err, not Ok(1): {err:?}"
        );
    }

    #[test]
    fn inspection_var_tree_count_key_only_not_value_scan_error() {
        let (_dir, seq, tree) = open_var_tree();
        tree.put(&key(1), &Row { n: 1 }).unwrap();
        tree.put(&key(2), &Row { n: POISON }).unwrap();
        tree.put(&key(3), &Row { n: 3 }).unwrap();
        let strict_rows = tree
            .try_range_bounds(Bound::Unbounded, Bound::Unbounded)
            .collect::<Result<Vec<_>, _>>();
        assert!(
            matches!(strict_rows, Err(DbError::CorruptedEntry { .. })),
            "value-validating strict scan must fail, not return a partial page"
        );
        assert_eq!(adapter_for_var_tree(&tree, &seq).exact_count().unwrap(), 3);
    }

    #[test]
    fn inspection_var_map_count_includes_poison_key() {
        let (_dir, seq, map) = open_var_map(false);
        map.put(&key(1), &Row { n: 1 }).unwrap();
        map.put(&key(2), &Row { n: POISON }).unwrap();
        assert_eq!(adapter_for_var_map(&map, &seq).exact_count().unwrap(), 2);
    }
}