armdb 0.5.3

use std::hash::Hash;
use std::marker::PhantomData;
use std::ops::Bound;

use crate::Key;
use crate::byte_view::ByteView;
use crate::codec::Codec;
use crate::compaction::CompactionIndex;
use crate::config::Config;
use crate::disk_loc::DiskLoc;
use crate::error::{DbError, DbResult};
use crate::hook::{NoHook, TypedWriteHook, VarTypedHookAdapter};
use crate::map_index::{Dir, KeyPage, MapIter, MapIterView, Page, dir_from};
use crate::var_map::{VarMap, VarMapShard};

/// A map with fixed-size keys and typed values `T`. Values are encoded via a
/// [`Codec`] and stored on disk (variable length), with a `BlockCache` for reads.
/// Uses per-shard HashMap for O(1) lookup. Ordered scans are available when
/// opened with `iterable(true)` via [`iter_view`](Self::iter_view). Use
/// [`VarTypedTree`](crate::VarTypedTree) for prefix/range scans without the iterable flag.
///
/// Thin wrapper around [`VarMap<K, VarTypedHookAdapter<K, T, C, H>>`].
///
/// # Error handling
///
/// Same convention as [`VarTypedTree`](crate::VarTypedTree): `get` returns
/// `None` on decode errors. `migrate` keeps entries that fail to decode.
///
/// # Write hooks
///
/// Uses [`TypedWriteHook<K, T>`] via [`VarTypedHookAdapter`]. The hook receives
/// `&T` directly; the adapter decodes raw bytes via the codec. `on_write` fires
/// on `put`/`insert`/`delete`/`cas`/`compare_delete`/`update` and inside `atomic()`.
pub struct VarTypedMap<
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T> = NoHook,
> {
    inner: VarMap<K, VarTypedHookAdapter<K, T, C, H>>,
    codec: C,
    _marker: PhantomData<fn() -> T>,
}

impl<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T> + Clone> VarTypedMap<K, T, C> {
    /// Open or create a `VarTypedMap` at the given path.
    pub fn open(path: impl AsRef<std::path::Path>, config: Config, codec: C) -> DbResult<Self> {
        Self::open_hooked_inner(path, config, codec, NoHook)
    }
}

impl<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>>
    VarTypedMap<K, T, C, H>
{
    /// Open or create a `VarTypedMap` with a typed write hook.
    pub fn open_hooked(
        path: impl AsRef<std::path::Path>,
        config: Config,
        codec: C,
        hook: H,
    ) -> DbResult<Self> {
        Self::open_hooked_inner(path, config, codec, hook)
    }

    fn open_hooked_inner(
        path: impl AsRef<std::path::Path>,
        config: Config,
        codec: C,
        hook: H,
    ) -> DbResult<Self> {
        let adapter = VarTypedHookAdapter {
            inner: hook,
            codec: codec.clone(),
            _marker: PhantomData,
        };
        let inner = VarMap::open_hooked(path, config, adapter)?;
        Ok(Self {
            inner,
            codec,
            _marker: PhantomData,
        })
    }

    /// Graceful shutdown: write hint files (if enabled), flush write buffers + fsync.
    pub fn close(self) -> DbResult<()> {
        self.inner.close()
    }

    /// Flush all shard write buffers to disk (without fsync).
    pub fn flush_buffers(&self) -> DbResult<()> {
        self.inner.flush_buffers()
    }

    /// Get the database configuration.
    pub fn config(&self) -> &Config {
        self.inner.config()
    }

    /// Trigger a compaction pass across all shards.
    pub fn compact(&self) -> DbResult<usize> {
        self.inner.compact()
    }

    /// Write hint files for all active shard files. Call during graceful shutdown.
    pub fn sync_hints(&self) -> DbResult<()> {
        self.inner.sync_hints()
    }

    /// Pre-populate the block cache with blocks containing live values.
    pub fn warmup(&self) -> DbResult<()> {
        self.inner.warmup()
    }

    /// Access the underlying `VarMap`.
    pub fn as_inner(&self) -> &VarMap<K, VarTypedHookAdapter<K, T, C, H>> {
        &self.inner
    }

    /// Access the codec used for encoding / decoding values.
    pub fn codec(&self) -> &C {
        &self.codec
    }

    // -- Reads ----------------------------------------------------------------

    /// Get and decode a value by key. Returns `None` if absent or undecodable.
    /// O(1) average index lookup under a brief per-shard mutex; the `pread` on a cache miss
    /// runs outside the lock, then decode.
    pub fn get(&self, key: &K) -> Option<T> {
        let bytes = self.inner.get(key)?;
        self.codec.decode_from(&bytes).ok()
    }

    /// Get a value by key, returning `Err(KeyNotFound)` if absent or
    /// `Err(CorruptedEntry)` if present but undecodable.
    pub fn get_or_err(&self, key: &K) -> DbResult<T> {
        self.try_get(key)?.ok_or(DbError::KeyNotFound)
    }

    /// Strict read: `Ok(None)` only when absent; `Err` when present but the value
    /// cannot be read or decoded.
    pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
        match self.inner.try_get(key)? {
            Some(bytes) => self.codec.decode_from(&bytes).map(Some),
            None => Ok(None),
        }
    }

    pub fn contains(&self, key: &K) -> bool {
        self.inner.contains(key)
    }

    // -- Read-only scan -------------------------------------------------------

    /// Read-only pass over all live entries, decoding each value. Order is
    /// unspecified — intended for schema validation on a quiet database.
    /// Entries whose value fails to decode are skipped (logged at `debug`),
    /// matching `VarTypedTree::iter` / `VarTypedMap::get`.
    pub fn for_each(&self, mut f: impl FnMut(K, T)) {
        self.inner
            .for_each(|key, bytes| match self.codec.decode_from(&bytes) {
                Ok(value) => f(key, value),
                Err(_) => {
                    tracing::debug!(
                        value_len = bytes.len(),
                        "var_typed_map for_each: decode error, skipping entry"
                    );
                }
            });
    }

    // -- Ordered iteration ----------------------------------------------------

    /// Ordered scan view. `None` unless the collection was opened with
    /// `iterable(true)`. Direction follows `Config::reversed` (default DESC).
    ///
    /// **Occasional admin/maintenance API, not a hot path.** Use it to browse a
    /// collection or bulk-clean stale rows (`retain`); for scan-heavy access reach
    /// for the `*Tree` sibling — a Map's strength is O(1) point lookup, and the
    /// companion key index costs extra memory plus O(log N) per write. Unlike a
    /// `Tree` scan it is **not a point-in-time snapshot**: the k-way merge drops
    /// each shard lock between batches, so concurrent writes may be only partially
    /// reflected (run on a quiet DB if you need a coherent view).
    pub fn iter_view(&self) -> Option<VarTypedMapIterView<'_, K, T, C, H>> {
        self.inner.iter_view()?;
        Some(VarTypedMapIterView {
            map: self,
            inner: MapIterView::new(&self.inner, dir_from(self.inner.config().reversed)),
        })
    }

    // -- Writes ---------------------------------------------------------------

    pub fn put(&self, key: &K, value: &T) -> DbResult<()> {
        let mut buf = Vec::new();
        self.codec.encode_to(value, &mut buf)?;
        self.inner.put(key, &buf)
    }

    pub fn insert(&self, key: &K, value: &T) -> DbResult<()> {
        let mut buf = Vec::new();
        self.codec.encode_to(value, &mut buf)?;
        self.inner.insert(key, &buf)
    }

    pub fn delete(&self, key: &K) -> DbResult<bool> {
        self.inner.delete(key)
    }

    pub fn cas(&self, key: &K, expected: &T, new_value: &T) -> DbResult<()> {
        let mut exp_buf = Vec::new();
        self.codec.encode_to(expected, &mut exp_buf)?;
        let mut new_buf = Vec::new();
        self.codec.encode_to(new_value, &mut new_buf)?;
        self.inner.cas(key, &exp_buf, &new_buf)
    }

    /// Compare-and-delete based on encoded bytes. Relies on deterministic codec output.
    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
    /// `Err(KeyNotFound)` if the key doesn't exist.
    pub fn compare_delete(&self, key: &K, expected: &T) -> DbResult<()> {
        let mut exp_buf = Vec::new();
        self.codec.encode_to(expected, &mut exp_buf)?;
        self.inner.compare_delete(key, &exp_buf)
    }

    pub fn update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        use std::cell::Cell;
        let out: Cell<Option<T>> = Cell::new(None);
        let result = self.inner.try_update_inner(
            key,
            |bytes| {
                let current = self.codec.decode_from(bytes)?;
                let new_val = f(&current);
                let mut buf = Vec::new();
                self.codec.encode_to(&new_val, &mut buf)?;
                out.set(Some(new_val));
                Ok(Some(ByteView::from_vec(buf)))
            },
            false,
        )?;
        if result.is_none() {
            return Ok(None);
        }
        Ok(out.into_inner())
    }

    pub fn fetch_update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        use std::cell::Cell;
        let out: Cell<Option<T>> = Cell::new(None);
        let result = self.inner.try_update_inner(
            key,
            |bytes| {
                let current = self.codec.decode_from(bytes)?;
                let new_val = f(&current);
                let mut buf = Vec::new();
                self.codec.encode_to(&new_val, &mut buf)?;
                out.set(Some(current));
                Ok(Some(ByteView::from_vec(buf)))
            },
            true,
        )?;
        if result.is_none() {
            return Ok(None);
        }
        Ok(out.into_inner())
    }

    // -- Atomic ---------------------------------------------------------------

    pub fn atomic<R>(
        &self,
        shard_key: &K,
        f: impl FnOnce(&mut VarTypedMapShard<'_, K, T, C, H>) -> DbResult<R>,
    ) -> DbResult<R> {
        self.inner.atomic(shard_key, |var_shard| {
            // SAFETY: erase VarMapShard lifetime via `*mut ()`; see VarTypedMapShard doc.
            let inner_ptr: *mut () = (var_shard as *mut VarMapShard<'_, _, _>).cast();
            let mut shard = VarTypedMapShard {
                tree: self,
                inner: inner_ptr,
                _marker: PhantomData,
            };
            f(&mut shard)
        })
    }

    // -- Info -----------------------------------------------------------------

    pub fn len(&self) -> usize {
        self.inner.len()
    }

    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    pub fn shard_for(&self, key: &K) -> usize {
        self.inner.shard_for(key)
    }

    /// Encoded value byte length for `key`, or `None` if absent.
    /// Reads only the in-memory index entry (`DiskLoc::len`); no disk I/O.
    pub fn entry_len(&self, key: &K) -> Option<u32> {
        self.inner.entry_len(key)
    }

    // -- Migration ------------------------------------------------------------

    pub fn migrate(&self, f: impl Fn(&K, &T) -> crate::MigrateAction<T>) -> DbResult<usize> {
        use crate::MigrateAction;
        self.inner
            .migrate(|key, bytes| match self.codec.decode_from(bytes) {
                Ok(current) => match f(key, &current) {
                    MigrateAction::Keep => MigrateAction::Keep,
                    MigrateAction::Update(new) => {
                        let mut buf = Vec::new();
                        match self.codec.encode_to(&new, &mut buf) {
                            Ok(()) => MigrateAction::Update(ByteView::from_vec(buf)),
                            Err(_) => {
                                tracing::warn!(
                                    "var_typed_map migrate: encode error, keeping entry"
                                );
                                MigrateAction::Keep
                            }
                        }
                    }
                    MigrateAction::Delete => MigrateAction::Delete,
                },
                Err(_) => {
                    tracing::warn!("var_typed_map migrate: decode error, keeping entry");
                    MigrateAction::Keep
                }
            })
    }

    /// Replay `on_init` for every live entry. Used by Db when no migration ran.
    pub(crate) fn replay_init(&self) {
        self.inner.replay_init();
    }
}

impl<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>>
    CompactionIndex<K> for VarTypedMap<K, T, C, H>
{
    fn update_if_match(&self, key: &K, old_loc: DiskLoc, new_loc: DiskLoc) -> bool {
        self.inner.update_if_match(key, old_loc, new_loc)
    }

    fn invalidate_blocks(&self, shard_id: u8, file_id: u32, total_bytes: u64) {
        self.inner.invalidate_blocks(shard_id, file_id, total_bytes);
    }

    fn contains_key(&self, key: &K) -> bool {
        self.inner.contains(key)
    }

    fn is_live(&self, shard_id: u8, key: &K, loc: DiskLoc) -> bool {
        self.inner.is_live(shard_id, key, loc)
    }
}

#[cfg(feature = "replication")]
impl<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>>
    crate::replication::ReplicationTarget for VarTypedMap<K, T, C, H>
{
    fn apply_entry(
        &self,
        shard_inner: &mut crate::shard::ShardInner,
        shard_id: u8,
        file_id: u32,
        entry_offset: u64,
        header: &crate::entry::EntryHeader,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<crate::replication::ApplyOutcome> {
        self.inner.apply_entry(
            shard_inner,
            shard_id,
            file_id,
            entry_offset,
            header,
            key,
            value,
        )
    }

    fn try_apply_entry(
        &self,
        shard_inner: &mut crate::shard::ShardInner,
        shard_id: u8,
        file_id: u32,
        entry_offset: u64,
        header: &crate::entry::EntryHeader,
        raw_after_header: &[u8],
    ) -> DbResult<crate::replication::ApplyOutcome> {
        self.inner.try_apply_entry(
            shard_inner,
            shard_id,
            file_id,
            entry_offset,
            header,
            raw_after_header,
        )
    }

    fn key_len(&self) -> usize {
        self.inner.key_len()
    }
}

// ---------------------------------------------------------------------------
// VarTypedMapShard
// ---------------------------------------------------------------------------

/// Handle for atomic multi-key operations on a single shard of a [`VarTypedMap`].
/// Obtained via [`VarTypedMap::atomic`]. The shard + index locks are held for the
/// lifetime of this struct — keep the closure short.
pub struct VarTypedMapShard<
    'tree,
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
> {
    tree: &'tree VarTypedMap<K, T, C, H>,
    // Type-erased raw pointer to a `VarMapShard<'_, K, VarTypedHookAdapter<K, T, C, H>>`.
    // Invariance of the inner lifetime would force it to `'static` if typed; we
    // erase via `*mut ()` and reconstruct on demand. Valid only within the
    // enclosing `atomic()` closure.
    inner: *mut (),
    _marker: PhantomData<&'tree mut ()>,
}

// SAFETY: see VarTypedShard for the same reasoning.
unsafe impl<
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
> Send for VarTypedMapShard<'_, K, T, C, H>
{
}

impl<K: Key + Send + Sync + Hash + Eq, T: Send + Sync, C: Codec<T> + Clone, H: TypedWriteHook<K, T>>
    VarTypedMapShard<'_, K, T, C, H>
{
    fn inner_mut(&mut self) -> &mut VarMapShard<'_, K, VarTypedHookAdapter<K, T, C, H>> {
        // SAFETY: pointer was set from a live `&mut VarMapShard` and is only
        // dereferenced inside the enclosing `atomic()` closure.
        unsafe { &mut *(self.inner as *mut VarMapShard<'_, K, VarTypedHookAdapter<K, T, C, H>>) }
    }

    fn inner_ref(&self) -> &VarMapShard<'_, K, VarTypedHookAdapter<K, T, C, H>> {
        // SAFETY: see `inner_mut`.
        unsafe { &*(self.inner as *const VarMapShard<'_, K, VarTypedHookAdapter<K, T, C, H>>) }
    }

    pub fn put(&mut self, key: &K, value: &T) -> DbResult<()> {
        let mut buf = Vec::new();
        self.tree.codec.encode_to(value, &mut buf)?;
        self.inner_mut().put(key, &buf)
    }

    pub fn insert(&mut self, key: &K, value: &T) -> DbResult<()> {
        let mut buf = Vec::new();
        self.tree.codec.encode_to(value, &mut buf)?;
        self.inner_mut().insert(key, &buf)
    }

    pub fn delete(&mut self, key: &K) -> DbResult<bool> {
        self.inner_mut().delete(key)
    }

    pub fn get(&self, key: &K) -> Option<T> {
        let bytes = self.inner_ref().get(key)?;
        self.tree.codec.decode_from(&bytes).ok()
    }

    pub fn get_or_err(&self, key: &K) -> DbResult<T> {
        let bytes = self.inner_ref().get_or_err(key)?;
        self.tree.codec.decode_from(&bytes)
    }

    pub fn contains(&self, key: &K) -> bool {
        self.inner_ref().contains(key)
    }
}

#[cfg(feature = "armour")]
impl<T, C, H> crate::armour::collection::Collection for VarTypedMap<T::SelfId, T, C, H>
where
    T: crate::CollectionMeta + Send + Sync,
    C: Codec<T> + Clone + 'static,
    H: TypedWriteHook<T::SelfId, T>,
    T::SelfId: crate::Key + Send + Sync + Hash + Eq,
{
    fn name(&self) -> &str {
        T::NAME
    }
    fn len(&self) -> usize {
        self.len()
    }
    fn compact(&self) -> DbResult<usize> {
        self.compact()
    }
    fn flush(&self) -> DbResult<()> {
        self.flush_buffers()?;
        self.sync_hints()?;
        Ok(())
    }
    fn periodic_flush(&self) -> DbResult<()> {
        self.flush_buffers()
    }
}

// ==========================================================================
// MultiTx — cross-collection transaction support (feature `armour`)
//
// VarTypedMap wraps a VarMap whose hook is VarTypedHookAdapter; the multi-tx
// handle wraps the inner VarMapTx and layers the codec for `T <-> bytes`.
// ==========================================================================

/// Multi-shard transaction handle for [`VarTypedMap`] inside `Db::atomicN`.
#[cfg(feature = "armour")]
pub struct VarTypedMapTx<'a, K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    inner: crate::var_map::VarMapTx<'a, K, VarTypedHookAdapter<K, T, C, H>>,
    codec: &'a C,
}

#[cfg(feature = "armour")]
impl<K, T, C, H> VarTypedMapTx<'_, K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
        match self.inner.try_get(key)? {
            Some(bytes) => self.codec.decode_from(&bytes).map(Some),
            None => Ok(None),
        }
    }

    pub fn try_contains(&self, key: &K) -> DbResult<bool> {
        self.inner.try_contains(key)
    }

    pub fn get_or_err(&self, key: &K) -> DbResult<T> {
        self.try_get(key)?.ok_or(DbError::KeyNotFound)
    }

    pub fn put(&mut self, key: &K, value: &T) -> DbResult<()> {
        let mut buf = Vec::new();
        self.codec.encode_to(value, &mut buf)?;
        self.inner.put(key, &buf)
    }

    pub fn insert(&mut self, key: &K, value: &T) -> DbResult<()> {
        let mut buf = Vec::new();
        self.codec.encode_to(value, &mut buf)?;
        self.inner.insert(key, &buf)
    }

    pub fn delete(&mut self, key: &K) -> DbResult<bool> {
        self.inner.delete(key)
    }
}

#[cfg(feature = "armour")]
impl<K, T, C, H> crate::armour::multi_tx::MultiTx for VarTypedMap<K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    type Key = K;
    type Tx<'a>
        = VarTypedMapTx<'a, K, T, C, H>
    where
        Self: 'a;

    fn shard_for_key(&self, key: &K) -> usize {
        self.inner.shard_for(key)
    }

    fn begin_tx(&self) -> VarTypedMapTx<'_, K, T, C, H> {
        VarTypedMapTx {
            inner: crate::armour::multi_tx::MultiTx::begin_tx(&self.inner),
            codec: &self.codec,
        }
    }

    fn lock_shard_into<'a>(&'a self, shard_id: usize, tx: &mut VarTypedMapTx<'a, K, T, C, H>) {
        crate::armour::multi_tx::MultiTx::lock_shard_into(&self.inner, shard_id, &mut tx.inner)
    }

    fn release_locks(
        &self,
        tx: &mut VarTypedMapTx<'_, K, T, C, H>,
    ) -> crate::armour::multi_tx::SyncNeeds {
        crate::armour::multi_tx::MultiTx::release_locks(&self.inner, &mut tx.inner)
    }

    fn run_sync(&self, needs: crate::armour::multi_tx::SyncNeeds) -> DbResult<()> {
        crate::armour::multi_tx::MultiTx::run_sync(&self.inner, needs)
    }

    fn replay_hooks(&self, tx: VarTypedMapTx<'_, K, T, C, H>) {
        crate::armour::multi_tx::MultiTx::replay_hooks(&self.inner, tx.inner)
    }
}

// ---------------------------------------------------------------------------
// VarTypedMapIterView / VarTypedMapIter
// ---------------------------------------------------------------------------

/// View over an iterable [`VarTypedMap`]; exposes ordered iteration yielding
/// freshly decoded `T` values.
pub struct VarTypedMapIterView<
    'a,
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
> {
    map: &'a VarTypedMap<K, T, C, H>,
    inner: MapIterView<'a, VarMap<K, VarTypedHookAdapter<K, T, C, H>>>,
}

impl<'a, K, T, C, H> VarTypedMapIterView<'a, K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    /// Full ordered scan in the collection's configured direction.
    pub fn iter(&self) -> VarTypedMapIter<'a, K, T, C, H> {
        VarTypedMapIter {
            map: self.map,
            inner: self.inner.iter(),
        }
    }

    /// Iterate entries in `[start, end)` — start inclusive, end exclusive.
    pub fn range(&self, start: &K, end: &K) -> VarTypedMapIter<'a, K, T, C, H> {
        VarTypedMapIter {
            map: self.map,
            inner: self.inner.range(start, end),
        }
    }

    /// Iterate entries in range defined by `start` and `end` bounds.
    pub fn range_bounds(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
    ) -> VarTypedMapIter<'a, K, T, C, H> {
        VarTypedMapIter {
            map: self.map,
            inner: self.inner.range_bounds(start, end),
        }
    }

    /// Iterate entries whose keys start with `prefix`.
    pub fn prefix_iter(&self, prefix: &[u8]) -> VarTypedMapIter<'a, K, T, C, H> {
        VarTypedMapIter {
            map: self.map,
            inner: self.inner.prefix_iter(prefix),
        }
    }

    /// Keyset pagination in the collection's configured direction.
    ///
    /// No `T: Clone` bound: `iter()` already yields freshly decoded owned `T`, so
    /// the page is collected without cloning.
    pub fn paginate(&self, after: Option<&K>, limit: usize) -> Page<K, T> {
        let iter = match (after, dir_from(self.map.inner.config().reversed)) {
            (None, _) => self.iter(),
            (Some(k), Dir::Asc) => self.range_bounds(Bound::Excluded(k), Bound::Unbounded),
            (Some(k), Dir::Desc) => self.range_bounds(Bound::Unbounded, Bound::Excluded(k)),
        };
        let items: Vec<_> = iter.take(limit).collect();
        let next = if items.len() == limit {
            items.last().map(|(k, _)| *k)
        } else {
            None
        };
        Page { items, next }
    }

    // -- Keys-only iteration -------------------------------------------------
    //
    // Keys come straight from the companion `BTreeSet` of the underlying
    // `VarMap`, so these perform **no off-disk read and no codec decode**.

    /// Full ordered **keys-only** scan — no off-disk read, no decode.
    pub fn keys(&self) -> impl Iterator<Item = K> + '_ {
        self.inner.keys()
    }

    /// Keys-only iteration over `[start, end)`.
    pub fn keys_range(&self, start: &K, end: &K) -> impl Iterator<Item = K> + '_ {
        self.inner.keys_range(start, end)
    }

    /// Keys-only iteration over a bounded range.
    pub fn keys_range_bounds(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
    ) -> impl Iterator<Item = K> + '_ {
        self.inner.keys_range_bounds(start, end)
    }

    /// Keys-only iteration over keys starting with `prefix`.
    pub fn keys_prefix(&self, prefix: &[u8]) -> impl Iterator<Item = K> + '_ {
        self.inner.keys_prefix(prefix)
    }

    /// Keys-only pagination — cheap, no off-disk read, no decode.
    pub fn keys_paginate(&self, after: Option<&K>, limit: usize) -> KeyPage<K> {
        self.inner.keys_paginate(after, limit)
    }

    /// Keep entries where `f` returns `true`; delete the rest via loc-revalidating
    /// compare-delete.
    ///
    /// The predicate receives `&T` after codec decode. Entries that fail to decode
    /// are skipped and not deleted (R4). If the `DiskLoc` changed between iteration
    /// and delete (`CasMismatch`) or the key was removed concurrently (`KeyNotFound`),
    /// the delete is skipped. Not atomic across the collection; a genuine storage
    /// error aborts the scan and is returned.
    pub fn retain(&self, mut f: impl FnMut(&K, &T) -> bool) -> DbResult<()> {
        for (key, entry) in self.inner.iter() {
            let keep = match self.map.codec.decode_from(&entry.value) {
                Ok(val) => f(&key, &val),
                Err(_) => {
                    tracing::warn!(
                        value_len = entry.value.len(),
                        "var_typed_map retain: decode error, skipping entry"
                    );
                    continue;
                }
            };
            if keep {
                continue;
            }
            match self.map.inner.compare_delete_if_loc(&key, entry.loc) {
                Ok(()) => {}
                Err(DbError::CasMismatch) | Err(DbError::KeyNotFound) => {
                    tracing::debug!("retain: compare_delete skipped (concurrent change)");
                }
                Err(e) => return Err(e),
            }
        }
        Ok(())
    }
}

/// Iterator over entries in an iterable [`VarTypedMap`], yielding `(K, T)`.
///
/// Weakly-consistent: concurrent inserts/updates may be visible during iteration.
/// Entries with decode errors are skipped (see `tracing::warn!`).
pub struct VarTypedMapIter<
    'a,
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
> {
    map: &'a VarTypedMap<K, T, C, H>,
    inner: MapIter<'a, VarMap<K, VarTypedHookAdapter<K, T, C, H>>>,
}

impl<'a, K, T, C, H> Iterator for VarTypedMapIter<'a, K, T, C, H>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Send + Sync,
    C: Codec<T> + Clone,
    H: TypedWriteHook<K, T>,
{
    type Item = (K, T);

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            let (k, entry) = self.inner.next()?;
            match self.map.codec.decode_from(&entry.value) {
                Ok(v) => return Some((k, v)),
                Err(_) => {
                    tracing::warn!(
                        value_len = entry.value.len(),
                        "var_typed_map iter: decode error, skipping entry"
                    );
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::codec::{Codec, RapiraCodec};
    use crate::config::Config;
    use crate::hook::TypedWriteHook;
    use rapira::Rapira;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
    use tempfile::tempdir;

    #[derive(Debug, Clone, PartialEq, Rapira)]
    struct Profile {
        name: String,
        email: String,
        scores: Vec<i64>,
    }

    type ProfileMap = VarTypedMap<[u8; 16], Profile, RapiraCodec>;

    fn open(dir: &std::path::Path) -> ProfileMap {
        VarTypedMap::open(dir, Config::test(), RapiraCodec).unwrap()
    }

    fn profile(id: u64) -> Profile {
        Profile {
            name: format!("user_{id}"),
            email: format!("user_{id}@ex.com"),
            scores: vec![id as i64, (id * 2) as i64, (id * 3) as i64],
        }
    }

    #[test]
    fn for_each_visits_all_entries() {
        let dir = tempdir().unwrap();
        let map = open(dir.path());

        let p1 = profile(1);
        let p2 = profile(2);
        let p3 = profile(3);
        map.put(&[1u8; 16], &p1).unwrap();
        map.put(&[2u8; 16], &p2).unwrap();
        map.put(&[3u8; 16], &p3).unwrap();

        let mut seen: std::collections::HashMap<[u8; 16], Profile> =
            std::collections::HashMap::new();
        map.for_each(|key, value| {
            seen.insert(key, value);
        });

        assert_eq!(seen.len(), 3);
        assert_eq!(seen[&[1u8; 16]], p1);
        assert_eq!(seen[&[2u8; 16]], p2);
        assert_eq!(seen[&[3u8; 16]], p3);
    }

    #[test]
    fn compare_delete_match_mismatch_absent() {
        let dir = tempdir().unwrap();
        let map = open(dir.path());

        let k = [1u8; 16];
        let v = profile(1);
        let other = profile(99);
        map.put(&k, &v).unwrap();

        assert!(matches!(
            map.compare_delete(&k, &other),
            Err(DbError::CasMismatch)
        ));
        assert_eq!(map.get(&k), Some(v.clone()));

        assert!(map.compare_delete(&k, &v).is_ok());
        assert!(map.get(&k).is_none());

        assert!(matches!(
            map.compare_delete(&k, &v),
            Err(DbError::KeyNotFound)
        ));
    }

    #[test]
    fn entry_len_returns_some_for_existing_key() {
        let tmp = tempdir().expect("tmp");
        let map: VarTypedMap<[u8; 8], Vec<u8>, RapiraCodec> =
            VarTypedMap::open(tmp.path(), Config::test(), RapiraCodec).expect("open");
        let key = 1u64.to_be_bytes();
        map.put(&key, &vec![10u8, 20, 30, 40]).expect("put");
        // RapiraCodec encodes Vec<u8> as 4-byte length prefix + data bytes.
        // vec![10, 20, 30, 40] → 4 + 4 = 8 encoded bytes.
        assert_eq!(map.entry_len(&key), Some(8u32));
    }

    #[test]
    fn entry_len_returns_none_for_missing_key() {
        let tmp = tempdir().expect("tmp");
        let map: VarTypedMap<[u8; 8], Vec<u8>, RapiraCodec> =
            VarTypedMap::open(tmp.path(), Config::test(), RapiraCodec).expect("open");
        let key = 99u64.to_be_bytes();
        assert_eq!(map.entry_len(&key), None);
    }

    #[test]
    fn typed_map_decode_fault_surfaced_on_result_methods() {
        use crate::test_faults::{POISON, PoisonCodec};
        let dir = tempdir().unwrap();
        let map: VarTypedMap<[u8; 8], u64, PoisonCodec> =
            VarTypedMap::open(dir.path(), Config::test(), PoisonCodec).unwrap();

        let k = 1u64.to_be_bytes();
        map.put(&k, &POISON).unwrap();

        assert!(matches!(
            map.get_or_err(&k),
            Err(DbError::CorruptedEntry { .. })
        ));
        assert!(matches!(
            map.try_get(&k),
            Err(DbError::CorruptedEntry { .. })
        ));
        assert!(matches!(
            map.update(&k, |v| v + 1),
            Err(DbError::CorruptedEntry { .. })
        ));
        assert!(matches!(
            map.fetch_update(&k, |v| v + 1),
            Err(DbError::CorruptedEntry { .. })
        ));

        // raw-byte cas
        assert!(matches!(
            map.cas(&k, &123u64, &5u64),
            Err(DbError::CasMismatch)
        ));
        assert!(map.cas(&k, &POISON, &5u64).is_ok());

        let absent = 999u64.to_be_bytes();
        assert!(matches!(map.try_get(&absent), Ok(None)));
        assert!(matches!(map.get_or_err(&absent), Err(DbError::KeyNotFound)));
    }

    #[derive(Clone)]
    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(|_| DbError::CorruptedEntry { offset: 0 })?;
            Ok(u64::from_be_bytes(arr))
        }
    }

    #[derive(Default)]
    struct TRecHookState {
        writes: AtomicUsize,
        last_new: crate::sync::Mutex<Option<u64>>,
    }

    #[derive(Clone, Default)]
    struct TRecHook {
        state: Arc<TRecHookState>,
    }

    impl TypedWriteHook<[u8; 8], u64> for TRecHook {
        fn on_write(&self, _key: &[u8; 8], _old: Option<&u64>, new: Option<&u64>) {
            self.state.writes.fetch_add(1, AtomicOrdering::Relaxed);
            *crate::sync::lock(&self.state.last_new) = new.copied();
        }
    }

    fn open_var_typed_hooked(
        dir: &std::path::Path,
        hook: TRecHook,
    ) -> VarTypedMap<[u8; 8], u64, U64Codec, TRecHook> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        VarTypedMap::open_hooked(dir, cfg, U64Codec, hook).expect("open hooked")
    }

    #[test]
    fn var_typed_map_iter_decodes_desc() {
        #[derive(rapira::Rapira, PartialEq, Debug)]
        struct Doc {
            id: u32,
            body: String,
        }
        let dir = tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .build();
        let m =
            VarTypedMap::<[u8; 2], Doc, RapiraCodec>::open(dir.path(), cfg, RapiraCodec).unwrap();
        m.put(
            &[1, 0],
            &Doc {
                id: 1,
                body: "x".into(),
            },
        )
        .unwrap();
        m.put(
            &[2, 0],
            &Doc {
                id: 2,
                body: "y".into(),
            },
        )
        .unwrap();
        let v = m.iter_view().unwrap();
        let ids: Vec<u32> = v.iter().map(|(_, d)| d.id).collect();
        assert_eq!(ids, [2, 1]);
        // keys-only: no off-disk read, no codec decode.
        let keys: Vec<[u8; 2]> = v.keys().collect();
        assert_eq!(keys, [[2, 0], [1, 0]]);
        v.retain(|_k, doc| doc.id % 2 == 0).unwrap();
        assert_eq!(m.iter_view().unwrap().iter().count(), 1);
    }

    #[test]
    fn var_typed_map_atomic_fires_typed_hook() {
        let dir = tempdir().unwrap();
        let hook = TRecHook::default();
        let state = hook.state.clone();
        let map = open_var_typed_hooked(dir.path(), hook);
        let k = 7u64.to_be_bytes();
        map.atomic(&k, |s| {
            s.put(&k, &42)?;
            Ok(())
        })
        .expect("atomic");
        assert_eq!(state.writes.load(AtomicOrdering::Relaxed), 1);
        assert_eq!(*crate::sync::lock(&state.last_new), Some(42));
    }
}