armdb 0.7.2

use std::hash::Hash;
use std::mem::size_of;
use std::num::NonZeroU32;
use std::ops::Bound;
use std::sync::Arc;

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

use armour_core::{CollectionShape, StorageBackend, StorageClass, ValueModel};

use super::inspection::{
    InspectionCollection, InspectionEntry, InspectionError, InspectionHandler, InspectionOperation,
    InspectionPage, InspectionSchema, IterationDirection, NormalizedBounds, ScanCapability,
    SnapshotAccumulator, collect_page,
};
use crate::codec::InspectionEncoding;
use crate::hook::TypedWriteHook;
use crate::map_index::inspection_batch;
#[cfg(feature = "rpc")]
use crate::zero_tree::from_value_bytes;
use crate::zero_tree::to_bytes;
use crate::{Codec, DbError, DbResult, Key, TypedMap, TypedTree, ZeroMap, ZeroTree};

pub type KeyBytes = Vec<u8>;
pub type ValueBytes = Vec<u8>;

pub(crate) trait InspectionDurability {
    const BACKEND: StorageBackend;
}

impl InspectionDurability for crate::durability::Bitcask {
    const BACKEND: StorageBackend = StorageBackend::Bitcask;
}

impl InspectionDurability for crate::durability::Fixed {
    const BACKEND: StorageBackend = StorageBackend::Fixed;
}

pub(crate) fn iteration_direction(reversed: bool) -> IterationDirection {
    if reversed {
        IterationDirection::Descending
    } else {
        IterationDirection::Ascending
    }
}

pub(crate) fn bound_to_key_bound_owned<K: Key>(bound: &Bound<Vec<u8>>) -> DbResult<Bound<K>> {
    match bound {
        Bound::Included(b) => Ok(Bound::Included(K::from_bytes(b))),
        Bound::Excluded(b) => Ok(Bound::Excluded(K::from_bytes(b))),
        Bound::Unbounded => Ok(Bound::Unbounded),
    }
}

#[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 upsert(&self, key: UpsertKey, 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 ─────────────────────────────────────────────────────────────────

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: UpsertKey,
    seq: &super::seq::SeqGen,
    name: &str,
) -> DbResult<(K, KeyBytes)> {
    let key_bytes = match key {
        UpsertKey::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()
        }
        UpsertKey::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(feature = "rpc")]
#[allow(dead_code)]
fn check_flag(tree_contains: bool, flag: Option<bool>) -> DbResult<()> {
    if let Some(update_only) = flag {
        if update_only && !tree_contains {
            return Err(DbError::KeyNotFound);
        }
        if !update_only && tree_contains {
            return Err(DbError::KeyExists);
        }
    }
    Ok(())
}

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;

#[cfg(feature = "rpc")]
pub(crate) fn apply_limit<I: Iterator>(iter: I, limit: u32) -> impl Iterator<Item = I::Item> {
    let take = if limit == 0 {
        UNLIMITED_RANGE_CAP
    } else {
        limit as usize
    };
    iter.take(take)
}

#[cfg(feature = "rpc")]
pub(crate) fn collect_range<K, V, I, F>(
    iter: I,
    limit: u32,
    encode: F,
) -> DbResult<Vec<(KeyBytes, ValueBytes)>>
where
    K: Key,
    I: Iterator<Item = (K, V)>,
    F: Fn(&V, &mut Vec<u8>) -> DbResult<()>,
{
    let mut result = Vec::new();
    for (key, value) in apply_limit(iter, limit) {
        let mut bytes = Vec::new();
        encode(&value, &mut bytes)?;
        result.push((key.as_bytes().to_vec(), bytes));
    }
    Ok(result)
}

#[cfg(feature = "rpc")]
pub(crate) fn collect_keys<K: Key>(iter: impl Iterator<Item = K>, limit: u32) -> Vec<KeyBytes> {
    apply_limit(iter, limit)
        .map(|key| key.as_bytes().to_vec())
        .collect()
}

// ─── TypedTreeHandler ────────────────────────────────────────────────────────

/// RPC handler for TypedTree — bridges bytes-level RPC to typed tree operations.
pub struct TypedTreeHandler<K, T, C, H: TypedWriteHook<K, T> = crate::NoHook>
where
    K: Key + Ord,
    T: Send + Sync,
    C: Codec<T>,
{
    pub name: String,
    pub typ_hash: u64,
    pub ty: armour_core::Typ,
    pub key_scheme: armour_core::KeyScheme,
    pub version: u16,
    pub tree: Arc<TypedTree<K, T, C, H>>,
    pub codec: Arc<C>,
    pub seq: Arc<super::seq::SeqGen>,
}

#[cfg(feature = "rpc")]
impl<K, T, C, H> RpcHandler for TypedTreeHandler<K, T, C, H>
where
    K: Key + Ord + Send + Sync,
    T: Send + Sync,
    C: Codec<T>,
    H: TypedWriteHook<K, T>,
{
    fn name(&self) -> &str {
        &self.name
    }

    fn info(&self) -> (u64, u16) {
        (self.typ_hash, self.version)
    }

    fn schema(&self) -> armour_rpc::SchemaResponse {
        armour_rpc::SchemaResponse {
            name: self.name.clone(),
            version: self.version,
            typ: self.ty,
            key_scheme: self.key_scheme,
        }
    }

    fn scannable(&self) -> bool {
        true
    }

    fn get(&self, key: &[u8]) -> DbResult<Option<ValueBytes>> {
        let key: K = check_key_len(key)?;
        match self.tree.get(&key) {
            Some(val) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&val, &mut buf)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn contains(&self, key: &[u8]) -> DbResult<bool> {
        let k = check_key_len::<K>(key)?;
        Ok(self.tree.contains(&k))
    }

    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>> {
        let k = check_key_len::<K>(key)?;
        Ok(self.tree.get(&k).map(|v| self.codec.size(&v) as u32))
    }

    fn first(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        match self.tree.first() {
            Some((k, v)) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&v, &mut buf)?;
                Ok(Some((k.as_bytes().to_vec(), buf)))
            }
            None => Ok(None),
        }
    }

    fn last(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        match self.tree.last() {
            Some((k, v)) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&v, &mut buf)?;
                Ok(Some((k.as_bytes().to_vec(), buf)))
            }
            None => Ok(None),
        }
    }

    fn range(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<(KeyBytes, ValueBytes)>> {
        let sb = bound_to_key_bound::<K>(&start)?;
        let eb = bound_to_key_bound::<K>(&end)?;
        let iter = self.tree.range_bounds(sb.as_ref(), eb.as_ref());
        let mut result = Vec::new();
        for (k, v) in apply_limit(iter, limit) {
            let mut buf = Vec::new();
            self.codec.encode_to(v, &mut buf)?;
            result.push((k.as_bytes().to_vec(), buf));
        }
        Ok(result)
    }

    fn range_keys(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<ValueBytes>> {
        let sb = bound_to_key_bound::<K>(&start)?;
        let eb = bound_to_key_bound::<K>(&end)?;
        let iter = self.tree.range_bounds(sb.as_ref(), eb.as_ref());
        let mut result = Vec::new();
        for (k, _) in apply_limit(iter, limit) {
            result.push(k.as_bytes().to_vec());
        }
        Ok(result)
    }

    fn upsert(
        &self,
        key: UpsertKey,
        flag: Option<bool>,
        value: ValueBytes,
    ) -> DbResult<ValueBytes> {
        let (typed_key, key_bytes) = resolve_key::<K>(key, &self.seq, &self.name)?;
        let decoded = self.codec.decode_from(&value)?;
        match flag {
            None => {
                self.tree.put(&typed_key, decoded)?;
                Ok(key_bytes)
            }
            Some(false) => self
                .tree
                .atomic(&typed_key, |shard| {
                    if shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyExists);
                    }
                    shard.put(&typed_key, decoded)?;
                    Ok(())
                })
                .map(|_| key_bytes),
            Some(true) => self
                .tree
                .atomic(&typed_key, |shard| {
                    if !shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyNotFound);
                    }
                    shard.put(&typed_key, decoded)?;
                    Ok(())
                })
                .map(|_| key_bytes),
        }
    }

    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        self.tree.delete(&k).map(|_| ())
    }

    fn take(&self, key: &[u8], soft: bool) -> DbResult<Option<ValueBytes>> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        let codec = Arc::clone(&self.codec);
        self.tree.atomic(&k, |shard| {
            if let Some(v) = shard.try_get(&k)? {
                let mut buf = Vec::new();
                codec.encode_to(v, &mut buf)?;
                shard.delete(&k)?;
                Ok(Some(buf))
            } else {
                Ok(None)
            }
        })
    }

    fn count(&self, exact: bool) -> DbResult<u64> {
        Ok(if exact {
            self.tree.iter().count() as u64
        } else {
            self.tree.len() as u64
        })
    }

    fn apply_batch(&self, items: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
        // Decode every item in a single pass — short-circuits on first error,
        // so no partial writes occur on validation failure.
        let decoded_items: Vec<(K, Option<T>)> = items
            .into_iter()
            .map(|(key, val)| {
                if key.len() != size_of::<K>() {
                    return Err(DbError::Client("invalid key length in batch"));
                }
                let decoded_val = match val {
                    Some(v) => Some(
                        self.codec
                            .decode_from(&v)
                            .map_err(|_| DbError::Client("invalid value bytes in batch"))?,
                    ),
                    None => None,
                };
                Ok((K::from_bytes(&key), decoded_val))
            })
            .collect::<DbResult<_>>()?;
        // Apply from the decoded vec — no second decode.
        for (k, val) in decoded_items {
            match val {
                Some(decoded) => {
                    self.tree.put(&k, decoded)?;
                }
                None => {
                    self.tree.delete(&k)?;
                }
            }
        }
        Ok(())
    }
}

// ─── TypedMapHandler ─────────────────────────────────────────────────────────

/// RPC handler for TypedMap — like TypedTreeHandler but without ordering operations.
pub struct TypedMapHandler<K, T, C, H: TypedWriteHook<K, T> = crate::NoHook>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T>,
{
    pub name: String,
    pub typ_hash: u64,
    pub ty: armour_core::Typ,
    pub key_scheme: armour_core::KeyScheme,
    pub version: u16,
    pub map: Arc<TypedMap<K, T, C, H>>,
    pub codec: Arc<C>,
    pub seq: Arc<super::seq::SeqGen>,
}

#[cfg(feature = "rpc")]
impl<K, T, C, H> RpcHandler for TypedMapHandler<K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T>,
    H: TypedWriteHook<K, T>,
{
    fn name(&self) -> &str {
        &self.name
    }

    fn info(&self) -> (u64, u16) {
        (self.typ_hash, self.version)
    }

    fn schema(&self) -> armour_rpc::SchemaResponse {
        armour_rpc::SchemaResponse {
            name: self.name.clone(),
            version: self.version,
            typ: self.ty,
            key_scheme: self.key_scheme,
        }
    }

    fn scannable(&self) -> bool {
        self.map.iter_view().is_some()
    }

    fn get(&self, key: &[u8]) -> DbResult<Option<ValueBytes>> {
        let key: K = check_key_len(key)?;
        match self.map.get(&key) {
            Some(val) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&val, &mut buf)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn contains(&self, key: &[u8]) -> DbResult<bool> {
        let k = check_key_len::<K>(key)?;
        Ok(self.map.contains(&k))
    }

    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>> {
        let k = check_key_len::<K>(key)?;
        Ok(self.map.get(&k).map(|v| self.codec.size(&v) as u32))
    }

    fn first(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        self.map
            .first()?
            .map(|(key, value)| {
                let mut bytes = Vec::new();
                self.codec.encode_to(&value, &mut bytes)?;
                Ok((key.as_bytes().to_vec(), bytes))
            })
            .transpose()
    }

    fn last(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        self.map
            .last()?
            .map(|(key, value)| {
                let mut bytes = Vec::new();
                self.codec.encode_to(&value, &mut bytes)?;
                Ok((key.as_bytes().to_vec(), bytes))
            })
            .transpose()
    }

    fn range(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<(KeyBytes, ValueBytes)>> {
        let view = self.map.iter_view().ok_or(DbError::NotImplemented)?;
        let start = bound_to_key_bound::<K>(&start)?;
        let end = bound_to_key_bound::<K>(&end)?;
        collect_range(
            view.range_bounds(start.as_ref(), end.as_ref()),
            limit,
            |value, bytes| self.codec.encode_to(value, bytes),
        )
    }

    fn range_keys(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<ValueBytes>> {
        let view = self.map.iter_view().ok_or(DbError::NotImplemented)?;
        let start = bound_to_key_bound::<K>(&start)?;
        let end = bound_to_key_bound::<K>(&end)?;
        Ok(collect_keys(
            view.keys_range_bounds(start.as_ref(), end.as_ref()),
            limit,
        ))
    }

    fn upsert(
        &self,
        key: UpsertKey,
        flag: Option<bool>,
        value: ValueBytes,
    ) -> DbResult<ValueBytes> {
        let (typed_key, key_bytes) = resolve_key::<K>(key, &self.seq, &self.name)?;
        let decoded = self.codec.decode_from(&value)?;
        match flag {
            None => {
                self.map.put(&typed_key, decoded)?;
                Ok(key_bytes)
            }
            Some(false) => self
                .map
                .atomic(&typed_key, |shard| {
                    if shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyExists);
                    }
                    shard.put(&typed_key, decoded)?;
                    Ok(())
                })
                .map(|_| key_bytes),
            Some(true) => self
                .map
                .atomic(&typed_key, |shard| {
                    if !shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyNotFound);
                    }
                    shard.put(&typed_key, decoded)?;
                    Ok(())
                })
                .map(|_| key_bytes),
        }
    }

    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        self.map.delete(&k).map(|_| ())
    }

    fn take(&self, key: &[u8], soft: bool) -> DbResult<Option<ValueBytes>> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        let codec = Arc::clone(&self.codec);
        self.map.atomic(&k, |shard| {
            if let Some(v) = shard.try_get(&k)? {
                let mut buf = Vec::new();
                codec.encode_to(v, &mut buf)?;
                shard.delete(&k)?;
                Ok(Some(buf))
            } else {
                Ok(None)
            }
        })
    }

    fn count(&self, _exact: bool) -> DbResult<u64> {
        // Map index is exact by construction; len() is the authoritative count
        // regardless of the exact flag.
        Ok(self.map.len() as u64)
    }

    fn apply_batch(&self, items: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
        // Decode every item in a single pass — short-circuits on first error,
        // so no partial writes occur on validation failure.
        let decoded_items: Vec<(K, Option<T>)> = items
            .into_iter()
            .map(|(key, val)| {
                if key.len() != size_of::<K>() {
                    return Err(DbError::Client("invalid key length in batch"));
                }
                let decoded_val = match val {
                    Some(v) => Some(
                        self.codec
                            .decode_from(&v)
                            .map_err(|_| DbError::Client("invalid value bytes in batch"))?,
                    ),
                    None => None,
                };
                Ok((K::from_bytes(&key), decoded_val))
            })
            .collect::<DbResult<_>>()?;
        // Apply from the decoded vec — no second decode.
        for (k, val) in decoded_items {
            match val {
                Some(decoded) => {
                    self.map.put(&k, decoded)?;
                }
                None => {
                    self.map.delete(&k)?;
                }
            }
        }
        Ok(())
    }
}

// ─── ZeroTreeHandler ─────────────────────────────────────────────────────────

/// RPC handler for ZeroTree — bridges bytes-level RPC using zerocopy encoding.
pub struct ZeroTreeHandler<
    K,
    const V: usize,
    T,
    H: TypedWriteHook<K, T> = crate::NoHook,
    D: crate::durability::Durability = crate::durability::Bitcask,
> where
    K: Key + Ord,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
{
    pub name: String,
    pub typ_hash: u64,
    pub ty: armour_core::Typ,
    pub key_scheme: armour_core::KeyScheme,
    pub version: u16,
    pub tree: Arc<ZeroTree<K, V, T, H, D>>,
    pub seq: Arc<super::seq::SeqGen>,
}

#[cfg(feature = "rpc")]
impl<K, const V: usize, T, H, D> RpcHandler for ZeroTreeHandler<K, V, T, H, D>
where
    K: Key + Ord + Send + Sync,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
    H: TypedWriteHook<K, T>,
    D: crate::durability::Durability,
{
    fn name(&self) -> &str {
        &self.name
    }

    fn info(&self) -> (u64, u16) {
        (self.typ_hash, self.version)
    }

    fn schema(&self) -> armour_rpc::SchemaResponse {
        armour_rpc::SchemaResponse {
            name: self.name.clone(),
            version: self.version,
            typ: self.ty,
            key_scheme: self.key_scheme,
        }
    }

    fn scannable(&self) -> bool {
        true
    }

    fn get(&self, key: &[u8]) -> DbResult<Option<ValueBytes>> {
        let key: K = check_key_len(key)?;
        match self.tree.get(&key) {
            Some(val) => Ok(Some(to_bytes::<V, T>(&val).to_vec())),
            None => Ok(None),
        }
    }

    fn contains(&self, key: &[u8]) -> DbResult<bool> {
        let k = check_key_len::<K>(key)?;
        Ok(self.tree.contains(&k))
    }

    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>> {
        let k = check_key_len::<K>(key)?;
        Ok(if self.tree.contains(&k) {
            Some(V as u32)
        } else {
            None
        })
    }

    fn first(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        match self.tree.first() {
            Some((k, v)) => Ok(Some((k.as_bytes().to_vec(), to_bytes::<V, T>(&v).to_vec()))),
            None => Ok(None),
        }
    }

    fn last(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        match self.tree.last() {
            Some((k, v)) => Ok(Some((k.as_bytes().to_vec(), to_bytes::<V, T>(&v).to_vec()))),
            None => Ok(None),
        }
    }

    fn range(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<(KeyBytes, ValueBytes)>> {
        let sb = bound_to_key_bound::<K>(&start)?;
        let eb = bound_to_key_bound::<K>(&end)?;
        Ok(apply_limit(
            self.tree
                .range_bounds(sb.as_ref(), eb.as_ref())
                .map(|(k, v)| (k.as_bytes().to_vec(), to_bytes::<V, T>(&v).to_vec())),
            limit,
        )
        .collect())
    }

    fn range_keys(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<ValueBytes>> {
        let sb = bound_to_key_bound::<K>(&start)?;
        let eb = bound_to_key_bound::<K>(&end)?;
        Ok(apply_limit(
            self.tree
                .range_bounds(sb.as_ref(), eb.as_ref())
                .map(|(k, _)| k.as_bytes().to_vec()),
            limit,
        )
        .collect())
    }

    fn upsert(
        &self,
        key: UpsertKey,
        flag: Option<bool>,
        value: ValueBytes,
    ) -> DbResult<ValueBytes> {
        let (typed_key, key_bytes) = resolve_key::<K>(key, &self.seq, &self.name)?;
        if value.len() != V {
            return Err(DbError::Client("invalid value length for zero collection"));
        }
        let zero_val: T = from_value_bytes::<V, T>(
            value[..V]
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset: 0 })?,
        );
        match flag {
            None => {
                self.tree.put(&typed_key, &zero_val)?;
                Ok(key_bytes)
            }
            Some(false) => self
                .tree
                .atomic(&typed_key, |shard| {
                    if shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyExists);
                    }
                    shard.put(&typed_key, &zero_val)?;
                    Ok(())
                })
                .map(|_| key_bytes),
            Some(true) => self
                .tree
                .atomic(&typed_key, |shard| {
                    if !shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyNotFound);
                    }
                    shard.put(&typed_key, &zero_val)?;
                    Ok(())
                })
                .map(|_| key_bytes),
        }
    }

    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        self.tree.delete(&k).map(|_| ())
    }

    fn take(&self, key: &[u8], soft: bool) -> DbResult<Option<ValueBytes>> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        self.tree.atomic(&k, |shard| {
            let current = shard.try_get(&k)?;
            let bytes = current.as_ref().map(|v| to_bytes::<V, T>(v).to_vec());
            if current.is_some() {
                shard.delete(&k)?;
            }
            Ok(bytes)
        })
    }

    fn count(&self, exact: bool) -> DbResult<u64> {
        Ok(if exact {
            self.tree.iter().count() as u64
        } else {
            self.tree.len() as u64
        })
    }

    fn apply_batch(&self, items: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
        // Pre-validate every item — length mismatches must abort the whole batch.
        for (key, val) in &items {
            if key.len() != size_of::<K>() {
                return Err(DbError::Client("invalid key length in batch"));
            }
            if let Some(v) = val
                && v.len() != V
            {
                return Err(DbError::Client("invalid value length in batch"));
            }
        }
        // Apply.
        for (key_bytes, val) in items {
            let k = K::from_bytes(&key_bytes);
            match val {
                Some(v) => {
                    let zv: T = from_value_bytes::<V, T>(
                        v[..V]
                            .try_into()
                            .map_err(|_| DbError::CorruptedEntry { offset: 0 })?,
                    );
                    self.tree.put(&k, &zv)?;
                }
                None => {
                    self.tree.delete(&k)?;
                }
            }
        }
        Ok(())
    }
}

// ─── ZeroMapHandler ──────────────────────────────────────────────────────────

/// RPC handler for ZeroMap — zerocopy encoding without ordering operations.
pub struct ZeroMapHandler<
    K,
    const V: usize,
    T,
    H: TypedWriteHook<K, T> = crate::NoHook,
    D: crate::durability::Durability = crate::durability::Bitcask,
> where
    K: Key + Send + Sync + Hash + Eq,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
{
    pub name: String,
    pub typ_hash: u64,
    pub ty: armour_core::Typ,
    pub key_scheme: armour_core::KeyScheme,
    pub version: u16,
    pub map: Arc<ZeroMap<K, V, T, H, D>>,
    pub seq: Arc<super::seq::SeqGen>,
}

#[cfg(feature = "rpc")]
impl<K, const V: usize, T, H, D> RpcHandler for ZeroMapHandler<K, V, T, H, D>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
    H: TypedWriteHook<K, T>,
    D: crate::durability::Durability,
{
    fn name(&self) -> &str {
        &self.name
    }

    fn info(&self) -> (u64, u16) {
        (self.typ_hash, self.version)
    }

    fn schema(&self) -> armour_rpc::SchemaResponse {
        armour_rpc::SchemaResponse {
            name: self.name.clone(),
            version: self.version,
            typ: self.ty,
            key_scheme: self.key_scheme,
        }
    }

    fn scannable(&self) -> bool {
        self.map.iter_view().is_some()
    }

    fn get(&self, key: &[u8]) -> DbResult<Option<ValueBytes>> {
        let key: K = check_key_len(key)?;
        match self.map.get(&key) {
            Some(val) => Ok(Some(to_bytes::<V, T>(&val).to_vec())),
            None => Ok(None),
        }
    }

    fn contains(&self, key: &[u8]) -> DbResult<bool> {
        let k = check_key_len::<K>(key)?;
        Ok(self.map.contains(&k))
    }

    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>> {
        let k = check_key_len::<K>(key)?;
        Ok(if self.map.contains(&k) {
            Some(V as u32)
        } else {
            None
        })
    }

    fn first(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        Ok(self
            .map
            .first()?
            .map(|(key, value)| (key.as_bytes().to_vec(), to_bytes::<V, T>(&value).to_vec())))
    }

    fn last(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        Ok(self
            .map
            .last()?
            .map(|(key, value)| (key.as_bytes().to_vec(), to_bytes::<V, T>(&value).to_vec())))
    }

    fn range(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<(KeyBytes, ValueBytes)>> {
        let view = self.map.iter_view().ok_or(DbError::NotImplemented)?;
        let start = bound_to_key_bound::<K>(&start)?;
        let end = bound_to_key_bound::<K>(&end)?;
        collect_range(
            view.range_bounds(start.as_ref(), end.as_ref()),
            limit,
            |value, bytes| {
                bytes.extend_from_slice(&to_bytes::<V, T>(value));
                Ok(())
            },
        )
    }

    fn range_keys(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<ValueBytes>> {
        let view = self.map.iter_view().ok_or(DbError::NotImplemented)?;
        let start = bound_to_key_bound::<K>(&start)?;
        let end = bound_to_key_bound::<K>(&end)?;
        Ok(collect_keys(
            view.keys_range_bounds(start.as_ref(), end.as_ref()),
            limit,
        ))
    }

    fn upsert(
        &self,
        key: UpsertKey,
        flag: Option<bool>,
        value: ValueBytes,
    ) -> DbResult<ValueBytes> {
        let (typed_key, key_bytes) = resolve_key::<K>(key, &self.seq, &self.name)?;
        if value.len() != V {
            return Err(DbError::Client("invalid value length for zero collection"));
        }
        let zero_val: T = from_value_bytes::<V, T>(
            value[..V]
                .try_into()
                .map_err(|_| DbError::CorruptedEntry { offset: 0 })?,
        );
        match flag {
            None => {
                self.map.put(&typed_key, &zero_val)?;
                Ok(key_bytes)
            }
            Some(false) => self
                .map
                .atomic(&typed_key, |shard| {
                    if shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyExists);
                    }
                    shard.put(&typed_key, &zero_val)?;
                    Ok(())
                })
                .map(|_| key_bytes),
            Some(true) => self
                .map
                .atomic(&typed_key, |shard| {
                    if !shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyNotFound);
                    }
                    shard.put(&typed_key, &zero_val)?;
                    Ok(())
                })
                .map(|_| key_bytes),
        }
    }

    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        self.map.delete(&k).map(|_| ())
    }

    fn take(&self, key: &[u8], soft: bool) -> DbResult<Option<ValueBytes>> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        self.map.atomic(&k, |shard| {
            let current = shard.try_get(&k)?;
            let bytes = current.as_ref().map(|v| to_bytes::<V, T>(v).to_vec());
            if current.is_some() {
                shard.delete(&k)?;
            }
            Ok(bytes)
        })
    }

    fn count(&self, _exact: bool) -> DbResult<u64> {
        // Map index is exact by construction; len() is the authoritative count
        // regardless of the exact flag.
        Ok(self.map.len() as u64)
    }

    fn apply_batch(&self, items: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
        // Pre-validate every item — length mismatches must abort the whole batch.
        for (key, val) in &items {
            if key.len() != size_of::<K>() {
                return Err(DbError::Client("invalid key length in batch"));
            }
            if let Some(v) = val
                && v.len() != V
            {
                return Err(DbError::Client("invalid value length in batch"));
            }
        }
        // Apply.
        for (key_bytes, val) in items {
            let k = K::from_bytes(&key_bytes);
            match val {
                Some(v) => {
                    let zv: T = from_value_bytes::<V, T>(
                        v[..V]
                            .try_into()
                            .map_err(|_| DbError::CorruptedEntry { offset: 0 })?,
                    );
                    self.map.put(&k, &zv)?;
                }
                None => {
                    self.map.delete(&k)?;
                }
            }
        }
        Ok(())
    }
}

// ─── InspectionHandler (Typed / Zero) ────────────────────────────────────────

impl<K, T, C, H> InspectionHandler for TypedTreeHandler<K, T, C, H>
where
    K: Key + Ord + Send + Sync,
    T: Send + Sync,
    C: Codec<T>,
    H: TypedWriteHook<K, T>,
{
    fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
        Ok(InspectionCollection {
            name: self.name.clone(),
            schema: InspectionSchema {
                version: self.version,
                typ: self.ty,
                key_scheme: self.key_scheme,
                key_size: size_of::<K>() as u32,
                encoding: C::INSPECTION_ENCODING,
            },
            storage: StorageClass {
                model: ValueModel::Typed,
                shape: CollectionShape::Tree,
                backend: StorageBackend::Bitcask,
            },
            count: Some(self.tree.len() as u64),
            scan: ScanCapability::Ordered,
            direction: Some(iteration_direction(self.tree.reversed())),
        })
    }

    fn get_raw(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        let key = K::from_bytes(key);
        match self.tree.get(&key) {
            Some(val) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&val, &mut buf)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn exact_count(&self) -> DbResult<u64> {
        Ok(self.tree.iter().count() as u64)
    }

    fn ordered_page(
        &self,
        bounds: &NormalizedBounds,
        limit: NonZeroU32,
    ) -> DbResult<InspectionPage> {
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        let start = bound_to_key_bound_owned::<K>(&bounds.start)?;
        let end = bound_to_key_bound_owned::<K>(&bounds.end)?;
        let codec = Arc::clone(&self.codec);
        collect_page(
            self.tree.range_bounds(start.as_ref(), end.as_ref()).map(Ok),
            limit,
            move |k, v| {
                let mut bytes = Vec::new();
                codec.encode_to(v, &mut bytes)?;
                Ok(InspectionEntry {
                    key: k.as_bytes().to_vec(),
                    value: bytes,
                })
            },
        )
    }

    fn unordered_snapshot(
        &self,
        _max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError> {
        Err(InspectionError::UnsupportedOperation {
            name: self.name.clone(),
            operation: InspectionOperation::SnapshotUnordered,
        })
    }
}

impl<K, T, C, H> InspectionHandler for TypedMapHandler<K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T>,
    H: TypedWriteHook<K, T>,
{
    fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
        let iterable = self.map.iter_view().is_some();
        Ok(InspectionCollection {
            name: self.name.clone(),
            schema: InspectionSchema {
                version: self.version,
                typ: self.ty,
                key_scheme: self.key_scheme,
                key_size: size_of::<K>() as u32,
                encoding: C::INSPECTION_ENCODING,
            },
            storage: StorageClass {
                model: ValueModel::Typed,
                shape: CollectionShape::Map,
                backend: StorageBackend::Bitcask,
            },
            count: Some(self.map.len() as u64),
            scan: if iterable {
                ScanCapability::Ordered
            } else {
                ScanCapability::UnorderedSnapshot
            },
            direction: if iterable {
                Some(iteration_direction(self.map.reversed()))
            } else {
                None
            },
        })
    }

    fn get_raw(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        let key = K::from_bytes(key);
        match self.map.get(&key) {
            Some(val) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&val, &mut buf)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn exact_count(&self) -> DbResult<u64> {
        Ok(self.map.len() as u64)
    }

    fn ordered_page(
        &self,
        bounds: &NormalizedBounds,
        limit: NonZeroU32,
    ) -> DbResult<InspectionPage> {
        if self.map.iter_view().is_none() {
            return Err(DbError::NotImplemented);
        }
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        let start = bound_to_key_bound_owned::<K>(&bounds.start)?;
        let end = bound_to_key_bound_owned::<K>(&bounds.end)?;
        let batch = inspection_batch(limit, self.map.inspection_shard_count());
        let codec = Arc::clone(&self.codec);
        let _guard = self.map.inspection_guard();
        collect_page(
            self.map
                .inspection_range_bounds(start.as_ref(), end.as_ref(), batch),
            limit,
            move |k, ptr| {
                let data = unsafe { &*ptr };
                let mut bytes = Vec::new();
                codec.encode_to(&data.value, &mut bytes)?;
                Ok(InspectionEntry {
                    key: k.as_bytes().to_vec(),
                    value: bytes,
                })
            },
        )
    }

    fn unordered_snapshot(
        &self,
        max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError> {
        if self.map.iter_view().is_some() {
            return Err(InspectionError::UnsupportedOperation {
                name: self.name.clone(),
                operation: InspectionOperation::SnapshotUnordered,
            });
        }
        let exact_count = self.map.len() as u64;
        let cap = max_entries.get();
        if exact_count > u64::from(cap) {
            return Err(InspectionError::SnapshotTooLarge {
                limit: cap,
                observed: Some(exact_count),
            });
        }
        let codec = Arc::clone(&self.codec);
        let mut acc = SnapshotAccumulator::new(max_entries);
        let mut db_error = None;
        let _ = self.map.for_each_while(|k, v| {
            let mut bytes = Vec::new();
            if let Err(error) = codec.encode_to(v, &mut bytes) {
                db_error = Some(error);
                return std::ops::ControlFlow::Break(());
            }
            acc.push(InspectionEntry {
                key: k.as_bytes().to_vec(),
                value: bytes,
            })
        });
        if let Some(error) = db_error {
            return Err(InspectionError::Database(error));
        }
        acc.finish()
    }
}

impl<K, const V: usize, T, H, D> InspectionHandler for ZeroTreeHandler<K, V, T, H, D>
where
    K: Key + Ord + Send + Sync,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
    H: TypedWriteHook<K, T>,
    D: crate::durability::Durability + InspectionDurability,
{
    fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
        Ok(InspectionCollection {
            name: self.name.clone(),
            schema: InspectionSchema {
                version: self.version,
                typ: self.ty,
                key_scheme: self.key_scheme,
                key_size: size_of::<K>() as u32,
                encoding: InspectionEncoding::RawLayout,
            },
            storage: StorageClass {
                model: ValueModel::Zero,
                shape: CollectionShape::Tree,
                backend: D::BACKEND,
            },
            count: Some(self.tree.len() as u64),
            scan: ScanCapability::Ordered,
            direction: Some(iteration_direction(self.tree.reversed())),
        })
    }

    fn get_raw(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        let key = K::from_bytes(key);
        Ok(self
            .tree
            .get(&key)
            .map(|val| to_bytes::<V, T>(&val).to_vec()))
    }

    fn exact_count(&self) -> DbResult<u64> {
        Ok(self.tree.iter().count() as u64)
    }

    fn ordered_page(
        &self,
        bounds: &NormalizedBounds,
        limit: NonZeroU32,
    ) -> DbResult<InspectionPage> {
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        let start = bound_to_key_bound_owned::<K>(&bounds.start)?;
        let end = bound_to_key_bound_owned::<K>(&bounds.end)?;
        collect_page(
            self.tree.range_bounds(start.as_ref(), end.as_ref()).map(Ok),
            limit,
            |k, v| {
                Ok(InspectionEntry {
                    key: k.as_bytes().to_vec(),
                    value: to_bytes::<V, T>(&v).to_vec(),
                })
            },
        )
    }

    fn unordered_snapshot(
        &self,
        _max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError> {
        Err(InspectionError::UnsupportedOperation {
            name: self.name.clone(),
            operation: InspectionOperation::SnapshotUnordered,
        })
    }
}

impl<K, const V: usize, T, H, D> InspectionHandler for ZeroMapHandler<K, V, T, H, D>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
    H: TypedWriteHook<K, T>,
    D: crate::durability::Durability + InspectionDurability,
{
    fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
        let iterable = self.map.iter_view().is_some();
        Ok(InspectionCollection {
            name: self.name.clone(),
            schema: InspectionSchema {
                version: self.version,
                typ: self.ty,
                key_scheme: self.key_scheme,
                key_size: size_of::<K>() as u32,
                encoding: InspectionEncoding::RawLayout,
            },
            storage: StorageClass {
                model: ValueModel::Zero,
                shape: CollectionShape::Map,
                backend: D::BACKEND,
            },
            count: Some(self.map.len() as u64),
            scan: if iterable {
                ScanCapability::Ordered
            } else {
                ScanCapability::UnorderedSnapshot
            },
            direction: if iterable {
                Some(iteration_direction(self.map.reversed()))
            } else {
                None
            },
        })
    }

    fn get_raw(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        let key = K::from_bytes(key);
        Ok(self
            .map
            .get(&key)
            .map(|val| to_bytes::<V, T>(&val).to_vec()))
    }

    fn exact_count(&self) -> DbResult<u64> {
        Ok(self.map.len() as u64)
    }

    fn ordered_page(
        &self,
        bounds: &NormalizedBounds,
        limit: NonZeroU32,
    ) -> DbResult<InspectionPage> {
        if self.map.iter_view().is_none() {
            return Err(DbError::NotImplemented);
        }
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        let start = bound_to_key_bound_owned::<K>(&bounds.start)?;
        let end = bound_to_key_bound_owned::<K>(&bounds.end)?;
        let batch = inspection_batch(limit, self.map.inspection_shard_count());
        collect_page(
            self.map
                .inspection_range_bounds(start.as_ref(), end.as_ref(), batch),
            limit,
            |k, v| {
                Ok(InspectionEntry {
                    key: k.as_bytes().to_vec(),
                    value: to_bytes::<V, T>(&v).to_vec(),
                })
            },
        )
    }

    fn unordered_snapshot(
        &self,
        max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError> {
        if self.map.iter_view().is_some() {
            return Err(InspectionError::UnsupportedOperation {
                name: self.name.clone(),
                operation: InspectionOperation::SnapshotUnordered,
            });
        }
        let exact_count = self.map.len() as u64;
        let cap = max_entries.get();
        if exact_count > u64::from(cap) {
            return Err(InspectionError::SnapshotTooLarge {
                limit: cap,
                observed: Some(exact_count),
            });
        }
        let mut acc = SnapshotAccumulator::new(max_entries);
        let _ = self.map.for_each_while(|k, v| {
            acc.push(InspectionEntry {
                key: k.as_bytes().to_vec(),
                value: to_bytes::<V, T>(v).to_vec(),
            })
        });
        acc.finish()
    }
}

// ─── VarTypedTreeHandler ─────────────────────────────────────────────────────

/// RPC handler for [`VarTypedTree`](crate::VarTypedTree) — disk-resident typed values.
#[cfg(feature = "var-collections")]
pub struct VarTypedTreeHandler<K, T, C, H: TypedWriteHook<K, T> = crate::NoHook>
where
    K: Key + Ord,
    T: Send + Sync,
    C: Codec<T> + Clone,
{
    pub name: String,
    pub typ_hash: u64,
    pub ty: armour_core::Typ,
    pub key_scheme: armour_core::KeyScheme,
    pub version: u16,
    pub tree: Arc<crate::VarTypedTree<K, T, C, H>>,
    pub codec: Arc<C>,
    pub seq: Arc<super::seq::SeqGen>,
}

#[cfg(all(feature = "rpc", feature = "var-collections"))]
impl<K, T, C, H> RpcHandler for VarTypedTreeHandler<K, T, C, H>
where
    K: Key + Ord + Send + Sync,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    fn name(&self) -> &str {
        &self.name
    }

    fn info(&self) -> (u64, u16) {
        (self.typ_hash, self.version)
    }

    fn schema(&self) -> armour_rpc::SchemaResponse {
        armour_rpc::SchemaResponse {
            name: self.name.clone(),
            version: self.version,
            typ: self.ty,
            key_scheme: self.key_scheme,
        }
    }

    fn scannable(&self) -> bool {
        true
    }

    fn get(&self, key: &[u8]) -> DbResult<Option<ValueBytes>> {
        let key: K = check_key_len(key)?;
        match self.tree.get(&key) {
            Some(val) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&val, &mut buf)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn contains(&self, key: &[u8]) -> DbResult<bool> {
        let k = check_key_len::<K>(key)?;
        Ok(self.tree.contains(&k))
    }

    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>> {
        let k = check_key_len::<K>(key)?;
        Ok(self.tree.entry_len(&k))
    }

    fn first(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        match self.tree.first() {
            Some((k, v)) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&v, &mut buf)?;
                Ok(Some((k.as_bytes().to_vec(), buf)))
            }
            None => Ok(None),
        }
    }

    fn last(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        match self.tree.last() {
            Some((k, v)) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&v, &mut buf)?;
                Ok(Some((k.as_bytes().to_vec(), buf)))
            }
            None => Ok(None),
        }
    }

    fn range(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<(KeyBytes, ValueBytes)>> {
        let sb = bound_to_key_bound::<K>(&start)?;
        let eb = bound_to_key_bound::<K>(&end)?;
        let iter = self.tree.range_bounds(sb.as_ref(), eb.as_ref());
        let mut result = Vec::new();
        for (k, v) in apply_limit(iter, limit) {
            let mut buf = Vec::new();
            self.codec.encode_to(&v, &mut buf)?;
            result.push((k.as_bytes().to_vec(), buf));
        }
        Ok(result)
    }

    fn range_keys(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<ValueBytes>> {
        let sb = bound_to_key_bound::<K>(&start)?;
        let eb = bound_to_key_bound::<K>(&end)?;
        let iter = self.tree.range_bounds(sb.as_ref(), eb.as_ref());
        Ok(apply_limit(iter, limit)
            .map(|(k, _)| k.as_bytes().to_vec())
            .collect())
    }

    fn upsert(
        &self,
        key: UpsertKey,
        flag: Option<bool>,
        value: ValueBytes,
    ) -> DbResult<ValueBytes> {
        let (typed_key, key_bytes) = resolve_key::<K>(key, &self.seq, &self.name)?;
        // Pre-decode to validate value bytes before write.
        let decoded = self.codec.decode_from(&value)?;
        match flag {
            None => {
                self.tree.put(&typed_key, &decoded)?;
                Ok(key_bytes)
            }
            Some(false) => self
                .tree
                .atomic(&typed_key, |shard| {
                    if shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyExists);
                    }
                    shard.put(&typed_key, &decoded)?;
                    Ok(())
                })
                .map(|_| key_bytes),
            Some(true) => self
                .tree
                .atomic(&typed_key, |shard| {
                    if !shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyNotFound);
                    }
                    shard.put(&typed_key, &decoded)?;
                    Ok(())
                })
                .map(|_| key_bytes),
        }
    }

    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        self.tree.delete(&k).map(|_| ())
    }

    fn take(&self, key: &[u8], soft: bool) -> DbResult<Option<ValueBytes>> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        let codec = Arc::clone(&self.codec);
        self.tree.atomic(&k, |shard| {
            let current = shard.try_get(&k)?;
            let bytes = if let Some(ref v) = current {
                let mut buf = Vec::new();
                codec.encode_to(v, &mut buf)?;
                Some(buf)
            } else {
                None
            };
            if current.is_some() {
                shard.delete(&k)?;
            }
            Ok(bytes)
        })
    }

    fn count(&self, exact: bool) -> DbResult<u64> {
        Ok(if exact {
            self.tree.iter().count() as u64
        } else {
            self.tree.len() as u64
        })
    }

    fn apply_batch(&self, items: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
        // Decode every item in a single pass — short-circuits on first error,
        // so no partial writes occur on validation failure.
        let decoded_items: Vec<(K, Option<T>)> = items
            .into_iter()
            .map(|(key, val)| {
                if key.len() != size_of::<K>() {
                    return Err(DbError::Client("invalid key length in batch"));
                }
                let decoded_val = match val {
                    Some(v) => Some(
                        self.codec
                            .decode_from(&v)
                            .map_err(|_| DbError::Client("invalid value bytes in batch"))?,
                    ),
                    None => None,
                };
                Ok((K::from_bytes(&key), decoded_val))
            })
            .collect::<DbResult<_>>()?;
        // Apply from the decoded vec — no second decode.
        for (k, val) in decoded_items {
            match val {
                Some(decoded) => {
                    self.tree.put(&k, &decoded)?;
                }
                None => {
                    self.tree.delete(&k)?;
                }
            }
        }
        Ok(())
    }
}

// ─── VarTypedMapHandler ──────────────────────────────────────────────────────

/// RPC handler for [`VarTypedMap`](crate::VarTypedMap) — disk-resident typed values.
#[cfg(feature = "var-collections")]
pub struct VarTypedMapHandler<K, T, C, H: TypedWriteHook<K, T> = crate::NoHook>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
{
    pub name: String,
    pub typ_hash: u64,
    pub ty: armour_core::Typ,
    pub key_scheme: armour_core::KeyScheme,
    pub version: u16,
    pub map: Arc<crate::VarTypedMap<K, T, C, H>>,
    pub codec: Arc<C>,
    pub seq: Arc<super::seq::SeqGen>,
}

#[cfg(all(feature = "rpc", feature = "var-collections"))]
impl<K, T, C, H> RpcHandler for VarTypedMapHandler<K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    fn name(&self) -> &str {
        &self.name
    }

    fn info(&self) -> (u64, u16) {
        (self.typ_hash, self.version)
    }

    fn schema(&self) -> armour_rpc::SchemaResponse {
        armour_rpc::SchemaResponse {
            name: self.name.clone(),
            version: self.version,
            typ: self.ty,
            key_scheme: self.key_scheme,
        }
    }

    fn scannable(&self) -> bool {
        self.map.iter_view().is_some()
    }

    fn get(&self, key: &[u8]) -> DbResult<Option<ValueBytes>> {
        let key: K = check_key_len(key)?;
        match self.map.get(&key) {
            Some(val) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&val, &mut buf)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn contains(&self, key: &[u8]) -> DbResult<bool> {
        let k = check_key_len::<K>(key)?;
        Ok(self.map.contains(&k))
    }

    fn entry_len(&self, key: &[u8]) -> DbResult<Option<u32>> {
        let k = check_key_len::<K>(key)?;
        Ok(self.map.entry_len(&k))
    }

    fn first(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        self.map
            .first()?
            .map(|(key, value)| {
                let mut bytes = Vec::new();
                self.codec.encode_to(&value, &mut bytes)?;
                Ok((key.as_bytes().to_vec(), bytes))
            })
            .transpose()
    }

    fn last(&self) -> DbResult<Option<(KeyBytes, ValueBytes)>> {
        self.map
            .last()?
            .map(|(key, value)| {
                let mut bytes = Vec::new();
                self.codec.encode_to(&value, &mut bytes)?;
                Ok((key.as_bytes().to_vec(), bytes))
            })
            .transpose()
    }

    fn range(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<(KeyBytes, ValueBytes)>> {
        let view = self.map.iter_view().ok_or(DbError::NotImplemented)?;
        let start = bound_to_key_bound::<K>(&start)?;
        let end = bound_to_key_bound::<K>(&end)?;
        collect_range(
            view.range_bounds(start.as_ref(), end.as_ref()),
            limit,
            |value, bytes| self.codec.encode_to(value, bytes),
        )
    }

    fn range_keys(
        &self,
        start: Bound<KeyBytes>,
        end: Bound<KeyBytes>,
        limit: u32,
    ) -> DbResult<Vec<ValueBytes>> {
        let view = self.map.iter_view().ok_or(DbError::NotImplemented)?;
        let start = bound_to_key_bound::<K>(&start)?;
        let end = bound_to_key_bound::<K>(&end)?;
        Ok(collect_keys(
            view.keys_range_bounds(start.as_ref(), end.as_ref()),
            limit,
        ))
    }

    fn upsert(
        &self,
        key: UpsertKey,
        flag: Option<bool>,
        value: ValueBytes,
    ) -> DbResult<ValueBytes> {
        let (typed_key, key_bytes) = resolve_key::<K>(key, &self.seq, &self.name)?;
        // Pre-decode to validate value bytes before write.
        let decoded = self.codec.decode_from(&value)?;
        match flag {
            None => {
                self.map.put(&typed_key, &decoded)?;
                Ok(key_bytes)
            }
            Some(false) => self
                .map
                .atomic(&typed_key, |shard| {
                    if shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyExists);
                    }
                    shard.put(&typed_key, &decoded)?;
                    Ok(())
                })
                .map(|_| key_bytes),
            Some(true) => self
                .map
                .atomic(&typed_key, |shard| {
                    if !shard.try_contains(&typed_key)? {
                        return Err(DbError::KeyNotFound);
                    }
                    shard.put(&typed_key, &decoded)?;
                    Ok(())
                })
                .map(|_| key_bytes),
        }
    }

    fn remove(&self, key: &[u8], soft: bool) -> DbResult<()> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        self.map.delete(&k).map(|_| ())
    }

    fn take(&self, key: &[u8], soft: bool) -> DbResult<Option<ValueBytes>> {
        if soft {
            return Err(DbError::NotImplemented);
        }
        let k = check_key_len::<K>(key)?;
        let codec = Arc::clone(&self.codec);
        self.map.atomic(&k, |shard| {
            let current = shard.try_get(&k)?;
            let bytes = if let Some(ref v) = current {
                let mut buf = Vec::new();
                codec.encode_to(v, &mut buf)?;
                Some(buf)
            } else {
                None
            };
            if current.is_some() {
                shard.delete(&k)?;
            }
            Ok(bytes)
        })
    }

    fn count(&self, _exact: bool) -> DbResult<u64> {
        Ok(self.map.len() as u64)
    }

    fn apply_batch(&self, items: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
        // Decode every item in a single pass — short-circuits on first error,
        // so no partial writes occur on validation failure.
        let decoded_items: Vec<(K, Option<T>)> = items
            .into_iter()
            .map(|(key, val)| {
                if key.len() != size_of::<K>() {
                    return Err(DbError::Client("invalid key length in batch"));
                }
                let decoded_val = match val {
                    Some(v) => Some(
                        self.codec
                            .decode_from(&v)
                            .map_err(|_| DbError::Client("invalid value bytes in batch"))?,
                    ),
                    None => None,
                };
                Ok((K::from_bytes(&key), decoded_val))
            })
            .collect::<DbResult<_>>()?;
        // Apply from the decoded vec — no second decode.
        for (k, val) in decoded_items {
            match val {
                Some(decoded) => {
                    self.map.put(&k, &decoded)?;
                }
                None => {
                    self.map.delete(&k)?;
                }
            }
        }
        Ok(())
    }
}

// ─── InspectionHandler (VarTyped) ────────────────────────────────────────────

#[cfg(feature = "var-collections")]
impl<K, T, C, H> InspectionHandler for VarTypedTreeHandler<K, T, C, H>
where
    K: Key + Ord + Send + Sync,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
        Ok(InspectionCollection {
            name: self.name.clone(),
            schema: InspectionSchema {
                version: self.version,
                typ: self.ty,
                key_scheme: self.key_scheme,
                key_size: size_of::<K>() as u32,
                encoding: C::INSPECTION_ENCODING,
            },
            storage: StorageClass {
                model: ValueModel::VarTyped,
                shape: CollectionShape::Tree,
                backend: StorageBackend::Bitcask,
            },
            count: Some(self.tree.len() as u64),
            scan: ScanCapability::Ordered,
            direction: Some(iteration_direction(self.tree.config().reversed)),
        })
    }

    fn get_raw(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        let key = K::from_bytes(key);
        match self.tree.try_get(&key)? {
            Some(val) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&val, &mut buf)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn exact_count(&self) -> DbResult<u64> {
        self.tree.try_count_live_keys()
    }

    fn ordered_page(
        &self,
        bounds: &NormalizedBounds,
        limit: NonZeroU32,
    ) -> DbResult<InspectionPage> {
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        let start = bound_to_key_bound_owned::<K>(&bounds.start)?;
        let end = bound_to_key_bound_owned::<K>(&bounds.end)?;
        let codec = Arc::clone(&self.codec);
        collect_page(
            self.tree.try_range_bounds(start.as_ref(), end.as_ref()),
            limit,
            move |k, v| {
                let mut bytes = Vec::new();
                codec.encode_to(&v, &mut bytes)?;
                Ok(InspectionEntry {
                    key: k.as_bytes().to_vec(),
                    value: bytes,
                })
            },
        )
    }

    fn unordered_snapshot(
        &self,
        _max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError> {
        Err(InspectionError::UnsupportedOperation {
            name: self.name.clone(),
            operation: InspectionOperation::SnapshotUnordered,
        })
    }
}

#[cfg(feature = "var-collections")]
impl<K, T, C, H> InspectionHandler for VarTypedMapHandler<K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
        let iterable = self.map.iter_view().is_some();
        Ok(InspectionCollection {
            name: self.name.clone(),
            schema: InspectionSchema {
                version: self.version,
                typ: self.ty,
                key_scheme: self.key_scheme,
                key_size: size_of::<K>() as u32,
                encoding: C::INSPECTION_ENCODING,
            },
            storage: StorageClass {
                model: ValueModel::VarTyped,
                shape: CollectionShape::Map,
                backend: StorageBackend::Bitcask,
            },
            count: Some(self.map.len() as u64),
            scan: if iterable {
                ScanCapability::Ordered
            } else {
                ScanCapability::UnorderedSnapshot
            },
            direction: if iterable {
                Some(iteration_direction(self.map.config().reversed))
            } else {
                None
            },
        })
    }

    fn get_raw(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        let key = K::from_bytes(key);
        match self.map.try_get(&key)? {
            Some(val) => {
                let mut buf = Vec::new();
                self.codec.encode_to(&val, &mut buf)?;
                Ok(Some(buf))
            }
            None => Ok(None),
        }
    }

    fn exact_count(&self) -> DbResult<u64> {
        Ok(self.map.len() as u64)
    }

    fn ordered_page(
        &self,
        bounds: &NormalizedBounds,
        limit: NonZeroU32,
    ) -> DbResult<InspectionPage> {
        if self.map.iter_view().is_none() {
            return Err(DbError::NotImplemented);
        }
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        let start = bound_to_key_bound_owned::<K>(&bounds.start)?;
        let end = bound_to_key_bound_owned::<K>(&bounds.end)?;
        let batch = inspection_batch(limit, self.map.inspection_shard_count());
        let codec = Arc::clone(&self.codec);
        collect_page(
            self.map
                .inspection_range_bounds(start.as_ref(), end.as_ref(), batch),
            limit,
            move |k, v| {
                let mut bytes = Vec::new();
                codec.encode_to(&v, &mut bytes)?;
                Ok(InspectionEntry {
                    key: k.as_bytes().to_vec(),
                    value: bytes,
                })
            },
        )
    }

    fn unordered_snapshot(
        &self,
        max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError> {
        if self.map.iter_view().is_some() {
            return Err(InspectionError::UnsupportedOperation {
                name: self.name.clone(),
                operation: InspectionOperation::SnapshotUnordered,
            });
        }
        let exact_count = self.map.len() as u64;
        let cap = max_entries.get();
        if exact_count > u64::from(cap) {
            return Err(InspectionError::SnapshotTooLarge {
                limit: cap,
                observed: Some(exact_count),
            });
        }
        let codec = Arc::clone(&self.codec);
        let mut acc = SnapshotAccumulator::new(max_entries);
        let mut db_error = None;
        if let Err(error) = self.map.try_for_each_while(|k, v| {
            if db_error.is_some() {
                return std::ops::ControlFlow::Break(());
            }
            let mut bytes = Vec::new();
            if let Err(error) = codec.encode_to(&v, &mut bytes) {
                db_error = Some(error);
                return std::ops::ControlFlow::Break(());
            }
            acc.push(InspectionEntry {
                key: k.as_bytes().to_vec(),
                value: bytes,
            })
        }) {
            return Err(InspectionError::Database(error));
        }
        if let Some(error) = db_error {
            return Err(InspectionError::Database(error));
        }
        acc.finish()
    }
}

#[cfg(all(test, feature = "rpc"))]
mod range_helper_tests {
    use super::{DbError, UNLIMITED_RANGE_CAP, apply_limit, collect_keys, collect_range};

    #[test]
    fn explicit_limit_is_honored_exactly() {
        assert_eq!(apply_limit(0u32.., 3).count(), 3);
        assert_eq!(
            apply_limit(0u32..2, 10).count(),
            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.
        assert_eq!(
            apply_limit(std::iter::repeat(0u8), 0).count(),
            UNLIMITED_RANGE_CAP,
            "an infinite source must terminate at the cap, not hang"
        );
        // Anything smaller than the cap is still returned in full.
        assert_eq!(apply_limit(0u32..500, 0).count(), 500);
    }

    #[test]
    fn collect_range_honors_limit_and_propagates_encode_error() {
        let rows = (1_u64..=4).map(|n| (n.to_be_bytes(), n));
        let got = collect_range(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 = collect_range(
            std::iter::once((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!(
            collect_keys((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!(
            collect_keys(std::iter::repeat(1_u64.to_be_bytes()), 0).len(),
            UNLIMITED_RANGE_CAP,
        );
    }
}

#[cfg(test)]
mod inspection_handler_tests {
    use std::collections::HashSet;
    use std::num::NonZeroU32;
    use std::ops::Bound;
    use std::sync::Arc;

    use armour_core::{
        CollectionShape, GetType, KeyScheme, KeyType, ScalarTyp, StorageBackend, Typ, ValueModel,
    };
    use rapira::Rapira;
    use tempfile::tempdir;

    use super::{
        InspectionHandler, TypedMapHandler, TypedTreeHandler, ZeroMapHandler, ZeroTreeHandler,
    };
    use crate::armour::inspection::{IterationDirection, NormalizedBounds, ScanCapability};
    use crate::armour::seq::SeqGen;
    use crate::codec::{Codec, InspectionEncoding, RapiraCodec};
    use crate::durability::{Bitcask, Fixed};
    use crate::{CollectionMeta, Config, DbError, DbResult, FixedConfig, Key, 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_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,
        }
    }

    fn typed_tree_handler(
        env: &TestEnv,
        reversed: bool,
    ) -> TypedTreeHandler<Key8, Row, RapiraCodec, NoHook> {
        let mut cfg = Config::test();
        cfg.reversed = reversed;
        let tree =
            Arc::new(crate::TypedTree::open(env.path("typed_tree"), cfg, RapiraCodec).unwrap());
        TypedTreeHandler {
            name: Row::NAME.to_owned(),
            typ_hash: Row::TYPE.h(),
            ty: Row::TYPE,
            key_scheme: <Key8 as Key>::KEY_SCHEME,
            version: Row::VERSION,
            tree,
            codec: Arc::new(RapiraCodec),
            seq: Arc::clone(&env.seq),
        }
    }

    fn typed_map_handler(
        env: &TestEnv,
        iterable: bool,
        reversed: bool,
    ) -> TypedMapHandler<Key8, Row, RapiraCodec, NoHook> {
        let mut cfg = Config::test();
        cfg.iterable = iterable;
        cfg.reversed = reversed;
        let map = Arc::new(
            crate::TypedMap::open(
                env.path(if iterable {
                    "typed_map_iter"
                } else {
                    "typed_map_snap"
                }),
                cfg,
                RapiraCodec,
            )
            .unwrap(),
        );
        TypedMapHandler {
            name: Row::NAME.to_owned(),
            typ_hash: Row::TYPE.h(),
            ty: Row::TYPE,
            key_scheme: <Key8 as Key>::KEY_SCHEME,
            version: Row::VERSION,
            map,
            codec: Arc::new(RapiraCodec),
            seq: Arc::clone(&env.seq),
        }
    }

    fn opaque_tree_handler(env: &TestEnv) -> TypedTreeHandler<Key8, u64, OpaqueCodec, NoHook> {
        let tree = Arc::new(
            crate::TypedTree::open(env.path("opaque_tree"), Config::test(), OpaqueCodec).unwrap(),
        );
        TypedTreeHandler {
            name: "opaque_tree".to_owned(),
            typ_hash: Row::TYPE.h(),
            ty: Typ::Scalar(ScalarTyp::U64),
            key_scheme: KeyScheme::Typed(&[KeyType::U64]),
            version: 1,
            tree,
            codec: Arc::new(OpaqueCodec),
            seq: Arc::clone(&env.seq),
        }
    }

    fn zero_bitcask_tree_handler(
        env: &TestEnv,
        name: &str,
        reversed: bool,
    ) -> ZeroTreeHandler<Key8, 4, ZeroRow, NoHook, Bitcask> {
        let mut cfg = Config::test();
        cfg.reversed = reversed;
        let tree = Arc::new(
            crate::ZeroTree::<Key8, 4, ZeroRow, NoHook, Bitcask>::open(env.path(name), cfg)
                .unwrap(),
        );
        ZeroTreeHandler {
            name: ZeroRow::NAME.to_owned(),
            typ_hash: ZeroRow::TYPE.h(),
            ty: ZeroRow::TYPE,
            key_scheme: <Key8 as Key>::KEY_SCHEME,
            version: ZeroRow::VERSION,
            tree,
            seq: Arc::clone(&env.seq),
        }
    }

    fn zero_fixed_tree_handler(
        env: &TestEnv,
        name: &str,
        reversed: bool,
    ) -> ZeroTreeHandler<Key8, 4, ZeroRow, NoHook, Fixed> {
        let cfg = FixedConfig {
            reversed,
            ..FixedConfig::test()
        };
        let tree = Arc::new(
            crate::ZeroTree::<Key8, 4, ZeroRow, NoHook, Fixed>::open(env.path(name), cfg).unwrap(),
        );
        ZeroTreeHandler {
            name: ZeroRow::NAME.to_owned(),
            typ_hash: ZeroRow::TYPE.h(),
            ty: ZeroRow::TYPE,
            key_scheme: <Key8 as Key>::KEY_SCHEME,
            version: ZeroRow::VERSION,
            tree,
            seq: Arc::clone(&env.seq),
        }
    }

    fn zero_bitcask_map_handler(
        env: &TestEnv,
        name: &str,
        iterable: bool,
        reversed: bool,
    ) -> ZeroMapHandler<Key8, 4, ZeroRow, NoHook, Bitcask> {
        let mut cfg = Config::test();
        cfg.iterable = iterable;
        cfg.reversed = reversed;
        let map = Arc::new(
            crate::ZeroMap::<Key8, 4, ZeroRow, NoHook, Bitcask>::open(env.path(name), cfg).unwrap(),
        );
        ZeroMapHandler {
            name: ZeroRow::NAME.to_owned(),
            typ_hash: ZeroRow::TYPE.h(),
            ty: ZeroRow::TYPE,
            key_scheme: <Key8 as Key>::KEY_SCHEME,
            version: ZeroRow::VERSION,
            map,
            seq: Arc::clone(&env.seq),
        }
    }

    fn zero_fixed_map_handler(
        env: &TestEnv,
        name: &str,
        iterable: bool,
        reversed: bool,
    ) -> ZeroMapHandler<Key8, 4, ZeroRow, NoHook, Fixed> {
        let cfg = FixedConfig {
            iterable,
            reversed,
            ..FixedConfig::test()
        };
        let map = Arc::new(
            crate::ZeroMap::<Key8, 4, ZeroRow, NoHook, Fixed>::open(env.path(name), cfg).unwrap(),
        );
        ZeroMapHandler {
            name: ZeroRow::NAME.to_owned(),
            typ_hash: ZeroRow::TYPE.h(),
            ty: ZeroRow::TYPE,
            key_scheme: <Key8 as Key>::KEY_SCHEME,
            version: ZeroRow::VERSION,
            map,
            seq: Arc::clone(&env.seq),
        }
    }

    fn encode_row(row: &Row) -> Vec<u8> {
        let mut buf = Vec::new();
        RapiraCodec.encode_to(row, &mut buf).unwrap();
        buf
    }

    #[test]
    fn inspection_handler_metadata_typed_tree() {
        let env = TestEnv::new();
        let handler = typed_tree_handler(&env, false);
        handler.tree.put(&key(1), Row { n: 1 }).unwrap();
        let meta = handler.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));
    }

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

    #[test]
    fn inspection_handler_metadata_non_iterable_typed_map() {
        let env = TestEnv::new();
        let handler = typed_map_handler(&env, false, false);
        handler.map.put(&key(1), Row { n: 1 }).unwrap();
        let meta = handler.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 handler = zero_bitcask_tree_handler(&env, "ztree_bc", false);
        handler.tree.put(&key(1), &ZeroRow { n: 1 }).unwrap();
        let meta = handler.metadata().unwrap();
        assert_eq!(meta.storage.model, ValueModel::Zero);
        assert_eq!(meta.storage.backend, StorageBackend::Bitcask);
        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 handler = zero_fixed_tree_handler(&env, "ztree_fx", true);
        handler.tree.put(&key(1), &ZeroRow { n: 1 }).unwrap();
        let meta = handler.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 = zero_bitcask_map_handler(&env, "zmap_bc", true, false);
        bitcask.map.put(&key(1), &ZeroRow { n: 1 }).unwrap();
        let meta = bitcask.metadata().unwrap();
        assert_eq!(meta.storage.backend, StorageBackend::Bitcask);
        assert_eq!(meta.scan, ScanCapability::Ordered);
        let fixed = zero_fixed_map_handler(&env, "zmap_fx", false, false);
        fixed.map.put(&key(1), &ZeroRow { n: 1 }).unwrap();
        let meta = 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 handler = opaque_tree_handler(&env);
        handler.tree.put(&key(1), 42).unwrap();
        assert_eq!(
            handler.metadata().unwrap().schema.encoding,
            InspectionEncoding::Opaque
        );
    }

    #[test]
    fn inspection_handler_exact_typed_tree() {
        let env = TestEnv::new();
        let handler = typed_tree_handler(&env, false);
        let row = Row { n: 9 };
        handler.tree.put(&key(1), row.clone()).unwrap();
        assert_eq!(handler.get_raw(&key(1)).unwrap().unwrap(), encode_row(&row));
        assert!(handler.get_raw(&key(2)).unwrap().is_none());
        assert_eq!(handler.exact_count().unwrap(), 1);
    }

    #[test]
    fn inspection_handler_exact_non_iterable_map_count_without_view() {
        let env = TestEnv::new();
        let handler = typed_map_handler(&env, false, false);
        handler.map.put(&key(1), Row { n: 1 }).unwrap();
        handler.map.put(&key(2), Row { n: 2 }).unwrap();
        assert!(handler.map.iter_view().is_none());
        assert_eq!(handler.exact_count().unwrap(), 2);
        assert!(handler.get_raw(&key(9)).unwrap().is_none());
    }

    #[test]
    fn inspection_handler_page_ascending_three_rows_cursor() {
        let env = TestEnv::new();
        let handler = typed_tree_handler(&env, false);
        for id in 1..=3 {
            handler.tree.put(&key(id), Row { n: id }).unwrap();
        }
        let page = handler
            .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()));
    }

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

    #[test]
    fn inspection_handler_page_descending_three_rows_cursor() {
        let env = TestEnv::new();
        let handler = typed_tree_handler(&env, true);
        for id in 1..=3 {
            handler.tree.put(&key(id), Row { n: id }).unwrap();
        }
        let page = handler
            .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()));
    }

    #[test]
    fn inspection_handler_page_iterable_map_ascending() {
        let env = TestEnv::new();
        let handler = typed_map_handler(&env, true, false);
        for id in 1..=3 {
            handler.map.put(&key(id), Row { n: id }).unwrap();
        }
        let page = handler
            .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()));
    }

    #[test]
    fn inspection_handler_snapshot_exact_set_unordered() {
        let env = TestEnv::new();
        let handler = typed_map_handler(&env, false, false);
        handler.map.put(&key(1), Row { n: 10 }).unwrap();
        handler.map.put(&key(2), Row { n: 20 }).unwrap();
        let entries = handler
            .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()]));
    }

    #[test]
    fn inspection_handler_snapshot_preflight_and_visitor_stop() {
        let env = TestEnv::new();
        let handler = typed_map_handler(&env, false, false);
        for id in 1..=3 {
            handler.map.put(&key(id), Row { n: id }).unwrap();
        }
        let err = handler
            .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 _ = handler.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")]
    #[test]
    fn rpc_malformed_key_returns_key_not_found() {
        use super::RpcHandler;
        let env = TestEnv::new();
        let handler = typed_tree_handler(&env, false);
        let err = RpcHandler::get(&handler, &[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 super::{InspectionHandler, VarTypedMapHandler, VarTypedTreeHandler};
    use crate::armour::inspection::NormalizedBounds;
    use crate::armour::seq::SeqGen;
    use crate::test_faults::{POISON, PoisonCodec};
    use crate::{CollectionMeta, Config, DbError, Key, 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;
    }

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

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

    fn var_tree_handler() -> (
        tempfile::TempDir,
        Arc<SeqGen>,
        VarTypedTreeHandler<Key8, u64, PoisonCodec, 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(), PoisonCodec)
                .unwrap(),
        );
        let handler = VarTypedTreeHandler {
            name: Row::NAME.to_owned(),
            typ_hash: Row::TYPE.h(),
            ty: Row::TYPE,
            key_scheme: <Key8 as Key>::KEY_SCHEME,
            version: Row::VERSION,
            tree,
            codec: Arc::new(PoisonCodec),
            seq: Arc::clone(&seq),
        };
        (dir, seq, handler)
    }

    fn var_map_handler(
        iterable: bool,
    ) -> (
        tempfile::TempDir,
        Arc<SeqGen>,
        VarTypedMapHandler<Key8, u64, PoisonCodec, 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,
                PoisonCodec,
            )
            .unwrap(),
        );
        let handler = VarTypedMapHandler {
            name: Row::NAME.to_owned(),
            typ_hash: Row::TYPE.h(),
            ty: Row::TYPE,
            key_scheme: <Key8 as Key>::KEY_SCHEME,
            version: Row::VERSION,
            map,
            codec: Arc::new(PoisonCodec),
            seq: Arc::clone(&seq),
        };
        (dir, seq, handler)
    }

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

    #[test]
    fn inspection_var_map_get_propagates_decode_error() {
        let (_dir, _seq, handler) = var_map_handler(false);
        handler.map.put(&key(1), &POISON).unwrap();
        let err = handler.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, handler) = var_tree_handler();
        handler.tree.put(&key(1), &1).unwrap();
        handler.tree.put(&key(2), &POISON).unwrap();
        handler.tree.put(&key(3), &3).unwrap();
        let err = handler
            .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, handler) = var_map_handler(true);
        handler.map.put(&key(1), &1).unwrap();
        handler.map.put(&key(2), &POISON).unwrap();
        handler.map.put(&key(3), &3).unwrap();
        let err = handler
            .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, handler) = var_map_handler(false);
        handler.map.put(&key(1), &1).unwrap();
        handler.map.put(&key(2), &POISON).unwrap();
        handler.map.put(&key(3), &3).unwrap();
        let err = handler
            .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, handler) = var_tree_handler();
        handler.tree.put(&key(1), &1).unwrap();
        handler.tree.put(&key(2), &POISON).unwrap();
        handler.tree.put(&key(3), &3).unwrap();
        assert_eq!(handler.exact_count().unwrap(), 3);
        assert_eq!(
            handler.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, handler) = var_tree_handler();
        handler.tree.put(&key(1), &1).unwrap();
        handler.tree.put(&key(2), &2).unwrap();
        handler.tree.put(&key(3), &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 = handler.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, handler) = var_tree_handler();
        handler.tree.put(&key(1), &1).unwrap();
        handler.tree.put(&key(2), &POISON).unwrap();
        handler.tree.put(&key(3), &3).unwrap();
        let strict_rows = handler
            .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!(handler.exact_count().unwrap(), 3);
    }

    #[test]
    fn inspection_var_map_count_includes_poison_key() {
        let (_dir, _seq, handler) = var_map_handler(false);
        handler.map.put(&key(1), &1).unwrap();
        handler.map.put(&key(2), &POISON).unwrap();
        assert_eq!(handler.exact_count().unwrap(), 2);
    }
}