armdb 0.14.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_into`],
    /// [`Self::last_into`], [`Self::range_into`], and
    /// [`Self::range_keys_into`]. 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_into(&self, key: &[u8], out: &mut Vec<u8>) -> DbResult<bool>;
    fn contains(&self, key: &[u8]) -> DbResult<bool>;
    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>>;
    /// Appends the first row and returns `true` only when exactly one row was
    /// accepted by `writer`; return `false` without appending for no match.
    fn first_into(&self, writer: &mut RowWriter<'_>) -> DbResult<bool>;
    /// Appends the last row and returns `true` only when exactly one row was
    /// accepted by `writer`; return `false` without appending for no match.
    fn last_into(&self, writer: &mut RowWriter<'_>) -> DbResult<bool>;
    /// Appends ordered key/value rows to `writer`.
    ///
    /// The returned count is advisory for compatibility; the response frame
    /// count is always derived from the rows accepted by `writer`.
    fn range_into(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        writer: &mut RowWriter<'_>,
    ) -> DbResult<u32>;
    /// Appends ordered key-only rows to `writer`.
    ///
    /// The returned count is advisory for compatibility; the response frame
    /// count is always derived from the rows accepted by `writer`.
    fn range_keys_into(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        writer: &mut RowWriter<'_>,
    ) -> DbResult<u32>;
    fn write(&self, key: WriteKey, flag: Option<bool>, value: ValueBytes) -> DbResult<ValueBytes>;
    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()>;
    fn take_into(&self, key: &[u8], soft: bool, out: &mut Vec<u8>) -> DbResult<bool>;
    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;

/// Streams encoded range rows directly into an RPC response frame.
///
/// A row has the wire layout `klen | key | vlen | value`, where both lengths
/// are big-endian `u32` values. The writer preserves a checkpoint before each
/// row so a failed codec or an oversized response cannot leave a partial row
/// in the frame.
#[cfg(feature = "rpc")]
pub struct RowWriter<'a> {
    buf: &'a mut Vec<u8>,
    take: usize,
    rows: u32,
    budget: usize,
    used: usize,
    collection: &'static str,
}

#[cfg(feature = "rpc")]
impl<'a> RowWriter<'a> {
    /// Creates a writer for a response frame.
    pub fn new(buf: &'a mut Vec<u8>, limit: u32, budget: usize, collection: &'static str) -> Self {
        let take = if limit == 0 {
            UNLIMITED_RANGE_CAP
        } else {
            limit as usize
        };
        Self {
            buf,
            take,
            rows: 0,
            budget,
            used: 0,
            collection,
        }
    }

    /// Returns the number of accepted rows.
    pub fn rows(&self) -> u32 {
        self.rows
    }

    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn take(&self) -> usize {
        self.take
    }

    #[cfg_attr(not(test), allow(dead_code))]
    pub(crate) fn len(&self) -> usize {
        self.buf.len()
    }

    pub(crate) fn check_len(&self, len: usize) -> DbResult<u32> {
        u32::try_from(len).map_err(|_| DbError::ResponseTooLarge {
            collection: self.collection,
            limit: self.budget,
        })
    }

    fn too_large(&self) -> DbError {
        DbError::ResponseTooLarge {
            collection: self.collection,
            limit: self.budget,
        }
    }

    fn accept(&mut self, checkpoint: usize) -> DbResult<ControlFlow<()>> {
        let written = self.buf.len() - checkpoint;
        let Some(used) = self
            .used
            .checked_add(written)
            .filter(|used| *used <= self.budget)
        else {
            self.buf.truncate(checkpoint);
            return Err(self.too_large());
        };
        self.used = used;
        self.rows += 1;
        if self.rows as usize >= self.take {
            return Ok(ControlFlow::Break(()));
        }
        Ok(ControlFlow::Continue(()))
    }

    /// Writes a key/value row, rolling back the buffer if encoding cannot finish.
    pub fn row<F>(&mut self, key: &[u8], write_value: F) -> DbResult<ControlFlow<()>>
    where
        F: FnOnce(&mut Vec<u8>) -> DbResult<()>,
    {
        if self.rows as usize >= self.take {
            return Ok(ControlFlow::Break(()));
        }
        let checkpoint = self.buf.len();
        let klen = self.check_len(key.len())?;
        self.buf.extend_from_slice(&klen.to_be_bytes());
        self.buf.extend_from_slice(key);
        let vlen_at = self.buf.len();
        self.buf.extend_from_slice(&[0u8; 4]);

        if let Err(error) = write_value(self.buf) {
            self.buf.truncate(checkpoint);
            return Err(error);
        }
        let vlen = match self.check_len(self.buf.len() - vlen_at - 4) {
            Ok(vlen) => vlen,
            Err(error) => {
                self.buf.truncate(checkpoint);
                return Err(error);
            }
        };
        self.buf[vlen_at..vlen_at + 4].copy_from_slice(&vlen.to_be_bytes());
        self.accept(checkpoint)
    }

    /// Writes a key-only row to the response frame.
    pub fn key_only(&mut self, key: &[u8]) -> DbResult<ControlFlow<()>> {
        if self.rows as usize >= self.take {
            return Ok(ControlFlow::Break(()));
        }
        let checkpoint = self.buf.len();
        let klen = self.check_len(key.len())?;
        self.buf.extend_from_slice(&klen.to_be_bytes());
        self.buf.extend_from_slice(key);
        self.accept(checkpoint)
    }
}

#[cfg(test)]
#[cfg(feature = "rpc")]
mod row_writer_tests {
    use super::*;

    fn writer(buf: &mut Vec<u8>, limit: u32, budget: usize) -> RowWriter<'_> {
        RowWriter::new(buf, limit, budget, "test")
    }

    #[test]
    fn row_encodes_klen_key_vlen_value() {
        let mut buf = Vec::new();
        let mut w = writer(&mut buf, 10, 1024);
        let _ = w
            .row(&[1, 2], |b| {
                b.extend_from_slice(&[9]);
                Ok(())
            })
            .expect("row");
        assert_eq!(buf, vec![0, 0, 0, 2, 1, 2, 0, 0, 0, 1, 9]);
    }

    #[test]
    fn limit_zero_becomes_unlimited_cap() {
        let mut buf = Vec::new();
        let w = writer(&mut buf, 0, 1 << 30);
        assert_eq!(w.take(), UNLIMITED_RANGE_CAP);
    }

    #[test]
    fn break_is_returned_on_the_last_accepted_row() {
        let mut buf = Vec::new();
        let mut w = writer(&mut buf, 2, 1024);
        assert!(matches!(
            w.row(&[1], |b| {
                b.push(1);
                Ok(())
            }),
            Ok(ControlFlow::Continue(()))
        ));
        assert!(matches!(
            w.row(&[2], |b| {
                b.push(2);
                Ok(())
            }),
            Ok(ControlFlow::Break(()))
        ));
        assert_eq!(w.rows(), 2);
    }

    #[test]
    fn budget_overflow_truncates_and_reports_413() {
        let mut buf = Vec::new();
        let mut w = writer(&mut buf, 10, 12); // room for exactly one small row
        let _ = w
            .row(&[1], |b| {
                b.push(1);
                Ok(())
            })
            .expect("first row fits");
        let before = w.len();
        let err = w
            .row(&[2], |b| {
                b.extend_from_slice(&[0u8; 64]);
                Ok(())
            })
            .expect_err("must exceed the budget");
        assert_eq!(err.status_code(), 413);
        assert_eq!(w.len(), before, "buffer was not truncated back");
    }

    #[test]
    fn a_codec_that_appends_then_fails_leaves_a_clean_buffer() {
        let mut buf = Vec::new();
        let mut w = writer(&mut buf, 10, 1024);
        let before = w.len();
        let _ = w.row(&[1], |b| {
            b.extend_from_slice(&[1, 2, 3]);
            Err(DbError::KeyNotFound)
        });
        assert_eq!(w.len(), before, "partial row left behind");
    }

    #[test]
    fn key_longer_than_u32_is_rejected() {
        // Guard the `as u32` cast; a 4 GiB key is unreachable in practice but the
        // cast must not silently wrap.
        let mut buf = Vec::new();
        let w = writer(&mut buf, 10, usize::MAX);
        assert!(w.check_len(u32::MAX as usize + 1).is_err());
    }
}

#[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 mut out = Vec::new();
        let err = RpcHandler::get_into(&adapter_for_typed_tree(&env, &tree), &[1, 2, 3], &mut out)
            .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);
    }
}