armdb 0.12.0

//! Read-only inspection types, errors, and shared navigation helpers.

use std::cmp::Ordering;
use std::num::NonZeroU32;
use std::ops::{Bound, ControlFlow};
use std::sync::Arc;

use armour_core::{KeyScheme, StorageClass, Typ};

use crate::error::{DbError, DbResult};
use crate::sync::lock;

use super::db::Db;

pub use crate::codec::InspectionEncoding;

/// How a collection can be scanned by the Inspector.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ScanCapability {
    /// Ordered range and prefix scans via byte-lexicographic bounds.
    Ordered,
    /// Full snapshot only; iteration order is unspecified.
    UnorderedSnapshot,
    /// Key counts via exact scan only; no ordered or snapshot reads.
    ExactOnly,
}

/// Byte-lexicographic iteration direction for ordered collections.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum IterationDirection {
    /// Smallest to largest key bytes.
    Ascending,
    /// Largest to smallest key bytes.
    Descending,
}

/// Whether inspection uses the cheap counter or a strict key scan.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CountMode {
    /// O(1) live-key counter when the handler exposes one.
    Current,
    /// Full key-only walk; includes unreadable values and propagates storage errors.
    Exact,
}

/// Inspector operation rejected by a handler's scan capability.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum InspectionOperation {
    /// [`Inspector::range`].
    Range,
    /// [`Inspector::prefix`].
    Prefix,
    /// [`Inspector::snapshot_unordered`].
    SnapshotUnordered,
}

/// Schema metadata returned for a registered collection.
#[derive(Clone, Debug)]
pub struct InspectionSchema {
    /// Collection migration / layout version.
    pub version: u16,
    /// Armour type descriptor for decoded values.
    pub typ: Typ,
    /// Key encoding scheme (e.g. branded ID vs raw bytes).
    pub key_scheme: KeyScheme,
    /// Fixed key width in bytes.
    pub key_size: u32,
    /// How values should be interpreted on read.
    pub encoding: InspectionEncoding,
}

/// Discovery metadata for one registered collection.
#[derive(Clone, Debug)]
pub struct InspectionCollection {
    /// Registered collection name.
    pub name: String,
    /// Schema and encoding metadata.
    pub schema: InspectionSchema,
    /// Storage family and backend class.
    pub storage: StorageClass,
    /// Cheap live-key count when available; `None` if only exact count is supported.
    pub count: Option<u64>,
    /// Which scan operations this collection supports.
    pub scan: ScanCapability,
    /// Iteration direction when [`scan`](Self::scan) is [`ScanCapability::Ordered`].
    pub direction: Option<IterationDirection>,
}

/// One raw key/value pair from an inspection read.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InspectionEntry {
    /// Raw key bytes (fixed width for the collection).
    pub key: Vec<u8>,
    /// Raw value bytes (encoding per [`InspectionSchema::encoding`]).
    pub value: Vec<u8>,
}

/// Bounded ordered scan page with optional continuation.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InspectionPage {
    /// Up to the requested limit of entries in scan order.
    pub entries: Vec<InspectionEntry>,
    /// Pass as `after` to continue pagination; `None` when this page is the last.
    pub next_key: Option<Vec<u8>>,
}

/// Inspector-specific failures. Storage and codec errors surface as [`Self::Database`].
#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum InspectionError {
    /// No handler registered under this name.
    #[error("collection not found: {name}")]
    CollectionNotFound { name: String },
    /// Key length does not match the collection's fixed key size.
    #[error("invalid key length: expected {expected}, got {actual}")]
    InvalidKey { expected: usize, actual: usize },
    /// Range or prefix bounds are inconsistent with collection order.
    #[error("invalid bounds: {reason}")]
    InvalidBounds { reason: &'static str },
    /// Handler [`ScanCapability`] does not allow this operation.
    #[error("unsupported operation {operation:?} on collection {name}")]
    UnsupportedOperation {
        name: String,
        operation: InspectionOperation,
    },
    /// Unordered snapshot exceeded the caller's entry limit.
    #[error("unordered snapshot exceeds limit {limit} (observed {observed:?})")]
    SnapshotTooLarge { limit: u32, observed: Option<u64> },
    /// Underlying storage or codec error.
    #[error(transparent)]
    Database(#[from] DbError),
}

/// Crate-private handler contract for registered collections.
pub(crate) trait InspectionHandler: Send + Sync {
    fn metadata(&self) -> Result<InspectionCollection, InspectionError>;
    fn key_group_bits(&self) -> u32 {
        0
    }
    fn get_raw(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>>;
    fn exact_count(&self) -> DbResult<u64>;
    fn ordered_page(
        &self,
        bounds: &NormalizedBounds,
        limit: NonZeroU32,
    ) -> DbResult<InspectionPage>;
    fn unordered_snapshot(
        &self,
        max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError>;
}

/// Borrowed read-only inspector over collections opened through [`Db`].
///
/// # Examples
///
/// ```
/// # use armdb::{CollectionMeta, Config, NoHook};
/// # use armdb::armour::{Db, CountMode};
/// # use armour_core::GetType;
/// # use zerocopy::{FromBytes, Immutable, IntoBytes};
/// # #[derive(Clone, Copy, Debug, PartialEq, GetType, FromBytes, Immutable, IntoBytes)]
/// # #[repr(C)]
/// # struct Widget { id: u64 }
/// # impl CollectionMeta for Widget {
/// #     type SelfId = [u8; 8];
/// #     const NAME: &'static str = "widgets";
/// #     const VERSION: u16 = 1;
/// # }
/// # let dir = tempfile::tempdir().unwrap();
/// # let db = Db::open_test(dir.path()).unwrap();
/// # let _tree = db.open_zero_tree::<Widget, { std::mem::size_of::<Widget>() }, _>(Config::test(), NoHook, &[]).unwrap();
/// let inspector = db.inspector();
/// let names: Vec<_> = inspector
///     .collections()
///     .unwrap()
///     .into_iter()
///     .map(|c| c.name)
///     .collect();
/// assert_eq!(names, vec![Widget::NAME.to_owned()]);
/// ```
pub struct Inspector<'db> {
    pub(crate) db: &'db Db,
}

impl Inspector<'_> {
    fn handler(&self, name: &str) -> Result<Arc<dyn InspectionHandler>, InspectionError> {
        let guard = lock(&self.db.inspection_handlers);
        guard
            .get(name)
            .cloned()
            .ok_or_else(|| InspectionError::CollectionNotFound {
                name: name.to_owned(),
            })
    }

    /// List every registered collection sorted by name.
    pub fn collections(&self) -> Result<Vec<InspectionCollection>, InspectionError> {
        let handlers: Vec<Arc<dyn InspectionHandler>> = {
            let guard = lock(&self.db.inspection_handlers);
            guard.values().cloned().collect()
        };
        let mut out = Vec::with_capacity(handlers.len());
        for handler in handlers {
            out.push(handler.metadata()?);
        }
        out.sort_by(|a, b| a.name.cmp(&b.name));
        Ok(out)
    }

    /// Return schema metadata for one collection.
    pub fn schema(&self, name: &str) -> Result<InspectionSchema, InspectionError> {
        let handler = self.handler(name)?;
        Ok(handler.metadata()?.schema)
    }

    /// Return the key-routing group width for one collection.
    pub fn key_group_bits(&self, name: &str) -> Result<u32, InspectionError> {
        Ok(self.handler(name)?.key_group_bits())
    }

    /// Exact key lookup. Validates key length before dispatch.
    pub fn get(&self, name: &str, key: &[u8]) -> Result<Option<InspectionEntry>, InspectionError> {
        let handler = self.handler(name)?;
        let key_size = handler.metadata()?.schema.key_size as usize;
        if key.len() != key_size {
            return Err(InspectionError::InvalidKey {
                expected: key_size,
                actual: key.len(),
            });
        }
        let value = handler.get_raw(key).map_err(InspectionError::Database)?;
        Ok(value.map(|value| InspectionEntry {
            key: key.to_vec(),
            value,
        }))
    }

    /// Cheap or exact live-key count.
    pub fn count(&self, name: &str, mode: CountMode) -> Result<u64, InspectionError> {
        let handler = self.handler(name)?;
        match mode {
            CountMode::Current => match handler.metadata()?.count {
                Some(n) => Ok(n),
                None => Err(InspectionError::Database(DbError::NotImplemented)),
            },
            CountMode::Exact => handler.exact_count().map_err(InspectionError::Database),
        }
    }

    /// Bounded ordered range scan.
    pub fn range(
        &self,
        name: &str,
        start: Bound<&[u8]>,
        end: Bound<&[u8]>,
        after: Option<&[u8]>,
        limit: NonZeroU32,
    ) -> Result<InspectionPage, InspectionError> {
        let handler = self.handler(name)?;
        let meta = handler.metadata()?;
        if meta.scan != ScanCapability::Ordered {
            return Err(InspectionError::UnsupportedOperation {
                name: name.to_owned(),
                operation: InspectionOperation::Range,
            });
        }
        let direction = meta.direction.unwrap_or(IterationDirection::Ascending);
        let bounds = normalize_range(meta.schema.key_size as usize, direction, start, end, after)?;
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        handler
            .ordered_page(&bounds, limit)
            .map_err(InspectionError::Database)
    }

    /// Bounded ordered prefix scan.
    pub fn prefix(
        &self,
        name: &str,
        prefix: &[u8],
        after: Option<&[u8]>,
        limit: NonZeroU32,
    ) -> Result<InspectionPage, InspectionError> {
        let handler = self.handler(name)?;
        let meta = handler.metadata()?;
        if meta.scan != ScanCapability::Ordered {
            return Err(InspectionError::UnsupportedOperation {
                name: name.to_owned(),
                operation: InspectionOperation::Prefix,
            });
        }
        let direction = meta.direction.unwrap_or(IterationDirection::Ascending);
        let bounds = normalize_prefix(meta.schema.key_size as usize, direction, prefix, after)?;
        if bounds.empty {
            return Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            });
        }
        handler
            .ordered_page(&bounds, limit)
            .map_err(InspectionError::Database)
    }

    /// Bounded unordered snapshot for non-iterable collections.
    pub fn snapshot_unordered(
        &self,
        name: &str,
        max_entries: NonZeroU32,
    ) -> Result<Vec<InspectionEntry>, InspectionError> {
        let handler = self.handler(name)?;
        let meta = handler.metadata()?;
        if meta.scan != ScanCapability::UnorderedSnapshot {
            return Err(InspectionError::UnsupportedOperation {
                name: name.to_owned(),
                operation: InspectionOperation::SnapshotUnordered,
            });
        }
        handler.unordered_snapshot(max_entries)
    }
}

/// Owned scan bounds after validation and continuation folding.
#[derive(Debug)]
pub(crate) struct NormalizedBounds {
    pub start: Bound<Vec<u8>>,
    pub end: Bound<Vec<u8>>,
    pub empty: bool,
}

pub(crate) fn normalize_range(
    key_size: usize,
    direction: IterationDirection,
    start: Bound<&[u8]>,
    end: Bound<&[u8]>,
    after: Option<&[u8]>,
) -> Result<NormalizedBounds, InspectionError> {
    validate_bound(start, key_size)?;
    validate_bound(end, key_size)?;
    if let Some(after_key) = after {
        validate_key(after_key, key_size)?;
    }

    let mut lower = bound_to_owned(start);
    let mut upper = bound_to_owned(end);

    if is_invalid_bounds_error(&lower, &upper) {
        return Err(InspectionError::InvalidBounds {
            reason: "start greater than end",
        });
    }

    if let Some(after_key) = after {
        if !key_in_domain(after_key, &lower, &upper) {
            return Err(InspectionError::InvalidBounds {
                reason: "after outside domain",
            });
        }
        let after_excluded = Bound::Excluded(after_key.to_vec());
        match direction {
            IterationDirection::Ascending => {
                lower = more_restrictive_lower(&lower, &after_excluded);
            }
            IterationDirection::Descending => {
                upper = more_restrictive_upper(&upper, &after_excluded);
            }
        }
    }

    let empty = is_empty_range(&lower, &upper);
    Ok(NormalizedBounds {
        start: lower,
        end: upper,
        empty,
    })
}

pub(crate) fn normalize_prefix(
    key_size: usize,
    direction: IterationDirection,
    prefix: &[u8],
    after: Option<&[u8]>,
) -> Result<NormalizedBounds, InspectionError> {
    if prefix.len() > key_size {
        return Err(InspectionError::InvalidBounds {
            reason: "prefix longer than key size",
        });
    }

    let (start, end) = if prefix.is_empty() {
        (Bound::Unbounded, Bound::Unbounded)
    } else {
        let mut start_key = vec![0u8; key_size];
        start_key[..prefix.len()].copy_from_slice(prefix);
        (
            Bound::Included(start_key),
            prefix_end_bound(prefix, key_size),
        )
    };

    let start = match &start {
        Bound::Included(k) => Bound::Included(k.as_slice()),
        Bound::Excluded(k) => Bound::Excluded(k.as_slice()),
        Bound::Unbounded => Bound::Unbounded,
    };
    let end = match &end {
        Bound::Included(k) => Bound::Included(k.as_slice()),
        Bound::Excluded(k) => Bound::Excluded(k.as_slice()),
        Bound::Unbounded => Bound::Unbounded,
    };
    normalize_range(key_size, direction, start, end, after)
}

/// One `ordered_page` worth of entries plus the single-row lookahead that
/// decides `next_key`.
///
/// The lookahead row is observed but **never encoded** — reading it is what
/// proves a further page exists, and encoding it would be wasted work on every
/// full page. Driven by the visitor-based Inspection [`Adapter`](crate::access::Adapter),
/// which cannot use an iterator because `Access::try_scan_range` hands rows to
/// a callback; tests drive it directly (see `page_from` below).
pub(crate) struct PageAccumulator {
    limit: u32,
    entries: Vec<InspectionEntry>,
    last_key: Option<Vec<u8>>,
    has_more: bool,
}

impl PageAccumulator {
    pub(crate) fn new(limit: NonZeroU32) -> Self {
        let cap = usize::try_from(limit.get()).unwrap_or(usize::MAX).min(1024);
        Self {
            limit: limit.get(),
            entries: Vec::with_capacity(cap),
            last_key: None,
            has_more: false,
        }
    }

    /// `Break` once the page is full: the caller must stop, and `encode` is not
    /// called for that row.
    pub(crate) fn push_with(
        &mut self,
        encode: impl FnOnce() -> DbResult<InspectionEntry>,
    ) -> DbResult<ControlFlow<()>> {
        if self.entries.len() as u64 >= u64::from(self.limit) {
            self.has_more = true;
            return Ok(ControlFlow::Break(()));
        }
        let entry = encode()?;
        self.last_key = Some(entry.key.clone());
        self.entries.push(entry);
        Ok(ControlFlow::Continue(()))
    }

    pub(crate) fn finish(self) -> InspectionPage {
        let next_key = match (self.has_more, self.last_key) {
            (true, Some(key)) => Some(key),
            _ => None,
        };
        InspectionPage {
            entries: self.entries,
            next_key,
        }
    }
}

pub(crate) struct SnapshotAccumulator {
    limit: u32,
    observed: u64,
    entries: Vec<InspectionEntry>,
    over_cap: bool,
}

impl SnapshotAccumulator {
    pub(crate) fn new(limit: NonZeroU32) -> Self {
        let cap = usize::try_from(limit.get()).unwrap_or(usize::MAX).min(1024);
        Self {
            limit: limit.get(),
            observed: 0,
            entries: Vec::with_capacity(cap),
            over_cap: false,
        }
    }

    pub(crate) fn push(&mut self, entry: InspectionEntry) -> ControlFlow<()> {
        self.observed += 1;
        if self.observed > u64::from(self.limit) {
            self.over_cap = true;
            return ControlFlow::Break(());
        }
        self.entries.push(entry);
        ControlFlow::Continue(())
    }

    pub(crate) fn finish(self) -> Result<Vec<InspectionEntry>, InspectionError> {
        if self.over_cap {
            Err(InspectionError::SnapshotTooLarge {
                limit: self.limit,
                observed: Some(self.observed),
            })
        } else {
            Ok(self.entries)
        }
    }
}

fn validate_key(key: &[u8], key_size: usize) -> Result<(), InspectionError> {
    if key.len() != key_size {
        return Err(InspectionError::InvalidKey {
            expected: key_size,
            actual: key.len(),
        });
    }
    Ok(())
}

fn validate_bound(bound: Bound<&[u8]>, key_size: usize) -> Result<(), InspectionError> {
    match bound {
        Bound::Included(key) | Bound::Excluded(key) => validate_key(key, key_size),
        Bound::Unbounded => Ok(()),
    }
}

fn bound_to_owned(bound: Bound<&[u8]>) -> Bound<Vec<u8>> {
    match bound {
        Bound::Included(key) => Bound::Included(key.to_vec()),
        Bound::Excluded(key) => Bound::Excluded(key.to_vec()),
        Bound::Unbounded => Bound::Unbounded,
    }
}

fn bound_key(bound: &Bound<Vec<u8>>) -> Option<&[u8]> {
    match bound {
        Bound::Included(key) | Bound::Excluded(key) => Some(key),
        Bound::Unbounded => None,
    }
}

fn is_invalid_bounds_error(start: &Bound<Vec<u8>>, end: &Bound<Vec<u8>>) -> bool {
    match (bound_key(start), bound_key(end)) {
        (Some(s), Some(e)) => s > e,
        _ => false,
    }
}

fn key_in_domain(key: &[u8], start: &Bound<Vec<u8>>, end: &Bound<Vec<u8>>) -> bool {
    key_satisfies_lower(key, start) && key_satisfies_upper(key, end)
}

fn key_satisfies_lower(key: &[u8], bound: &Bound<Vec<u8>>) -> bool {
    match bound {
        Bound::Unbounded => true,
        Bound::Included(k) => key >= k.as_slice(),
        Bound::Excluded(k) => key > k.as_slice(),
    }
}

fn key_satisfies_upper(key: &[u8], bound: &Bound<Vec<u8>>) -> bool {
    match bound {
        Bound::Unbounded => true,
        Bound::Included(k) => key <= k.as_slice(),
        Bound::Excluded(k) => key < k.as_slice(),
    }
}

fn cmp_lower_bounds(a: &Bound<Vec<u8>>, b: &Bound<Vec<u8>>) -> Ordering {
    match (a, b) {
        (Bound::Unbounded, Bound::Unbounded) => Ordering::Equal,
        (Bound::Unbounded, _) => Ordering::Less,
        (_, Bound::Unbounded) => Ordering::Greater,
        (Bound::Included(ka), Bound::Included(kb)) => ka.cmp(kb),
        (Bound::Excluded(ka), Bound::Excluded(kb)) => ka.cmp(kb),
        (Bound::Included(ka), Bound::Excluded(kb)) => match ka.cmp(kb) {
            Ordering::Equal => Ordering::Less,
            other => other,
        },
        (Bound::Excluded(ka), Bound::Included(kb)) => match ka.cmp(kb) {
            Ordering::Equal => Ordering::Greater,
            other => other,
        },
    }
}

fn cmp_upper_bounds(a: &Bound<Vec<u8>>, b: &Bound<Vec<u8>>) -> Ordering {
    match (a, b) {
        (Bound::Unbounded, Bound::Unbounded) => Ordering::Equal,
        (Bound::Unbounded, _) => Ordering::Greater,
        (_, Bound::Unbounded) => Ordering::Less,
        (Bound::Included(ka), Bound::Included(kb)) => ka.cmp(kb),
        (Bound::Excluded(ka), Bound::Excluded(kb)) => ka.cmp(kb),
        (Bound::Included(ka), Bound::Excluded(kb)) => match ka.cmp(kb) {
            Ordering::Equal => Ordering::Greater,
            other => other,
        },
        (Bound::Excluded(ka), Bound::Included(kb)) => match ka.cmp(kb) {
            Ordering::Equal => Ordering::Less,
            other => other,
        },
    }
}

fn more_restrictive_lower(a: &Bound<Vec<u8>>, b: &Bound<Vec<u8>>) -> Bound<Vec<u8>> {
    if cmp_lower_bounds(a, b) != Ordering::Less {
        a.clone()
    } else {
        b.clone()
    }
}

fn more_restrictive_upper(a: &Bound<Vec<u8>>, b: &Bound<Vec<u8>>) -> Bound<Vec<u8>> {
    if cmp_upper_bounds(a, b) != Ordering::Greater {
        a.clone()
    } else {
        b.clone()
    }
}

fn is_empty_range(lower: &Bound<Vec<u8>>, upper: &Bound<Vec<u8>>) -> bool {
    match (lower, upper) {
        (Bound::Unbounded, Bound::Unbounded) => false,
        (Bound::Unbounded, _) | (_, Bound::Unbounded) => false,
        (Bound::Included(l), Bound::Included(u)) => l > u,
        (Bound::Included(l), Bound::Excluded(u)) => l >= u,
        (Bound::Excluded(l), Bound::Included(u)) => l >= u,
        (Bound::Excluded(l), Bound::Excluded(u)) => l >= u,
    }
}

fn prefix_end_bound(prefix: &[u8], key_size: usize) -> Bound<Vec<u8>> {
    let mut incremented = prefix.to_vec();
    let mut carry = true;
    for byte in incremented.iter_mut().rev() {
        if carry {
            if *byte == 0xFF {
                *byte = 0x00;
            } else {
                *byte += 1;
                carry = false;
                break;
            }
        }
    }
    if carry {
        Bound::Unbounded
    } else {
        let mut end = vec![0u8; key_size];
        end[..incremented.len()].copy_from_slice(&incremented);
        Bound::Excluded(end)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use armour_core::{
        CollectionShape, KeyScheme, KeyType, ScalarTyp, StorageBackend, StorageClass, Typ,
        ValueModel,
    };

    fn key8(byte: u8) -> Vec<u8> {
        vec![byte, 0, 0, 0, 0, 0, 0, 0]
    }

    fn key8_pair(a: u8, b: u8) -> Vec<u8> {
        let mut k = vec![0u8; 8];
        k[0] = a;
        k[1] = b;
        k
    }

    /// Drives `PageAccumulator` the way both the deleted `collect_page` and the
    /// Inspection adapter do: encode each row until the accumulator asks to stop.
    fn page_from<K, V>(
        rows: Vec<DbResult<(K, V)>>,
        limit: NonZeroU32,
        mut encode: impl FnMut(K, V) -> DbResult<InspectionEntry>,
    ) -> DbResult<InspectionPage> {
        let mut page = PageAccumulator::new(limit);
        for row in rows {
            let (key, value) = row?;
            if page.push_with(|| encode(key, value))?.is_break() {
                break;
            }
        }
        Ok(page.finish())
    }

    #[test]
    fn inspection_error_variants_are_matchable() {
        let _ = InspectionError::CollectionNotFound {
            name: "users".into(),
        };
        let _ = InspectionError::InvalidKey {
            expected: 8,
            actual: 7,
        };
        let _ = InspectionError::InvalidBounds {
            reason: "start greater than end",
        };
        let _ = InspectionError::UnsupportedOperation {
            name: "map".into(),
            operation: InspectionOperation::Range,
        };
        let _ = InspectionError::SnapshotTooLarge {
            limit: 2,
            observed: Some(3),
        };
        let _ = InspectionError::Database(DbError::KeyNotFound);
    }

    #[test]
    fn database_error_preserves_source() {
        let error = InspectionError::from(DbError::CorruptedEntry { offset: 7 });
        assert!(matches!(
            error,
            InspectionError::Database(DbError::CorruptedEntry { offset: 7 })
        ));
    }

    #[test]
    fn inspection_error_public_types_construct() {
        let _ = ScanCapability::Ordered;
        let _ = ScanCapability::UnorderedSnapshot;
        let _ = ScanCapability::ExactOnly;
        let _ = IterationDirection::Ascending;
        let _ = IterationDirection::Descending;
        let _ = CountMode::Current;
        let _ = CountMode::Exact;
        let _ = InspectionOperation::Range;
        let _ = InspectionOperation::Prefix;
        let _ = InspectionOperation::SnapshotUnordered;

        let schema = InspectionSchema {
            version: 1,
            typ: Typ::Scalar(ScalarTyp::U64),
            key_scheme: KeyScheme::Typed(&[KeyType::U64]),
            key_size: 8,
            encoding: InspectionEncoding::Rapira,
        };
        let _ = InspectionCollection {
            name: "users".into(),
            schema,
            storage: StorageClass {
                model: ValueModel::Typed,
                shape: CollectionShape::Tree,
                backend: StorageBackend::Bitcask,
            },
            count: Some(0),
            scan: ScanCapability::Ordered,
            direction: Some(IterationDirection::Ascending),
        };
        let _ = InspectionEntry {
            key: key8(1),
            value: vec![9],
        };
        let _ = InspectionPage {
            entries: vec![],
            next_key: None,
        };
    }

    #[test]
    fn inspection_range_ascending_after_tightens_lower_bound() {
        let start_key = key8(1);
        let end_key = key8(9);
        let start = Bound::Included(start_key.as_slice());
        let end = Bound::Included(end_key.as_slice());
        let after = key8(3);
        let bounds =
            normalize_range(8, IterationDirection::Ascending, start, end, Some(&after)).unwrap();
        assert!(!bounds.empty);
        assert_eq!(bounds.start, Bound::Excluded(key8(3)));
        assert_eq!(bounds.end, Bound::Included(key8(9)));
    }

    #[test]
    fn inspection_range_descending_after_tightens_upper_bound() {
        let start_key = key8(1);
        let end_key = key8(9);
        let start = Bound::Included(start_key.as_slice());
        let end = Bound::Included(end_key.as_slice());
        let after = key8(7);
        let bounds =
            normalize_range(8, IterationDirection::Descending, start, end, Some(&after)).unwrap();
        assert!(!bounds.empty);
        assert_eq!(bounds.start, Bound::Included(key8(1)));
        assert_eq!(bounds.end, Bound::Excluded(key8(7)));
    }

    #[test]
    fn inspection_range_excluded_wins_over_included_at_same_key() {
        let start_key = key8(5);
        let start = Bound::Included(start_key.as_slice());
        let end = Bound::Unbounded;
        let after = key8(5);
        let bounds =
            normalize_range(8, IterationDirection::Ascending, start, end, Some(&after)).unwrap();
        assert_eq!(bounds.start, Bound::Excluded(key8(5)));
    }

    #[test]
    fn inspection_range_start_greater_than_end_is_invalid() {
        let start_key = key8(9);
        let end_key = key8(1);
        let err = normalize_range(
            8,
            IterationDirection::Ascending,
            Bound::Included(start_key.as_slice()),
            Bound::Included(end_key.as_slice()),
            None,
        )
        .unwrap_err();
        assert!(matches!(err, InspectionError::InvalidBounds { .. }));
    }

    #[test]
    fn inspection_range_equal_included_is_singleton() {
        let k = key8(4);
        let bounds = normalize_range(
            8,
            IterationDirection::Ascending,
            Bound::Included(k.as_slice()),
            Bound::Included(k.as_slice()),
            None,
        )
        .unwrap();
        assert!(!bounds.empty);
        assert_eq!(bounds.start, Bound::Included(k.clone()));
        assert_eq!(bounds.end, Bound::Included(k));
    }

    #[test]
    fn inspection_range_equal_with_excluded_is_empty() {
        let k = key8(4);
        let bounds = normalize_range(
            8,
            IterationDirection::Ascending,
            Bound::Included(k.as_slice()),
            Bound::Excluded(k.as_slice()),
            None,
        )
        .unwrap();
        assert!(bounds.empty);
    }

    #[test]
    fn inspection_range_after_outside_domain_is_invalid() {
        let start_key = key8(2);
        let end_key = key8(5);
        let after_key = key8(9);
        let err = normalize_range(
            8,
            IterationDirection::Ascending,
            Bound::Included(start_key.as_slice()),
            Bound::Included(end_key.as_slice()),
            Some(after_key.as_slice()),
        )
        .unwrap_err();
        assert!(matches!(err, InspectionError::InvalidBounds { .. }));
    }

    #[test]
    fn inspection_range_terminal_after_produces_empty_range() {
        let start_key = key8(2);
        let end_key = key8(5);
        let after_key = key8(5);
        let bounds = normalize_range(
            8,
            IterationDirection::Ascending,
            Bound::Included(start_key.as_slice()),
            Bound::Included(end_key.as_slice()),
            Some(after_key.as_slice()),
        )
        .unwrap();
        assert!(bounds.empty);
    }

    #[test]
    fn inspection_range_rejects_wrong_length_keys() {
        let short = vec![1u8; 7];
        let long = vec![1u8; 9];
        for key in [&short, &long] {
            let err = normalize_range(
                8,
                IterationDirection::Ascending,
                Bound::Included(key.as_slice()),
                Bound::Unbounded,
                None,
            )
            .unwrap_err();
            assert!(matches!(err, InspectionError::InvalidKey { .. }));
        }
    }

    #[test]
    fn inspection_prefix_empty_is_full_domain() {
        let bounds = normalize_prefix(8, IterationDirection::Ascending, &[], None).unwrap();
        assert!(!bounds.empty);
        assert_eq!(bounds.start, Bound::Unbounded);
        assert_eq!(bounds.end, Bound::Unbounded);
    }

    #[test]
    fn inspection_prefix_ordinary_prefix() {
        let prefix = [0x01, 0x02];
        let bounds = normalize_prefix(8, IterationDirection::Ascending, &prefix, None).unwrap();
        assert!(!bounds.empty);
        assert_eq!(bounds.start, Bound::Included(key8_pair(1, 2)));
        assert_eq!(bounds.end, Bound::Excluded(key8_pair(1, 3)));
    }

    #[test]
    fn inspection_prefix_all_ff_uses_unbounded_upper() {
        let prefix = [0xFF; 8];
        let bounds = normalize_prefix(8, IterationDirection::Ascending, &prefix, None).unwrap();
        assert!(!bounds.empty);
        assert_eq!(bounds.start, Bound::Included(prefix.to_vec()));
        assert_eq!(bounds.end, Bound::Unbounded);
    }

    #[test]
    fn inspection_prefix_overlong_is_invalid() {
        let prefix = vec![0u8; 9];
        let err = normalize_prefix(8, IterationDirection::Ascending, &prefix, None).unwrap_err();
        assert!(matches!(err, InspectionError::InvalidBounds { .. }));
    }

    #[test]
    fn inspection_prefix_after_outside_domain_is_invalid() {
        let prefix = [0x01];
        let after_key = key8(9);
        let err = normalize_prefix(
            8,
            IterationDirection::Ascending,
            &prefix,
            Some(after_key.as_slice()),
        )
        .unwrap_err();
        assert!(matches!(err, InspectionError::InvalidBounds { .. }));
    }

    #[test]
    fn inspection_page_fewer_than_limit_has_no_cursor() {
        let rows = vec![Ok((key8(1), 10u64)), Ok((key8(2), 20u64))];
        let page = page_from(rows, NonZeroU32::new(3).unwrap(), |k, v| {
            Ok(InspectionEntry {
                key: k,
                value: v.to_le_bytes().to_vec(),
            })
        })
        .unwrap();
        assert_eq!(page.entries.len(), 2);
        assert!(page.next_key.is_none());
    }

    #[test]
    fn inspection_page_exactly_limit_has_no_cursor() {
        let rows = vec![Ok((key8(1), 1u64)), Ok((key8(2), 2u64))];
        let page = page_from(rows, NonZeroU32::new(2).unwrap(), |k, v| {
            Ok(InspectionEntry {
                key: k,
                value: v.to_le_bytes().to_vec(),
            })
        })
        .unwrap();
        assert_eq!(page.entries.len(), 2);
        assert!(page.next_key.is_none());
    }

    #[test]
    fn inspection_page_limit_plus_one_sets_cursor_to_last_key() {
        let rows = vec![
            Ok((key8(1), 1u64)),
            Ok((key8(2), 2u64)),
            Ok((key8(3), 3u64)),
        ];
        let page = page_from(rows, NonZeroU32::new(2).unwrap(), |k, v| {
            Ok(InspectionEntry {
                key: k,
                value: v.to_le_bytes().to_vec(),
            })
        })
        .unwrap();
        assert_eq!(page.entries.len(), 2);
        assert_eq!(page.next_key, Some(key8(2)));
    }

    #[test]
    fn inspection_page_lookahead_error_fails_whole_page() {
        let rows: Vec<DbResult<(Vec<u8>, u64)>> =
            vec![Ok((key8(1), 1)), Err(DbError::CorruptedEntry { offset: 9 })];
        let err = page_from(rows, NonZeroU32::new(1).unwrap(), |k, v| {
            Ok(InspectionEntry {
                key: k,
                value: v.to_le_bytes().to_vec(),
            })
        })
        .unwrap_err();
        assert!(matches!(err, DbError::CorruptedEntry { offset: 9 }));
    }

    #[test]
    fn inspection_page_u32_max_uses_checked_capacity() {
        let page = page_from(
            Vec::<DbResult<(Vec<u8>, u64)>>::new(),
            NonZeroU32::new(u32::MAX).unwrap(),
            |k, v| {
                Ok(InspectionEntry {
                    key: k,
                    value: v.to_le_bytes().to_vec(),
                })
            },
        )
        .unwrap();
        assert!(page.entries.is_empty());
        assert!(page.entries.capacity() <= 1024);
    }

    #[test]
    fn inspection_snapshot_third_push_breaks_and_finish_reports_too_large() {
        let mut acc = SnapshotAccumulator::new(NonZeroU32::new(2).unwrap());
        let entry = |n: u8| InspectionEntry {
            key: key8(n),
            value: vec![n],
        };
        assert!(acc.push(entry(1)).is_continue());
        assert!(acc.push(entry(2)).is_continue());
        assert!(acc.push(entry(3)).is_break());
        let err = acc.finish().unwrap_err();
        assert!(matches!(
            err,
            InspectionError::SnapshotTooLarge {
                limit: 2,
                observed: Some(3),
            }
        ));
    }

    #[test]
    fn inspection_snapshot_two_pushes_finish_successfully() {
        let mut acc = SnapshotAccumulator::new(NonZeroU32::new(2).unwrap());
        let entry = |n: u8| InspectionEntry {
            key: key8(n),
            value: vec![n],
        };
        assert!(acc.push(entry(1)).is_continue());
        assert!(acc.push(entry(2)).is_continue());
        let entries = acc.finish().unwrap();
        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].key, key8(1));
        assert_eq!(entries[1].key, key8(2));
    }
}

#[cfg(test)]
mod inspection_facade_tests {
    use super::*;
    use std::ops::Bound;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

    use armour_core::{
        CollectionShape, KeyScheme, KeyType, ScalarTyp, StorageBackend, StorageClass, Typ,
        ValueModel,
    };

    use crate::armour::Db;
    use crate::codec::InspectionEncoding;

    struct FakeState {
        metadata_calls: AtomicUsize,
        get_raw_calls: AtomicUsize,
        exact_count_calls: AtomicUsize,
        ordered_page_calls: AtomicUsize,
        unordered_snapshot_calls: AtomicUsize,
        metadata_fail: bool,
        count: Option<u64>,
        scan: ScanCapability,
        key_size: u32,
        get_raw_present: AtomicBool,
        get_raw_error: bool,
        exact_count_value: u64,
        exact_count_error: bool,
        reenter_db: Option<Arc<Db>>,
        reenter_name: String,
        reenter_guard: AtomicUsize,
    }

    struct FakeHandler {
        name: &'static str,
        state: Arc<FakeState>,
    }

    impl InspectionHandler for FakeHandler {
        fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
            self.state.metadata_calls.fetch_add(1, Ordering::SeqCst);
            if self.state.metadata_fail {
                return Err(InspectionError::CollectionNotFound {
                    name: "metadata-fail".into(),
                });
            }
            if let Some(db) = &self.state.reenter_db {
                if self.state.reenter_guard.fetch_add(1, Ordering::SeqCst) == 0 {
                    let _ = db.inspector().schema(&self.state.reenter_name);
                }
            }
            Ok(InspectionCollection {
                name: self.name.to_owned(),
                schema: InspectionSchema {
                    version: 1,
                    typ: Typ::Scalar(ScalarTyp::U64),
                    key_scheme: KeyScheme::Typed(&[KeyType::U64]),
                    key_size: self.state.key_size,
                    encoding: InspectionEncoding::Rapira,
                },
                storage: StorageClass {
                    model: ValueModel::Typed,
                    shape: CollectionShape::Tree,
                    backend: StorageBackend::Bitcask,
                },
                count: self.state.count,
                scan: self.state.scan,
                direction: Some(IterationDirection::Ascending),
            })
        }

        fn get_raw(&self, _: &[u8]) -> DbResult<Option<Vec<u8>>> {
            self.state.get_raw_calls.fetch_add(1, Ordering::SeqCst);
            if self.state.get_raw_error {
                return Err(DbError::CorruptedEntry { offset: 9 });
            }
            if self.state.get_raw_present.load(Ordering::SeqCst) {
                Ok(Some(vec![9]))
            } else {
                Ok(None)
            }
        }

        fn exact_count(&self) -> DbResult<u64> {
            self.state.exact_count_calls.fetch_add(1, Ordering::SeqCst);
            if self.state.exact_count_error {
                Err(DbError::CorruptedEntry { offset: 1 })
            } else {
                Ok(self.state.exact_count_value)
            }
        }

        fn ordered_page(&self, _: &NormalizedBounds, _: NonZeroU32) -> DbResult<InspectionPage> {
            self.state.ordered_page_calls.fetch_add(1, Ordering::SeqCst);
            Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            })
        }

        fn unordered_snapshot(
            &self,
            _: NonZeroU32,
        ) -> Result<Vec<InspectionEntry>, InspectionError> {
            self.state
                .unordered_snapshot_calls
                .fetch_add(1, Ordering::SeqCst);
            Ok(vec![])
        }
    }

    struct FakeCollection(&'static str);
    impl crate::armour::Collection for FakeCollection {
        fn name(&self) -> &str {
            self.0
        }
        fn len(&self) -> usize {
            0
        }
        fn compact(&self) -> DbResult<usize> {
            Ok(0)
        }
        fn flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn periodic_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn clean_shutdown(&self) -> DbResult<()> {
            Ok(())
        }
    }

    fn seed_fake(db: &Db, name: &'static str, state: Arc<FakeState>) -> DbResult<()> {
        let collection: Arc<dyn crate::armour::Collection> = Arc::new(FakeCollection(name));
        let inspection: Arc<dyn InspectionHandler> = Arc::new(FakeHandler { name, state });
        #[cfg(feature = "rpc")]
        let rpc = None;
        db.publish_collection_for_test(
            name,
            collection,
            inspection,
            #[cfg(feature = "rpc")]
            rpc,
            None,
        )
    }

    fn key8(byte: u8) -> Vec<u8> {
        vec![byte, 0, 0, 0, 0, 0, 0, 0]
    }

    #[test]
    fn inspection_facade_collections_and_schema() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: Some(3),
            scan: ScanCapability::Ordered,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: false,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(&db, "widgets", state.clone()).unwrap();

        let inspector = db.inspector();
        let collections = inspector.collections().unwrap();
        assert_eq!(collections.len(), 1);
        assert_eq!(collections[0].name, "widgets");
        assert_eq!(state.metadata_calls.load(Ordering::SeqCst), 1);

        let schema = inspector.schema("widgets").unwrap();
        assert_eq!(schema.key_size, 8);
        assert_eq!(state.metadata_calls.load(Ordering::SeqCst), 2);
        assert_eq!(inspector.key_group_bits("widgets").unwrap(), 0);
    }

    #[test]
    fn inspection_facade_missing_collection() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let err = db.inspector().schema("missing").unwrap_err();
        assert!(matches!(
            err,
            InspectionError::CollectionNotFound { name } if name == "missing"
        ));
    }

    #[test]
    fn inspection_facade_get_present_absent_and_invalid_key() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: Some(0),
            scan: ScanCapability::Ordered,
            key_size: 8,
            get_raw_present: AtomicBool::new(true),
            get_raw_error: false,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(&db, "items", state.clone()).unwrap();
        let inspector = db.inspector();

        let entry = inspector.get("items", &key8(1)).unwrap().unwrap();
        assert_eq!(entry.value, vec![9]);
        assert_eq!(state.get_raw_calls.load(Ordering::SeqCst), 1);

        state.get_raw_present.store(false, Ordering::SeqCst);
        assert!(inspector.get("items", &key8(2)).unwrap().is_none());
        assert_eq!(state.get_raw_calls.load(Ordering::SeqCst), 2);

        let err = inspector.get("items", &[1u8; 7]).unwrap_err();
        assert!(matches!(err, InspectionError::InvalidKey { .. }));
        assert_eq!(state.get_raw_calls.load(Ordering::SeqCst), 2);
    }

    #[test]
    fn inspection_facade_count_current_and_exact() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: Some(5),
            scan: ScanCapability::Ordered,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: false,
            exact_count_value: 7,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(&db, "counts", state.clone()).unwrap();
        let inspector = db.inspector();

        assert_eq!(inspector.count("counts", CountMode::Current).unwrap(), 5);
        assert_eq!(state.exact_count_calls.load(Ordering::SeqCst), 0);
        assert_eq!(inspector.count("counts", CountMode::Exact).unwrap(), 7);
        assert_eq!(state.exact_count_calls.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn inspection_facade_count_current_none_is_not_implemented() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: None,
            scan: ScanCapability::Ordered,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: false,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(&db, "no_count", state.clone()).unwrap();
        let err = db
            .inspector()
            .count("no_count", CountMode::Current)
            .unwrap_err();
        assert!(matches!(
            err,
            InspectionError::Database(DbError::NotImplemented)
        ));
        assert_eq!(state.exact_count_calls.load(Ordering::SeqCst), 0);
    }

    #[test]
    fn inspection_facade_capability_rejection_before_dispatch() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: Some(0),
            scan: ScanCapability::UnorderedSnapshot,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: false,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(&db, "snap", state.clone()).unwrap();
        let inspector = db.inspector();
        let limit = NonZeroU32::new(1).unwrap();

        let err = inspector
            .range("snap", Bound::Unbounded, Bound::Unbounded, None, limit)
            .unwrap_err();
        assert!(matches!(
            err,
            InspectionError::UnsupportedOperation {
                operation: InspectionOperation::Range,
                ..
            }
        ));
        assert_eq!(state.ordered_page_calls.load(Ordering::SeqCst), 0);

        let err = inspector.prefix("snap", &[], None, limit).unwrap_err();
        assert!(matches!(
            err,
            InspectionError::UnsupportedOperation {
                operation: InspectionOperation::Prefix,
                ..
            }
        ));

        let entries = inspector.snapshot_unordered("snap", limit).unwrap();
        assert!(entries.is_empty());
        assert_eq!(state.unordered_snapshot_calls.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn inspection_facade_exact_only_rejects_multi_entry_ops() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: Some(0),
            scan: ScanCapability::ExactOnly,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: false,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(&db, "exact", state.clone()).unwrap();
        let inspector = db.inspector();
        let limit = NonZeroU32::new(1).unwrap();

        let err = inspector
            .range("exact", Bound::Unbounded, Bound::Unbounded, None, limit)
            .unwrap_err();
        assert!(matches!(err, InspectionError::UnsupportedOperation { .. }));

        let err = inspector.prefix("exact", &[], None, limit).unwrap_err();
        assert!(matches!(err, InspectionError::UnsupportedOperation { .. }));

        let err = inspector.snapshot_unordered("exact", limit).unwrap_err();
        assert!(matches!(err, InspectionError::UnsupportedOperation { .. }));
        assert_eq!(state.ordered_page_calls.load(Ordering::SeqCst), 0);
        assert_eq!(state.unordered_snapshot_calls.load(Ordering::SeqCst), 0);
    }

    #[test]
    fn inspection_facade_storage_error_conversion() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: Some(0),
            scan: ScanCapability::Ordered,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: true,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(&db, "bad", state).unwrap();
        let err = db.inspector().get("bad", &key8(1)).unwrap_err();
        assert!(matches!(
            err,
            InspectionError::Database(DbError::CorruptedEntry { offset: 9 })
        ));
    }

    #[test]
    fn inspection_facade_collections_metadata_failure_is_all_or_nothing() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let ok_state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: Some(0),
            scan: ScanCapability::Ordered,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: false,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        let bad_state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: true,
            count: Some(0),
            scan: ScanCapability::Ordered,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: false,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: None,
            reenter_name: String::new(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(&db, "alpha", ok_state).unwrap();
        seed_fake(&db, "beta", bad_state).unwrap();

        let err = db.inspector().collections().unwrap_err();
        assert!(matches!(
            err,
            InspectionError::CollectionNotFound { name } if name == "metadata-fail"
        ));
    }

    #[test]
    fn inspection_lock_registry_released_before_handler_reads() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let state = Arc::new(FakeState {
            metadata_calls: AtomicUsize::new(0),
            get_raw_calls: AtomicUsize::new(0),
            exact_count_calls: AtomicUsize::new(0),
            ordered_page_calls: AtomicUsize::new(0),
            unordered_snapshot_calls: AtomicUsize::new(0),
            metadata_fail: false,
            count: Some(0),
            scan: ScanCapability::Ordered,
            key_size: 8,
            get_raw_present: AtomicBool::new(false),
            get_raw_error: false,
            exact_count_value: 0,
            exact_count_error: false,
            reenter_db: Some(Arc::clone(&db)),
            reenter_name: "reenter".into(),
            reenter_guard: AtomicUsize::new(0),
        });
        seed_fake(db.as_ref(), "reenter", state).unwrap();
        db.inspector().schema("reenter").unwrap();
    }
}