armdb 0.14.2

// Two stacked gates rather than one `cfg(all(test, …))`: clippy reads
// `allow-unwrap-in-tests` off a literal `cfg(test)`, and merging the two
// hides it, which turns every `unwrap()` below into an `unwrap_used` warning.
#![cfg(test)]
#![cfg(feature = "typed-tree")]

use std::cell::RefCell;
#[cfg(feature = "var-collections")]
use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::ops::Bound;
use std::path::Path;
use std::rc::Rc;

#[cfg(feature = "fjall")]
use armour_core::GetType;
use armour_core::{KeyScheme, KeyType};
use zerocopy::{FromBytes, Immutable, IntoBytes};

use crate::codec::Codec;
use crate::config::Config;
use crate::const_map::ConstMap;
use crate::const_tree::ConstTree;
use crate::durability::Fixed;
use crate::error::{DbError, DbResult};
use crate::hook::NoHook;
use crate::impl_key_zerocopy;
use crate::key::Key;
use crate::map_index::{Dir, MapIterView};
#[cfg(feature = "var-collections")]
use crate::scope::VarTypedTreeRowIter;
use crate::scope::key_math;
use crate::scope::{
    BulkScan, ChunkedPage, MapScan, MappedPage, OrderedScan, OwnedRows, PageSpec, Scope,
    StopReason, StrictRows, TreeScan,
};
#[cfg(feature = "var-collections")]
use crate::scope::{merge_sorted, page_chunks};
use crate::typed_map::TypedMap;
use crate::typed_tree::TypedTree;
#[cfg(feature = "var-collections")]
use crate::var_map::VarMap;
#[cfg(feature = "var-collections")]
use crate::var_tree::VarTree;
#[cfg(feature = "var-collections")]
use crate::var_typed_map::VarTypedMap;
#[cfg(feature = "var-collections")]
use crate::var_typed_tree::VarTypedTree;
#[cfg(feature = "fjall")]
use crate::{CollectionMeta, RapiraCodec};
use crate::{FixedConfig, ZeroMap, ZeroTree};

mod bulk;

#[derive(
    Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, FromBytes, IntoBytes, Immutable,
)]
#[repr(C)]
struct K3 {
    a: u64,
    b: u64,
    c: u64,
}
impl_key_zerocopy!(
    K3,
    KeyScheme::Typed(&[KeyType::U64, KeyType::U64, KeyType::U64])
);

#[cfg(feature = "var-collections")]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, FromBytes, IntoBytes, Immutable)]
#[repr(C)]
struct ChannelKey {
    channel: u64,
    msg: u64,
}

#[cfg(feature = "var-collections")]
impl_key_zerocopy!(ChannelKey, KeyScheme::Typed(&[KeyType::U64, KeyType::U64]));

#[cfg(feature = "var-collections")]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, FromBytes, IntoBytes, Immutable)]
#[repr(C)]
struct PostKey {
    post: u64,
    msg: u64,
}

#[cfg(feature = "var-collections")]
impl_key_zerocopy!(PostKey, KeyScheme::Typed(&[KeyType::U64, KeyType::U64]));

#[cfg(feature = "var-collections")]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, FromBytes, IntoBytes, Immutable)]
#[repr(C)]
struct DirectKey {
    user: u64,
    msg: u64,
}

#[cfg(feature = "var-collections")]
impl_key_zerocopy!(DirectKey, KeyScheme::Typed(&[KeyType::U64, KeyType::U64]));

#[cfg(feature = "var-collections")]
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, FromBytes, IntoBytes, Immutable)]
#[repr(C)]
struct BodyKey {
    kind: u64,
    owner: u64,
    msg: u64,
}

#[cfg(feature = "var-collections")]
impl_key_zerocopy!(
    BodyKey,
    KeyScheme::Typed(&[KeyType::U64, KeyType::U64, KeyType::U64])
);

#[cfg(feature = "var-collections")]
#[derive(Clone, Copy, Debug)]
struct Candidate {
    msg: u64,
    body: BodyKey,
}

#[cfg(feature = "var-collections")]
impl PartialEq for Candidate {
    fn eq(&self, other: &Self) -> bool {
        self.msg == other.msg
    }
}

#[cfg(feature = "var-collections")]
impl Eq for Candidate {}

#[cfg(feature = "var-collections")]
impl PartialOrd for Candidate {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

#[cfg(feature = "var-collections")]
impl Ord for Candidate {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.msg.cmp(&other.msg)
    }
}

#[cfg(feature = "var-collections")]
enum CandidateRows<'a> {
    Channel(VarTypedTreeRowIter<'a, ChannelKey, u64, U64Codec, NoHook>),
    Post(VarTypedTreeRowIter<'a, PostKey, u64, U64Codec, NoHook>),
    Direct(VarTypedTreeRowIter<'a, DirectKey, u64, U64Codec, NoHook>),
}

#[cfg(feature = "var-collections")]
impl Iterator for CandidateRows<'_> {
    type Item = Candidate;

    fn next(&mut self) -> Option<Self::Item> {
        match self {
            Self::Channel(rows) => rows.next().map(|(key, _)| Candidate {
                msg: key.msg,
                body: BodyKey {
                    kind: 1,
                    owner: key.channel,
                    msg: key.msg,
                },
            }),
            Self::Post(rows) => rows.next().map(|(key, _)| Candidate {
                msg: key.msg,
                body: BodyKey {
                    kind: 2,
                    owner: key.post,
                    msg: key.msg,
                },
            }),
            Self::Direct(rows) => rows.next().map(|(key, _)| Candidate {
                msg: key.msg,
                body: BodyKey {
                    kind: 3,
                    owner: key.user,
                    msg: key.msg,
                },
            }),
        }
    }
}

#[derive(Clone, Copy)]
struct U64Codec;

impl Codec<u64> for U64Codec {
    fn encode_to(&self, value: &u64, buf: &mut Vec<u8>) -> DbResult<()> {
        buf.clear();
        buf.extend_from_slice(&value.to_be_bytes());
        Ok(())
    }

    fn decode_from(&self, bytes: &[u8]) -> DbResult<u64> {
        let arr: [u8; 8] = bytes
            .try_into()
            .map_err(|_| crate::error::DbError::CorruptedEntry { offset: 0 })?;
        Ok(u64::from_be_bytes(arr))
    }
}

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

fn prefix_match<const N: usize>(k: &K3, prefix: &K3) -> bool {
    (N == 0 || k.a == prefix.a) && (N < 2 || k.b == prefix.b) && (N < 3 || k.c == prefix.c)
}

fn field_offset(n: usize) -> usize {
    n * size_of::<u64>()
}

/// Consumer tail-saturation oracle — copied from the spec table; must not call
/// `key_math::saturate_range_bounds`.
///
/// Caveat: it is a *transcription* of the production fill table and it reuses
/// `key_math::prefix_to_end_bound` for the scope edge, so an identical mistake in
/// both places would pass. The independent check lives in `key_math::tests`, which
/// asserts concrete expected key values rather than mirroring the algorithm.
fn manual_saturate_range_bounds(
    scope_key: &K3,
    n: usize,
    m: usize,
    start: Bound<&K3>,
    end: Bound<&K3>,
) -> (Bound<K3>, Bound<K3>) {
    let n_off = field_offset(n);
    let m_off = field_offset(m);
    // Mirror production: Unbounded → scope edge, not collection-wide.
    let (scope_start, scope_end) = {
        let mut start_k = K3::zeroed();
        Key::as_bytes_mut(&mut start_k)[..n_off]
            .copy_from_slice(&Key::as_bytes(scope_key)[..n_off]);
        let end_b =
            crate::scope::key_math::prefix_to_end_bound::<K3>(&Key::as_bytes(scope_key)[..n_off]);
        (Bound::Included(start_k), end_b)
    };
    let sat = |b: Bound<&K3>, is_start: bool| -> Bound<K3> {
        match b {
            Bound::Unbounded => {
                if is_start {
                    scope_start
                } else {
                    scope_end
                }
            }
            Bound::Included(k) | Bound::Excluded(k) => {
                let fill = match (&b, is_start) {
                    (Bound::Included(_), true) => 0x00u8,
                    (Bound::Excluded(_), true) => 0xFF,
                    (Bound::Included(_), false) => 0xFF,
                    (Bound::Excluded(_), false) => 0x00,
                    _ => unreachable!(),
                };
                let mut out = K3::zeroed();
                Key::as_bytes_mut(&mut out)[..n_off]
                    .copy_from_slice(&Key::as_bytes(scope_key)[..n_off]);
                Key::as_bytes_mut(&mut out)[n_off..m_off]
                    .copy_from_slice(&Key::as_bytes(k)[n_off..m_off]);
                Key::as_bytes_mut(&mut out)[m_off..].fill(fill);
                match b {
                    Bound::Included(_) => Bound::Included(out),
                    Bound::Excluded(_) => Bound::Excluded(out),
                    Bound::Unbounded => unreachable!(),
                }
            }
        }
    };
    (sat(start, true), sat(end, false))
}

fn bound_ref(b: &Bound<K3>) -> Bound<&K3> {
    match b {
        Bound::Included(k) => Bound::Included(k),
        Bound::Excluded(k) => Bound::Excluded(k),
        Bound::Unbounded => Bound::Unbounded,
    }
}

trait ScopeHarness {
    type Scan<'a>: OrderedScan<Key = K3> + OwnedRows
    where
        Self: 'a;

    fn open(dir: &Path) -> Self;
    fn insert(&self, key: K3);
    fn delete(&self, key: &K3);
    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N>;
    fn all_keys(&self) -> BTreeSet<K3>;
    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3>;
    fn brute_prefix<const N: usize>(&self, key: &K3) -> BTreeSet<K3> {
        self.all_keys()
            .into_iter()
            .filter(|k| prefix_match::<N>(k, key))
            .collect()
    }
    /// Multi-shard iterable map — run cursor contract tests.
    const MULTI_SHARD_MAP: bool = false;
    /// Tree family — run cursor contract tests.
    const IS_TREE: bool = false;
}

struct ConstTreeHarnessWrapper<const REVERSED: bool>(ConstTree<K3, 8>);

fn open_const_tree(dir: &Path, reversed: bool) -> ConstTree<K3, 8> {
    let mut cfg = Config::test();
    cfg.shard_count = 1;
    cfg.reversed = reversed;
    ConstTree::<K3, 8>::open(dir, cfg).expect("open const tree")
}

impl<const REVERSED: bool> ScopeHarness for ConstTreeHarnessWrapper<REVERSED> {
    type Scan<'a> = TreeScan<&'a ConstTree<K3, 8>>;
    const IS_TREE: bool = true;

    fn open(dir: &Path) -> Self {
        Self(open_const_tree(dir, REVERSED))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, &val_bytes(key.b)).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key)
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        self.0
            .keys_range_bounds_dir(bound_ref(start), bound_ref(end), dir)
            .collect()
    }
}

type ConstTreeHarness = ConstTreeHarnessWrapper<false>;
type ConstTreeReversedHarness = ConstTreeHarnessWrapper<true>;

struct TypedTreeHarness(TypedTree<K3, u64, U64Codec>);

impl ScopeHarness for TypedTreeHarness {
    type Scan<'a> = TreeScan<&'a TypedTree<K3, u64, U64Codec>>;
    const IS_TREE: bool = true;

    fn open(dir: &Path) -> Self {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.reversed = false;
        Self(TypedTree::open(dir, cfg, U64Codec).expect("open typed tree"))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, key.b).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key)
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        self.0
            .keys_range_bounds_dir(bound_ref(start), bound_ref(end), dir)
            .collect()
    }
}

#[cfg(feature = "var-collections")]
struct VarTypedTreeHarness(VarTypedTree<K3, u64, U64Codec>);

#[cfg(feature = "var-collections")]
impl ScopeHarness for VarTypedTreeHarness {
    type Scan<'a> = TreeScan<&'a VarTypedTree<K3, u64, U64Codec>>;
    const IS_TREE: bool = true;

    fn open(dir: &Path) -> Self {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.reversed = false;
        Self(VarTypedTree::open(dir, cfg, U64Codec).expect("open var typed tree"))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, &key.b).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key)
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        self.0
            .keys_range_bounds_dir(bound_ref(start), bound_ref(end), dir)
            .collect()
    }
}

struct ZeroTreeHarness(ZeroTree<K3, 8, u64, NoHook, Fixed>);

impl ScopeHarness for ZeroTreeHarness {
    type Scan<'a> = TreeScan<&'a ZeroTree<K3, 8, u64, NoHook, Fixed>>;
    const IS_TREE: bool = true;

    fn open(dir: &Path) -> Self {
        Self(
            ZeroTree::<K3, 8, u64, NoHook, Fixed>::open(dir, FixedConfig::test())
                .expect("open fixed zero tree"),
        )
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, &key.b).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key)
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        self.0
            .keys_range_bounds_dir(bound_ref(start), bound_ref(end), dir)
            .collect()
    }
}

#[cfg(feature = "var-collections")]
struct VarTreeHarness(VarTree<K3>);

#[cfg(feature = "var-collections")]
impl ScopeHarness for VarTreeHarness {
    type Scan<'a> = TreeScan<&'a VarTree<K3>>;
    const IS_TREE: bool = true;

    fn open(dir: &Path) -> Self {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.reversed = false;
        Self(VarTree::open(dir, cfg).expect("open var tree"))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, &val_bytes(key.b)).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key)
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        self.0
            .keys_range_bounds_dir(bound_ref(start), bound_ref(end), dir)
            .collect()
    }
}

struct ConstMapHarnessWrapper<const REVERSED: bool>(ConstMap<K3, 8>);

impl<const REVERSED: bool> ScopeHarness for ConstMapHarnessWrapper<REVERSED> {
    type Scan<'a> = MapScan<&'a ConstMap<K3, 8>>;
    const MULTI_SHARD_MAP: bool = true;

    fn open(dir: &Path) -> Self {
        let cfg = Config::balanced()
            .shard_count(3)
            .hints(true)
            .iterable(true)
            .reversed(REVERSED)
            .build();
        Self(ConstMap::<K3, 8>::open(dir, cfg).expect("open const map"))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, &val_bytes(key.b)).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key).expect("iterable map")
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter_view().unwrap().iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        MapIterView::new(&self.0, dir)
            .keys_range_bounds(bound_ref(start), bound_ref(end))
            .collect()
    }
}

type ConstMapHarness = ConstMapHarnessWrapper<false>;
type ConstMapReversedHarness = ConstMapHarnessWrapper<true>;

struct TypedMapHarness(TypedMap<K3, u64, U64Codec>);

impl ScopeHarness for TypedMapHarness {
    type Scan<'a> = MapScan<&'a TypedMap<K3, u64, U64Codec>>;
    const MULTI_SHARD_MAP: bool = true;

    fn open(dir: &Path) -> Self {
        let cfg = Config::balanced()
            .shard_count(3)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build();
        Self(TypedMap::open(dir, cfg, U64Codec).expect("open typed map"))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, key.b).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key).expect("iterable map")
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter_view().unwrap().iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        MapIterView::new(&self.0, dir)
            .keys_range_bounds(bound_ref(start), bound_ref(end))
            .collect()
    }
}

struct ZeroMapHarness(ZeroMap<K3, 8, u64>);

impl ScopeHarness for ZeroMapHarness {
    type Scan<'a> = MapScan<&'a ZeroMap<K3, 8, u64>>;
    const MULTI_SHARD_MAP: bool = true;

    fn open(dir: &Path) -> Self {
        let cfg = Config::balanced()
            .shard_count(3)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build();
        Self(ZeroMap::<K3, 8, u64>::open(dir, cfg).expect("open zero map"))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, &key.b).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key).expect("iterable map")
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter_view().unwrap().iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        MapIterView::new(&self.0, dir)
            .keys_range_bounds(bound_ref(start), bound_ref(end))
            .collect()
    }
}

#[cfg(feature = "var-collections")]
struct VarMapHarness(VarMap<K3>);

#[cfg(feature = "var-collections")]
impl ScopeHarness for VarMapHarness {
    type Scan<'a> = MapScan<&'a VarMap<K3>>;
    const MULTI_SHARD_MAP: bool = true;

    fn open(dir: &Path) -> Self {
        let cfg = Config::balanced()
            .shard_count(3)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build();
        Self(VarMap::open(dir, cfg).expect("open var map"))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, &val_bytes(key.b)).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key).expect("iterable map")
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter_view().unwrap().iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        MapIterView::new(&self.0, dir)
            .keys_range_bounds(bound_ref(start), bound_ref(end))
            .collect()
    }
}

#[cfg(feature = "var-collections")]
struct VarTypedMapHarness(VarTypedMap<K3, u64, U64Codec>);

#[cfg(feature = "var-collections")]
impl ScopeHarness for VarTypedMapHarness {
    type Scan<'a> = MapScan<&'a VarTypedMap<K3, u64, U64Codec>>;
    const MULTI_SHARD_MAP: bool = true;

    fn open(dir: &Path) -> Self {
        let cfg = Config::balanced()
            .shard_count(3)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build();
        Self(VarTypedMap::open(dir, cfg, U64Codec).expect("open var typed map"))
    }

    fn insert(&self, key: K3) {
        self.0.insert(&key, &key.b).unwrap();
    }

    fn delete(&self, key: &K3) {
        self.0.delete(key).unwrap();
    }

    fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
        self.0.scope::<N>(key).expect("iterable map")
    }

    fn all_keys(&self) -> BTreeSet<K3> {
        self.0.iter_view().unwrap().iter().map(|(k, _)| k).collect()
    }

    fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
        MapIterView::new(self.0.as_inner(), dir)
            .keys_range_bounds(bound_ref(start), bound_ref(end))
            .collect()
    }
}

fn assert_bulk<H: ScopeHarness>()
where
    for<'a> H::Scan<'a>: BulkScan + StrictRows,
{
    fn requires_bulk<'a, S: BulkScan + StrictRows + 'a, const N: usize>(scope: Scope<'a, S, N>) {
        let _ = scope;
    }

    let _ = requires_bulk::<H::Scan<'_>, 0>;
}

#[test]
fn bulk_scan_shape_matrix() {
    assert_bulk::<ConstTreeHarness>();
    assert_bulk::<TypedTreeHarness>();
    assert_bulk::<ZeroTreeHarness>();
    #[cfg(feature = "var-collections")]
    assert_bulk::<VarTreeHarness>();
    #[cfg(feature = "var-collections")]
    assert_bulk::<VarTypedTreeHarness>();
    assert_bulk::<ConstMapHarness>();
    assert_bulk::<TypedMapHarness>();
    assert_bulk::<ZeroMapHarness>();
    #[cfg(feature = "var-collections")]
    assert_bulk::<VarMapHarness>();
    #[cfg(feature = "var-collections")]
    assert_bulk::<VarTypedMapHarness>();
    #[cfg(feature = "fjall")]
    assert_bulk::<LsmTreeHarness>();
}

#[cfg(feature = "fjall")]
mod lsm_harness {
    use super::*;
    use crate::lsm::{LsmConfig, LsmTree};
    use rapira::Rapira;

    #[derive(Clone, Debug, PartialEq, Eq, GetType, Rapira)]
    pub(super) struct K3Payload {
        pub(super) v: u64,
    }

    impl CollectionMeta for K3Payload {
        type SelfId = K3;
        const NAME: &'static str = "k3_scope_test";
    }

    pub(super) struct LsmTreeHarness {
        _dir: tempfile::TempDir,
        tree: LsmTree<K3, K3Payload, RapiraCodec, NoHook>,
    }

    impl ScopeHarness for LsmTreeHarness {
        type Scan<'a> = TreeScan<&'a LsmTree<K3, K3Payload, RapiraCodec, NoHook>>;
        const IS_TREE: bool = true;

        fn open(dir: &Path) -> Self {
            let owned = tempfile::TempDir::new_in(dir).expect("tempdir");
            let db = fjall::Database::builder(owned.path())
                .open()
                .expect("fjall db");
            let tree =
                LsmTree::open(db, LsmConfig::default(), RapiraCodec, NoHook).expect("lsm tree");
            Self { _dir: owned, tree }
        }

        fn insert(&self, key: K3) {
            self.tree.put(&key, &K3Payload { v: key.b }).unwrap();
        }

        fn delete(&self, key: &K3) {
            self.tree.delete(key).unwrap();
        }

        fn scope<'a, const N: usize>(&'a self, key: &K3) -> Scope<'a, Self::Scan<'a>, N> {
            self.tree.scope::<N>(key)
        }

        fn all_keys(&self) -> BTreeSet<K3> {
            self.tree.iter().map(|(k, _)| k).collect()
        }

        fn range_keys(&self, start: &Bound<K3>, end: &Bound<K3>, dir: Dir) -> Vec<K3> {
            self.tree
                .keys_range_bounds_dir(bound_ref(start), bound_ref(end), dir)
                .collect()
        }
    }

    impl LsmTreeHarness {
        pub(super) fn force_next_key_guard_error(&self) {
            self.tree.force_next_key_guard_error_for_test();
        }
    }
}

#[cfg(feature = "fjall")]
use lsm_harness::LsmTreeHarness;

fn seed_scope_keys<H: ScopeHarness>(h: &H) {
    let keys = [
        K3 { a: 1, b: 1, c: 1 },
        K3 { a: 1, b: 1, c: 2 },
        K3 { a: 1, b: 2, c: 1 },
        K3 { a: 1, b: 3, c: 9 },
        K3 { a: 2, b: 1, c: 1 },
        K3 { a: 100, b: 1, c: 0 },
        K3 { a: 100, b: 2, c: 0 },
        K3 { a: 100, b: 3, c: 0 },
        K3 { a: 100, b: 4, c: 0 },
        K3 { a: 100, b: 5, c: 0 },
        K3 { a: 100, b: 6, c: 0 },
        K3 { a: 200, b: 3, c: 0 },
    ];
    for k in keys {
        h.insert(k);
    }
}

fn scope_n_matches_brute_force<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    seed_scope_keys(&h);

    let probe = K3 { a: 1, b: 1, c: 0 };
    for n in 0..=3 {
        let got: BTreeSet<_> = match n {
            0 => h.scope::<0>(&probe).keys().collect(),
            1 => h.scope::<1>(&probe).keys().collect(),
            2 => h.scope::<2>(&probe).keys().collect(),
            3 => h.scope::<3>(&probe).keys().collect(),
            _ => unreachable!(),
        };
        let expected = match n {
            0 => h.brute_prefix::<0>(&probe),
            1 => h.brute_prefix::<1>(&probe),
            2 => h.brute_prefix::<2>(&probe),
            3 => h.brute_prefix::<3>(&probe),
            _ => unreachable!(),
        };
        assert_eq!(got, expected, "N={n}");
    }
}

fn assert_range_oracle<H: ScopeHarness>(
    h: &H,
    scope_key: &K3,
    n: usize,
    m: usize,
    lo_ref: Bound<&K3>,
    hi_ref: Bound<&K3>,
) {
    let (start, end) = manual_saturate_range_bounds(scope_key, n, m, lo_ref, hi_ref);
    for dir in [Dir::Asc, Dir::Desc] {
        let narrowed = match m {
            2 => h.scope::<1>(scope_key).range::<2>(lo_ref, hi_ref),
            3 => h.scope::<1>(scope_key).range::<3>(lo_ref, hi_ref),
            _ => panic!("unsupported M={m}"),
        };
        let scope_keys: BTreeSet<_> = match dir {
            Dir::Asc => narrowed.asc().keys().collect(),
            Dir::Desc => narrowed.desc().keys().collect(),
        };
        let manual_keys: BTreeSet<_> = h.range_keys(&start, &end, dir).into_iter().collect();
        assert_eq!(
            scope_keys, manual_keys,
            "dir={dir:?} lo={lo_ref:?} hi={hi_ref:?}"
        );
    }
}

fn range_m_matches_manual_saturation<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    seed_scope_keys(&h);

    let scope_key = K3 { a: 100, b: 0, c: 0 };
    let cases: &[(Bound<K3>, Bound<K3>)] = &[
        (
            Bound::Included(K3 { a: 0, b: 2, c: 0 }),
            Bound::Included(K3 { a: 0, b: 5, c: 0 }),
        ),
        (Bound::Excluded(K3 { a: 0, b: 2, c: 0 }), Bound::Unbounded),
        (Bound::Unbounded, Bound::Included(K3 { a: 0, b: 4, c: 0 })),
        (Bound::Unbounded, Bound::Excluded(K3 { a: 0, b: 4, c: 0 })),
    ];

    for (lo, hi) in cases {
        assert_range_oracle(&h, &scope_key, 1, 2, bound_ref(lo), bound_ref(hi));
    }

    // M = all fields (3)
    let lo_k = K3 { a: 0, b: 3, c: 1 };
    let lo = Bound::Included(lo_k);
    let hi = Bound::Included(lo_k);
    assert_range_oracle(&h, &scope_key, 1, 3, bound_ref(&lo), bound_ref(&hi));
}

/// `Bound::Unbounded` in `range` means the scope edge — not the whole collection.
fn unbounded_range_stays_in_scope<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    seed_scope_keys(&h);

    let scope_key = K3 { a: 100, b: 0, c: 0 };
    let keys: Vec<K3> = h
        .scope::<1>(&scope_key)
        .range::<2>(Bound::Unbounded, Bound::Unbounded)
        .keys()
        .collect();
    assert!(!keys.is_empty());
    assert!(keys.iter().all(|k| k.a == 100));

    // Descending: the saturated edges reach `bounds_for_page`/`merge_upper` as
    // concrete keys rather than `Unbounded`.
    let mut desc: Vec<K3> = h
        .scope::<1>(&scope_key)
        .desc()
        .range::<2>(Bound::Unbounded, Bound::Unbounded)
        .keys()
        .collect();
    let mut asc_sorted = keys.clone();
    asc_sorted.sort_by_key(|k| (k.a, k.b, k.c));
    desc.sort_by_key(|k| (k.a, k.b, k.c));
    assert_eq!(desc, asc_sorted, "desc must cover the same set");

    // One-sided: a concrete lower bound with an unbounded upper edge stays inside.
    let lo = K3 {
        a: 100,
        b: keys.iter().map(|k| k.b).max().unwrap(),
        c: 0,
    };
    let one_sided: Vec<K3> = h
        .scope::<1>(&scope_key)
        .range::<2>(Bound::Included(&lo), Bound::Unbounded)
        .keys()
        .collect();
    assert!(!one_sided.is_empty());
    assert!(one_sided.iter().all(|k| k.a == 100 && k.b >= lo.b));

    // `scope::<0>()` — the only case where the lower edge is `Included(zeroed)`
    // while the upper stays `Unbounded`.
    let all: Vec<K3> = h
        .scope::<0>(&scope_key)
        .range::<1>(Bound::Unbounded, Bound::Unbounded)
        .keys()
        .collect();
    let expected: BTreeSet<K3> = h.all_keys();
    assert_eq!(all.len(), expected.len());
    assert_eq!(all.iter().copied().collect::<BTreeSet<_>>(), expected);
}

fn page_through_covers_keyset_once<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 10, b: 0, c: 0 };
    for b in 1u64..=10 {
        h.insert(K3 { a: 10, b, c: 0 });
    }

    let scope = h.scope::<1>(&prefix).asc();
    let mut seen = Vec::new();
    let mut after: Option<K3> = None;
    loop {
        let page = scope.page(after.as_ref(), 3);
        if page.items.is_empty() {
            break;
        }
        seen.extend(page.items.iter().map(|(k, _)| *k));
        after = page.next;
        if after.is_none() {
            break;
        }
    }
    // `seen` before dedup: a cursor that re-reads a row would show up here.
    assert_eq!(seen.len(), 10, "no duplicates across pages");
    let mut sorted = seen.clone();
    sorted.sort();
    sorted.dedup();
    assert_eq!(sorted.len(), 10, "every key exactly once");
}

fn page_through_inside_range<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 20, b: 0, c: 0 };
    for b in 1u64..=10 {
        h.insert(K3 { a: 20, b, c: 0 });
    }

    let lo = K3 { a: 0, b: 3, c: 0 };
    let hi = K3 { a: 0, b: 7, c: 0 };
    let scope = h
        .scope::<1>(&prefix)
        .range::<2>(Bound::Included(&lo), Bound::Included(&hi))
        .asc();

    let expected: BTreeSet<_> = (3..=7).map(|b| K3 { a: 20, b, c: 0 }).collect();

    let mut seen = Vec::new();
    let mut after: Option<K3> = None;
    loop {
        let page = scope.page(after.as_ref(), 2);
        if page.items.is_empty() {
            break;
        }
        for (k, _) in &page.items {
            assert!(expected.contains(k), "key {k:?} outside range");
            seen.push(*k);
        }
        after = page.next;
        if after.is_none() {
            break;
        }
    }
    assert_eq!(seen.len(), expected.len());
    assert_eq!(seen.into_iter().collect::<BTreeSet<_>>(), expected);
}

fn chained_range_replaces_bounds<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());

    let prefix = K3 { a: 100, b: 0, c: 0 };
    let keys = [
        K3 { a: 100, b: 1, c: 0 },
        K3 { a: 100, b: 2, c: 0 },
        K3 { a: 100, b: 3, c: 0 },
        K3 { a: 100, b: 4, c: 0 },
        K3 { a: 100, b: 5, c: 0 },
        K3 { a: 100, b: 6, c: 0 },
        K3 { a: 200, b: 3, c: 0 },
    ];
    for k in keys {
        h.insert(k);
    }

    let lo1 = K3 { a: 0, b: 2, c: 0 };
    let hi1 = K3 { a: 0, b: 5, c: 0 };
    let lo2 = K3 { a: 0, b: 3, c: 0 };
    let hi2 = K3 { a: 0, b: 4, c: 0 };

    let mut got: Vec<K3> = h
        .scope::<1>(&prefix)
        .range::<2>(Bound::Included(&lo1), Bound::Included(&hi1))
        .range::<2>(Bound::Included(&lo2), Bound::Included(&hi2))
        .keys()
        .collect();

    let mut expected: Vec<K3> = keys
        .iter()
        .copied()
        .filter(|k| k.a == prefix.a && k.b >= lo2.b && k.b <= hi2.b)
        .collect();

    got.sort_by_key(|k| (k.a, k.b, k.c));
    expected.sort_by_key(|k| (k.a, k.b, k.c));
    assert_eq!(got, expected);
}

fn page_by_tops_up_on_scope<'a, S: OrderedScan<Key = K3> + OwnedRows, const N: usize>(
    scope: Scope<'a, S, N>,
) {
    let mut keys_seen: Vec<K3> = Vec::new();
    let page: MappedPage<K3, u64> = scope
        .page_by(None, 3, |row| {
            let k = S::peek_row_key(&row);
            keys_seen.push(k);
            if k.b % 2 == 0 {
                Ok(None)
            } else {
                Ok(Some(k.b))
            }
        })
        .unwrap();

    assert_eq!(page.items, vec![1, 3, 5]);
    assert_eq!(keys_seen.len(), 5);
    assert_eq!(page.next, Some(K3 { a: 10, b: 5, c: 0 }));
    assert_eq!(page.next, keys_seen.last().copied());
}

fn page_by_tops_up<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 10, b: 0, c: 0 };
    for b in 1u64..=10 {
        h.insert(K3 { a: 10, b, c: 0 });
    }

    page_by_tops_up_on_scope(h.scope::<1>(&prefix).asc());
}

fn page_by_error_on_scope<'a, S: OrderedScan<Key = K3> + OwnedRows, const N: usize>(
    scope: Scope<'a, S, N>,
) {
    let mut accepted = 0usize;
    let err = scope
        .page_by(None, 10, |row| {
            accepted += 1;
            if accepted == 2 {
                return Err(DbError::Internal("abort"));
            }
            Ok(Some(S::peek_row_key(&row).b))
        })
        .unwrap_err();

    assert!(matches!(err, DbError::Internal("abort")));
    assert_eq!(accepted, 2);
}

fn page_by_error<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 20, b: 0, c: 0 };
    for b in 1u64..=5 {
        h.insert(K3 { a: 20, b, c: 0 });
    }

    page_by_error_on_scope(h.scope::<1>(&prefix).asc());
}

fn page_by_chunks_matches_page_by_on_scope<'a, S: OrderedScan<Key = K3>, const N: usize>(
    scope: Scope<'a, S, N>,
) {
    let mapped = scope
        .clone()
        .page_by(None, 3, |row| Ok(Some(S::peek_row_key(&row).b)))
        .unwrap();
    let chunked: ChunkedPage<K3, u64> = scope
        .page_by_chunks(
            None,
            PageSpec {
                limit: 3,
                chunk: 2,
                max_examined: None,
            },
            |_| Ok(()),
            |row, &()| Ok(Some(S::peek_row_key(&row).b)),
        )
        .unwrap();

    assert_eq!(chunked.items, mapped.items);
    assert_eq!(chunked.next, mapped.next);
    assert_eq!(chunked.stop, StopReason::Filled);
}

fn page_by_chunks_matches_page_by<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 30, b: 0, c: 0 };
    for b in 1u64..=5 {
        h.insert(K3 { a: 30, b, c: 0 });
    }

    page_by_chunks_matches_page_by_on_scope(h.scope::<1>(&prefix).asc());
}

fn page_by_chunks_resumes_from_cursor_on_scope<'a, S: OrderedScan<Key = K3>, const N: usize>(
    scope: Scope<'a, S, N>,
    expected: &[u64],
) {
    let mut all = Vec::new();
    let mut cursor: Option<K3> = None;
    let mut pages = 0usize;

    loop {
        let page: ChunkedPage<K3, u64> = scope
            .page_by_chunks(
                cursor.as_ref(),
                PageSpec {
                    limit: 2,
                    chunk: 2,
                    max_examined: None,
                },
                |_| Ok(()),
                |row, &()| {
                    let key = S::peek_row_key(&row);
                    // Reject every third row so a page fills across a chunk edge
                    // and the cursor has to resume mid-chunk.
                    Ok((key.b % 3 != 0).then_some(key.b))
                },
            )
            .unwrap();

        all.extend(page.items);
        pages += 1;
        assert!(pages <= 8, "pagination did not terminate");

        match page.next {
            Some(next) => {
                assert_eq!(page.stop, StopReason::Filled);
                cursor = Some(next);
            }
            None => {
                assert_eq!(page.stop, StopReason::Exhausted);
                break;
            }
        }
    }

    // No row lost at a page boundary, none emitted twice.
    assert_eq!(all, expected);
}

fn page_by_chunks_resumes_from_cursor<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 45, b: 0, c: 0 };
    for b in 1u64..=9 {
        h.insert(K3 { a: 45, b, c: 0 });
    }

    page_by_chunks_resumes_from_cursor_on_scope(h.scope::<1>(&prefix).asc(), &[1, 2, 4, 5, 7, 8]);
    page_by_chunks_resumes_from_cursor_on_scope(h.scope::<1>(&prefix).desc(), &[8, 7, 5, 4, 2, 1]);
}

fn page_by_chunks_tops_up_on_scope<'a, S: OrderedScan<Key = K3>, const N: usize>(
    scope: Scope<'a, S, N>,
) {
    let prepared = Rc::new(RefCell::new(Vec::new()));
    let seen = Rc::clone(&prepared);
    let page = scope
        .page_by_chunks(
            None,
            PageSpec {
                limit: 3,
                chunk: 2,
                max_examined: None,
            },
            move |rows| {
                seen.borrow_mut().push(
                    rows.iter()
                        .map(|row| S::peek_row_key(row).b)
                        .collect::<Vec<_>>(),
                );
                Ok(())
            },
            |row, &()| {
                let key = S::peek_row_key(&row);
                Ok((key.b % 2 == 1).then_some(key.b))
            },
        )
        .unwrap();

    assert_eq!(page.items, vec![1, 3, 5]);
    assert_eq!(page.next, Some(K3 { a: 40, b: 5, c: 0 }));
    assert_eq!(page.stop, StopReason::Filled);
    assert_eq!(&*prepared.borrow(), &[vec![1, 2], vec![3, 4], vec![5, 6]]);
}

fn page_by_chunks_tops_up_in_prepared_chunks<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 40, b: 0, c: 0 };
    for b in 1u64..=6 {
        h.insert(K3 { a: 40, b, c: 0 });
    }

    page_by_chunks_tops_up_on_scope(h.scope::<1>(&prefix).asc());
}

fn page_by_chunks_propagates_errors_on_scope<'a, S: OrderedScan<Key = K3>, const N: usize>(
    scope: Scope<'a, S, N>,
) {
    let prepare_err = scope
        .clone()
        .page_by_chunks(
            None,
            PageSpec {
                limit: 1,
                chunk: 1,
                max_examined: None,
            },
            |_| Err(DbError::Internal("abort")),
            |_, _: &()| Ok(Some(())),
        )
        .unwrap_err();
    assert!(matches!(prepare_err, DbError::Internal("abort")));

    let project_err = scope
        .page_by_chunks(
            None,
            PageSpec {
                limit: 1,
                chunk: 1,
                max_examined: None,
            },
            |_| Ok(()),
            |_, &()| Err::<Option<()>, _>(DbError::Internal("abort")),
        )
        .unwrap_err();
    assert!(matches!(project_err, DbError::Internal("abort")));
}

fn page_by_chunks_propagates_errors<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 50, b: 0, c: 0 };
    h.insert(K3 { a: 50, b: 1, c: 0 });

    page_by_chunks_propagates_errors_on_scope(h.scope::<1>(&prefix).asc());
}

fn cursor_before_and_past_scope_asc<H: ScopeHarness>() {
    if !H::IS_TREE && !H::MULTI_SHARD_MAP {
        return;
    }
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 50, b: 0, c: 0 };
    for b in 1u64..=10 {
        h.insert(K3 { a: 50, b, c: 0 });
    }
    let scope = h.scope::<1>(&prefix).asc();

    let first_page = scope.page(None, 3);
    assert_eq!(
        first_page
            .items
            .iter()
            .map(|(k, _)| k.b)
            .collect::<Vec<_>>(),
        vec![1, 2, 3]
    );

    let before = K3 { a: 49, b: 99, c: 0 };
    let before_page = scope.page(Some(&before), 3);
    assert_eq!(
        before_page
            .items
            .iter()
            .map(|(k, _)| *k)
            .collect::<Vec<_>>(),
        first_page.items.iter().map(|(k, _)| *k).collect::<Vec<_>>()
    );
    assert_eq!(before_page.next, first_page.next);

    let past = K3 { a: 50, b: 99, c: 0 };
    let past_page = scope.page(Some(&past), 3);
    assert!(past_page.items.is_empty());
    assert!(past_page.next.is_none());

    let cursor = K3 { a: 50, b: 5, c: 0 };
    h.delete(&cursor);
    let after_deleted = h.scope::<1>(&prefix).asc().page(Some(&cursor), 3);
    assert_eq!(
        after_deleted
            .items
            .iter()
            .map(|(k, _)| k.b)
            .collect::<Vec<_>>(),
        vec![6, 7, 8]
    );
}

fn cursor_from_foreign_scope<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 50, b: 0, c: 0 };
    for b in 1u64..=5 {
        h.insert(K3 { a: 50, b, c: 0 });
        h.insert(K3 { a: 70, b, c: 0 });
    }
    // Cursor from a different scope: start saturates past the scope end -> start > end.
    let foreign = K3 { a: 70, b: 3, c: 0 };
    let asc = h.scope::<1>(&prefix).asc().page(Some(&foreign), 3);
    assert!(asc.items.is_empty(), "asc: {:?}", asc.items.len());
    assert!(asc.next.is_none());

    let foreign_low = K3 { a: 10, b: 3, c: 0 };
    let desc = h.scope::<1>(&prefix).desc().page(Some(&foreign_low), 3);
    assert!(desc.items.is_empty(), "desc: {:?}", desc.items.len());
    assert!(desc.next.is_none());
}

fn cursor_before_and_past_scope_desc<H: ScopeHarness>() {
    if !H::IS_TREE && !H::MULTI_SHARD_MAP {
        return;
    }
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 60, b: 0, c: 0 };
    for b in 1u64..=10 {
        h.insert(K3 { a: 60, b, c: 0 });
    }
    let scope = h.scope::<1>(&prefix).desc();

    let first_page = scope.page(None, 3);
    assert_eq!(
        first_page
            .items
            .iter()
            .map(|(k, _)| k.b)
            .collect::<Vec<_>>(),
        vec![10, 9, 8]
    );

    let before = K3 { a: 60, b: 99, c: 0 };
    let before_page = scope.page(Some(&before), 3);
    assert_eq!(
        before_page
            .items
            .iter()
            .map(|(k, _)| *k)
            .collect::<Vec<_>>(),
        first_page.items.iter().map(|(k, _)| *k).collect::<Vec<_>>()
    );
    assert_eq!(before_page.next, first_page.next);

    let past = K3 { a: 60, b: 0, c: 0 };
    let past_page = scope.page(Some(&past), 3);
    assert!(past_page.items.is_empty());
    assert!(past_page.next.is_none());
}

fn asc_eq_desc_reversed<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    let prefix = K3 { a: 70, b: 0, c: 0 };
    for b in 1u64..=5 {
        h.insert(K3 { a: 70, b, c: 0 });
    }

    let asc: Vec<K3> = h.scope::<1>(&prefix).asc().keys().collect();
    let desc: Vec<K3> = h.scope::<1>(&prefix).desc().keys().collect();
    assert_eq!(desc, asc.into_iter().rev().collect::<Vec<_>>());
}

fn scope_edges<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());

    // N = 0 — entire collection
    h.insert(K3 { a: 1, b: 1, c: 1 });
    h.insert(K3 { a: 2, b: 2, c: 2 });
    assert_eq!(
        h.scope::<0>(&K3::zeroed()).keys().count(),
        h.all_keys().len()
    );

    // N = all fields — exact key
    let exact = K3 { a: 1, b: 1, c: 1 };
    let got: BTreeSet<_> = h.scope::<3>(&exact).keys().collect();
    assert_eq!(got, BTreeSet::from([exact]));

    // M = N+1 inside populated prefix
    let prefix = K3 { a: 3, b: 0, c: 0 };
    for b in 1u64..=3 {
        h.insert(K3 { a: 3, b, c: 1 });
    }
    let lo = K3 { a: 0, b: 2, c: 0 };
    let hi = K3 { a: 0, b: 2, c: 0 };
    let narrowed: BTreeSet<_> = h
        .scope::<1>(&prefix)
        .range::<2>(Bound::Included(&lo), Bound::Included(&hi))
        .keys()
        .collect();
    assert_eq!(narrowed, BTreeSet::from([K3 { a: 3, b: 2, c: 1 }]));

    // empty scope
    let empty_prefix = K3 { a: 99, b: 0, c: 0 };
    assert!(h.scope::<1>(&empty_prefix).keys().next().is_none());

    // all-0xFF prefix → carry → Unbounded end
    let ff_key = K3 {
        a: u64::MAX,
        b: 0,
        c: 0,
    };
    let (_start, end) = key_math::scope_bounds(&ff_key, 1);
    assert!(matches!(end, Bound::Unbounded));
    h.insert(K3 {
        a: u64::MAX,
        b: 1,
        c: 0,
    });
    assert!(h.scope::<1>(&ff_key).keys().any(|k| k.a == u64::MAX));

    // limit = 0
    let scope = h.scope::<1>(&prefix).asc();
    let page = scope.page(None, 0);
    assert!(page.items.is_empty());
    assert!(page.next.is_none());
}

/// `iter()` yields real rows (each one owned through `OwnedRows`), and `keys()`
/// agrees with it key-for-key in the same order.
fn scope_iter_matches_keys_on_scope<'a, S: OrderedScan<Key = K3> + OwnedRows, const N: usize>(
    rows_scope: Scope<'a, S, N>,
    keys_scope: Scope<'a, S, N>,
    expected: &[K3],
) {
    let from_iter: Vec<K3> = rows_scope.iter().map(|row| S::own(row).0).collect();
    let from_keys: Vec<K3> = keys_scope.keys().collect();

    assert_eq!(from_iter, expected, "iter() rows");
    assert_eq!(from_iter, from_keys, "iter() and keys() must agree");
}

fn scope_iter_returns_rows<H: ScopeHarness>() {
    let dir = tempfile::tempdir().unwrap();
    let h = H::open(dir.path());
    for b in 1u64..=3 {
        h.insert(K3 { a: 7, b, c: 0 });
    }
    // Outside the scope — must not show up in either walk.
    h.insert(K3 { a: 8, b: 1, c: 0 });

    let prefix = K3 { a: 7, b: 0, c: 0 };
    let expected: Vec<K3> = (1u64..=3).map(|b| K3 { a: 7, b, c: 0 }).collect();
    scope_iter_matches_keys_on_scope(
        h.scope::<1>(&prefix).asc(),
        h.scope::<1>(&prefix).asc(),
        &expected,
    );

    // Single-field scope on a full key still returns exactly that row.
    let key = K3 { a: 7, b: 1, c: 0 };
    scope_iter_matches_keys_on_scope(h.scope::<2>(&key), h.scope::<2>(&key), &[key]);
}

macro_rules! scope_semantics_tests {
    ($mod_name:ident, $harness:ty) => {
        mod $mod_name {
            use super::*;

            #[test]
            fn scope_n_matches_brute_force() {
                super::scope_n_matches_brute_force::<$harness>();
            }

            #[test]
            fn range_m_matches_manual_saturation() {
                super::range_m_matches_manual_saturation::<$harness>();
            }

            #[test]
            fn unbounded_range_stays_in_scope() {
                super::unbounded_range_stays_in_scope::<$harness>();
            }

            #[test]
            fn page_through_covers_keyset_once() {
                super::page_through_covers_keyset_once::<$harness>();
            }

            #[test]
            fn page_through_inside_range() {
                super::page_through_inside_range::<$harness>();
            }

            #[test]
            fn chained_range_replaces_bounds() {
                super::chained_range_replaces_bounds::<$harness>();
            }

            #[test]
            fn page_by_tops_up() {
                super::page_by_tops_up::<$harness>();
            }

            #[test]
            fn page_by_error() {
                super::page_by_error::<$harness>();
            }

            #[test]
            fn page_by_chunks_matches_page_by() {
                super::page_by_chunks_matches_page_by::<$harness>();
            }

            #[test]
            fn page_by_chunks_resumes_from_cursor() {
                super::page_by_chunks_resumes_from_cursor::<$harness>();
            }

            #[test]
            fn page_by_chunks_tops_up_in_prepared_chunks() {
                super::page_by_chunks_tops_up_in_prepared_chunks::<$harness>();
            }

            #[test]
            fn page_by_chunks_propagates_errors() {
                super::page_by_chunks_propagates_errors::<$harness>();
            }

            #[test]
            fn cursor_before_and_past_scope_asc() {
                super::cursor_before_and_past_scope_asc::<$harness>();
            }

            #[test]
            fn cursor_before_and_past_scope_desc() {
                super::cursor_before_and_past_scope_desc::<$harness>();
            }

            #[test]
            fn asc_eq_desc_reversed() {
                super::asc_eq_desc_reversed::<$harness>();
            }

            #[test]
            fn scope_edges() {
                super::scope_edges::<$harness>();
            }

            #[test]
            fn cursor_from_foreign_scope() {
                super::cursor_from_foreign_scope::<$harness>();
            }

            #[test]
            fn scope_iter_returns_rows() {
                super::scope_iter_returns_rows::<$harness>();
            }
        }
    };
}

#[test]
fn scope_debug_prints_n_and_hex_keys() {
    let dir = tempfile::tempdir().unwrap();
    let h = ConstTreeHarness::open(dir.path());
    let k = K3 { a: 1, b: 0, c: 0 };
    let s = h.scope::<1>(&k);

    let text = format!("{s:?}");
    assert!(text.starts_with("Scope {"), "{text}");
    // `N` is the most useful field of a scope and is not a struct field, so a
    // derived `Debug` would drop it.
    assert!(text.contains("N: 1"), "{text}");
    // Keys render as their raw bytes in hex — `Key` does not promise `Debug`.
    let prefix = key_math::prefix_key(&k, 1);
    let hex: String = Key::as_bytes(&prefix)
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect();
    assert!(text.contains(&format!("prefix_key: 0x{hex}")), "{text}");
    assert!(text.contains("start: Included(0x"), "{text}");
}

#[test]
fn scope_clone_lets_keys_then_page() {
    let dir = tempfile::tempdir().unwrap();
    let h = ConstTreeHarness::open(dir.path());
    let k = K3 { a: 1, b: 0, c: 0 };
    h.insert(K3 { a: 1, b: 1, c: 1 });
    let s = h.scope::<1>(&k);
    let keys: Vec<_> = s.clone().keys().collect();
    assert!(!keys.is_empty());
    let page = s.page(None, 10);
    assert_eq!(page.items.len(), keys.len().min(10));
}

#[cfg(feature = "var-collections")]
#[test]
fn merge_chunked_composition_uses_complete_body_keys() {
    let dir = tempfile::tempdir().unwrap();
    let channels = VarTypedTree::<ChannelKey, u64, U64Codec>::open(
        dir.path().join("channels"),
        Config::test(),
        U64Codec,
    )
    .unwrap();
    let posts = VarTypedTree::<PostKey, u64, U64Codec>::open(
        dir.path().join("posts"),
        Config::test(),
        U64Codec,
    )
    .unwrap();
    let direct = VarTypedTree::<DirectKey, u64, U64Codec>::open(
        dir.path().join("direct"),
        Config::test(),
        U64Codec,
    )
    .unwrap();
    let bodies = VarTypedTree::<BodyKey, u64, U64Codec>::open(
        dir.path().join("bodies"),
        Config::test(),
        U64Codec,
    )
    .unwrap();

    channels
        .insert(&ChannelKey { channel: 2, msg: 9 }, &9)
        .unwrap();
    channels
        .insert(&ChannelKey { channel: 1, msg: 8 }, &8)
        .unwrap();
    posts.insert(&PostKey { post: 11, msg: 7 }, &7).unwrap();
    posts.insert(&PostKey { post: 10, msg: 6 }, &6).unwrap();
    direct.insert(&DirectKey { user: 42, msg: 6 }, &6).unwrap();
    direct.insert(&DirectKey { user: 42, msg: 5 }, &5).unwrap();

    let body_8 = BodyKey {
        kind: 1,
        owner: 1,
        msg: 8,
    };
    let body_6 = BodyKey {
        kind: 2,
        owner: 10,
        msg: 6,
    };
    let body_5 = BodyKey {
        kind: 3,
        owner: 42,
        msg: 5,
    };
    bodies.insert(&body_8, &80).unwrap();
    bodies.insert(&body_6, &60).unwrap();
    bodies.insert(&body_5, &50).unwrap();

    // These are concrete collection scan iterator types, not Vec-backed streams:
    // `merge_sorted` consumes them lazily and stops once the page fills.
    let merged = merge_sorted(
        Dir::Desc,
        vec![
            CandidateRows::Channel(channels.scope::<0>(&ChannelKey::zeroed()).desc().iter()),
            CandidateRows::Post(posts.scope::<0>(&PostKey::zeroed()).desc().iter()),
            CandidateRows::Direct(direct.scope::<0>(&DirectKey::zeroed()).desc().iter()),
        ],
    )
    .coalesce_equal(|_, _| {});
    let prepared = Rc::new(RefCell::new(Vec::new()));
    let seen = Rc::clone(&prepared);
    let page = page_chunks(
        merged,
        PageSpec {
            limit: 3,
            chunk: 5,
            max_examined: Some(5),
        },
        |row| row.msg,
        move |rows| {
            let keys = rows.iter().map(|row| row.body).collect::<Vec<_>>();
            seen.borrow_mut().extend(keys.iter().copied());
            Ok(keys
                .iter()
                .copied()
                .zip(bodies.get_many(&keys))
                .filter_map(|(key, value)| value.map(|value| (key, value)))
                .collect::<BTreeMap<_, _>>())
        },
        |row, bodies| Ok(bodies.get(&row.body).map(|value| (row.msg, *value))),
    )
    .unwrap();

    assert_eq!(page.items, vec![(8, 80), (6, 60), (5, 50)]);
    assert_eq!(page.next, Some(5));
    assert_eq!(page.stop, StopReason::Filled);
    assert!(prepared.borrow().contains(&body_8));
    assert!(prepared.borrow().contains(&body_6));
    assert!(prepared.borrow().contains(&body_5));
    assert!(prepared.borrow().contains(&BodyKey {
        kind: 1,
        owner: 2,
        msg: 9
    }));
    assert!(prepared.borrow().contains(&BodyKey {
        kind: 2,
        owner: 11,
        msg: 7
    }));
}

scope_semantics_tests!(const_tree, ConstTreeHarness);
scope_semantics_tests!(typed_tree, TypedTreeHarness);
#[cfg(feature = "var-collections")]
scope_semantics_tests!(var_typed_tree, VarTypedTreeHarness);
scope_semantics_tests!(const_map, ConstMapHarness);
scope_semantics_tests!(typed_map, TypedMapHarness);

// Reversed storage runs the *whole* suite, not a subset: `keys_range_bounds_dir`
// branches on `direction × storage order` and that matrix is duplicated per family.
scope_semantics_tests!(const_tree_reversed, ConstTreeReversedHarness);
scope_semantics_tests!(const_map_reversed, ConstMapReversedHarness);

#[cfg(feature = "fjall")]
scope_semantics_tests!(lsm_tree, LsmTreeHarness);