armdb 0.14.2

use std::hash::Hash;
use std::marker::PhantomData;
use std::ops::{Bound, ControlFlow};
use std::sync::Arc;

use crate::Key;

use crate::compaction::CompactionIndex;
use crate::config::Config;
use crate::const_map::{ConstMap, ConstMapShard};
use crate::durability::{Bitcask, Durability, Fixed};
use crate::error::{DbError, DbResult};
use crate::fixed::config::FixedConfig;
use crate::hook::{NoHook, TypedWriteHook, WriteHook, ZeroHookAdapter};
use crate::map_index::{Dir, KeyPage, MapIterView, Page, ShardSource, TryMapIter, dir_from};
use crate::projection::IndexKeyed;
use crate::upsert::Upsert;
use crate::watch::{self, WatchOptions, WatchReceiver, WatchSpec, ZeroWatchDecoder};
use crate::zero_tree::{from_value_bytes, to_bytes};

/// A map with fixed-size keys and zerocopy-compatible typed values.
///
/// Thin wrapper around [`ConstMap<K, V, D>`] that provides a typed API for values
/// implementing [`FromBytes`] + [`IntoBytes`] + [`Immutable`]. Conversions are
/// zero-cost — values are transmuted without serialization.
///
/// Generic over `D: Durability` (default: `Bitcask`). Use `Fixed` backend for
/// frequent updates without compaction: `ZeroMap::<K, V, Fixed, T>::open(...)`.
///
/// Requires `size_of::<T>() == V` (compile-time assertion in constructor).
///
/// # Write hooks
///
/// Uses [`TypedWriteHook<K, T>`] — the hook receives `&T` directly (not raw bytes).
/// `on_write` fires on `put`/`insert`/`delete`/`cas`/`compare_delete`/`update`/`fetch_update`,
/// and on each mutation inside `atomic()` (replayed after the lock is released, in application order).
/// Old value is always provided (it lives in
/// memory) — `NEEDS_OLD_VALUE` is ignored.
///
/// `on_init` fires once per live entry during `migrate()` or `replay_init()`
/// (enable via `NEEDS_INIT = true`).
///
/// # When to use
///
/// For `T` with trivial representation (no heap allocations, fixed layout):
/// structs with `#[derive(FromBytes, IntoBytes, Immutable, KnownLayout)]`.
/// All values live in memory — reads never touch disk. O(1) HashMap lookup.
///
/// Ordered iteration is available when opened with `Config::iterable(true)` via
/// [`iter_view`](Self::iter_view); otherwise use [`ZeroTree`](crate::ZeroTree) for prefix/range scans.
///
/// # Usage
///
/// ```ignore
/// use zerocopy::{FromBytes, IntoBytes, Immutable, KnownLayout};
///
/// #[derive(FromBytes, IntoBytes, Immutable, KnownLayout, Clone, Copy)]
/// #[repr(C)]
/// struct Session {
///     user_id: u64,
///     expires: u64,
/// }
///
/// let map = ZeroMap::<[u8; 16], { size_of::<Session>() }, Session>::open(
///     "data/sessions",
///     Config::balanced().build(),
/// )?;
/// map.put(&key, &Session { user_id: 42, expires: 0 })?;
/// if let Some(s) = map.get(&key) {
///     println!("user: {}", s.user_id);
/// }
/// ```
///
/// Facade mutations and `atomic()` forward to the inner Const map, so they ride
/// Const's `Family` / `Ctx` seam without a separate Zero `Family` impl.
/// [`ZeroMapShard`] stays `#[repr(transparent)]` over [`ConstMapShard`].
pub struct ZeroMap<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
    H: TypedWriteHook<K, T> = NoHook,
    D: Durability = Bitcask,
> {
    inner: Arc<ConstMap<K, V, ZeroHookAdapter<K, T, H>, D>>,
    _marker: PhantomData<T>,
}

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> watch::sealed::Spec
    for ZeroMap<K, V, T, H, D>
where
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
    D: Durability,
{
}
impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> WatchSpec for ZeroMap<K, V, T, H, D>
where
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
    D: Durability,
{
    type Key = K;
    type Value = T;
    type Decoder = ZeroWatchDecoder<T>;
}
impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T, H, D> ZeroMap<K, V, T, H, D>
where
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T> + watch::sealed::Enabled,
    D: Durability,
{
    /// Subscribe to this collection's mutations.
    ///
    /// Infallible and repeatable; every receiver is independent and owns a
    /// bounded ring. Delivery is **at-most-once** — a full ring drops the
    /// oldest event and reports [`WatchError::Lagged`] at the gap.
    ///
    /// The consumer recipe is `watch()` **first**, then scan, and rescan on
    /// `Lagged`. It is complete because of what this call guarantees on return:
    /// every write that decided not to publish to this receiver has already
    /// made its row visible. The scan half is why a watched Map must be opened
    /// with [`Config::iterable`] — without an iterator view the open itself is
    /// refused with [`DbError::WatchRequiresIterable`], so that this call can
    /// stay infallible.
    ///
    /// An event is a pointer, not a snapshot — a rebuild re-reads rows by key
    /// instead of applying payloads. There is no ordering across keys or
    /// collections, and transaction boundaries are invisible. A `new_only` hook
    /// reports every non-delete as [`WatchKind::Upsert`].
    ///
    /// Subscriptions end with the database: `Db::shutdown` (and `Drop`) closes
    /// every channel; `Db::close` leaves receivers live. See `docs/watch.md`.
    pub fn watch(&self, options: WatchOptions) -> WatchReceiver<Self> {
        WatchReceiver::subscribe(
            self.inner.watch_channel().expect("enabled watch hook"),
            ZeroWatchDecoder::new(),
            options,
        )
    }
}

// ==========================================================================
// Bitcask-specific impl blocks
// ==========================================================================

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
> ZeroMap<K, V, T, NoHook, Bitcask>
{
    /// Open or create a `ZeroMap` at the given path.
    /// Recovers the index from existing data files on disk.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open(path: impl AsRef<std::path::Path>, config: Config) -> DbResult<Self> {
        const { assert!(size_of::<T>() == V) }
        let adapter = ZeroHookAdapter {
            inner: NoHook,
            _marker: PhantomData,
        };
        Ok(Self {
            inner: Arc::new(ConstMap::open_inner(path, config, adapter)?),
            _marker: PhantomData,
        })
    }
}

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
> ZeroMap<K, V, T, H, Bitcask>
{
    /// Open or create a `ZeroMap` with a write hook for secondary index maintenance.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open_hooked(
        path: impl AsRef<std::path::Path>,
        config: Config,
        hook: H,
    ) -> DbResult<Self> {
        crate::counter_dirty::check_no_counters(&hook)?;
        crate::unique::check_no_claims(&hook)?;
        #[cfg(feature = "armour")]
        if H::EMITS {
            return Err(DbError::ChangeStreamRequiresManagedOpen);
        }
        Self::open_hooked_inner(path, config, hook)
    }

    /// Skips [`check_no_counters`](crate::counter_dirty::check_no_counters) —
    /// only `Db::open_managed` may call this, immediately followed by
    /// `ManagedOpen::bind_counter_dirty` before the collection is published.
    pub(crate) fn open_hooked_inner(
        path: impl AsRef<std::path::Path>,
        config: Config,
        hook: H,
    ) -> DbResult<Self> {
        const { assert!(size_of::<T>() == V) }
        if H::WATCHES && !config.iterable {
            return Err(DbError::WatchRequiresIterable);
        }
        let adapter = ZeroHookAdapter {
            inner: hook,
            _marker: PhantomData,
        };
        Ok(Self {
            inner: Arc::new(ConstMap::open_inner(path, config, adapter)?),
            _marker: PhantomData,
        })
    }

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

    pub fn close(self) -> DbResult<()> {
        self.clean_shutdown()
    }

    /// 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()
    }

    /// Access the underlying `ConstMap`.
    pub fn as_inner(&self) -> &ConstMap<K, V, ZeroHookAdapter<K, T, H>, Bitcask> {
        &self.inner
    }
}

// ==========================================================================
// Fixed-specific impl blocks
// ==========================================================================

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
> ZeroMap<K, V, T, NoHook, Fixed>
{
    /// Open or create a `ZeroMap` with Fixed (fixed-slot) backend.
    /// Recovers the index from existing data files on disk.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open(path: impl AsRef<std::path::Path>, config: FixedConfig) -> DbResult<Self> {
        const { assert!(size_of::<T>() == V) }
        let adapter = ZeroHookAdapter {
            inner: NoHook,
            _marker: PhantomData,
        };
        Ok(Self {
            inner: Arc::new(ConstMap::open_fixed_inner(path, config, adapter)?),
            _marker: PhantomData,
        })
    }
}

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
> ZeroMap<K, V, T, H, Fixed>
{
    /// Open or create a `ZeroMap` with a write hook, using Fixed (fixed-slot) backend.
    ///
    #[doc = include_str!("../docs/exclusive-access.md")]
    pub fn open_with_hook(
        path: impl AsRef<std::path::Path>,
        config: FixedConfig,
        hook: H,
    ) -> DbResult<Self> {
        crate::counter_dirty::check_no_counters(&hook)?;
        crate::unique::check_no_claims(&hook)?;
        #[cfg(feature = "armour")]
        if H::EMITS {
            return Err(DbError::ChangeStreamRequiresManagedOpen);
        }
        Self::open_with_hook_inner(path, config, hook)
    }

    /// Skips [`check_no_counters`](crate::counter_dirty::check_no_counters) —
    /// only `Db::open_managed` may call this, immediately followed by
    /// `ManagedOpen::bind_counter_dirty` before the collection is published.
    pub(crate) fn open_with_hook_inner(
        path: impl AsRef<std::path::Path>,
        config: FixedConfig,
        hook: H,
    ) -> DbResult<Self> {
        const { assert!(size_of::<T>() == V) }
        if H::WATCHES && !config.iterable {
            return Err(DbError::WatchRequiresIterable);
        }
        let adapter = ZeroHookAdapter {
            inner: hook,
            _marker: PhantomData,
        };
        Ok(Self {
            inner: Arc::new(ConstMap::open_fixed_inner(path, config, adapter)?),
            _marker: PhantomData,
        })
    }

    /// Perform a clean shutdown (Fixed backend).
    pub fn clean_shutdown(&self) -> DbResult<()> {
        self.inner.clean_shutdown()
    }

    pub fn close(self) -> DbResult<()> {
        self.clean_shutdown()
    }
}

// ==========================================================================
// Generic impl block — works with any D: Durability
// ==========================================================================

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
    D: Durability,
> ZeroMap<K, V, T, H, D>
{
    pub(crate) fn tx_inner(&self) -> &ConstMap<K, V, ZeroHookAdapter<K, T, H>, D> {
        self.inner.as_ref()
    }

    /// Register the `Arc` handle the inner [`ConstMap`] needs for transaction deferred replay.
    pub(crate) fn bind_tx_self(arc: &Arc<Self>) {
        ConstMap::bind_tx_self(&arc.inner);
    }

    pub(crate) fn tx_install_projection_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        self.tx_inner().tx_install_projection_metadata(db)
    }

    pub(crate) fn tx_preplan_projection_targets(
        &self,
    ) -> DbResult<Vec<Arc<dyn crate::txn::participant::TxParticipant>>> {
        self.tx_inner().tx_preplan_projection_targets()
    }

    pub(crate) fn tx_project_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        self.tx_inner().tx_project_writes(key, old, new)
    }

    pub(crate) fn tx_counter_deltas_for(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::TxCounterDelta>> {
        self.tx_inner().tx_counter_deltas_for(key, old, new)
    }

    /// Claim ops for one decided write inside a transaction (T10 Task 6) —
    /// `name` is passed straight through to [`ConstMap::tx_claims_for`],
    /// which cannot name `T` itself.
    pub(crate) fn tx_claims_for(
        &self,
        name: &'static str,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<crate::txn::participant::ClaimOp>> {
        self.tx_inner().tx_claims_for(name, key, old, new)
    }

    /// Build the per-shard TTL indexes by walking live rows.
    ///
    /// Called once by `Db::open_managed`, after migrations and before the handle
    /// is published. Not public API — visible only because the integration tests
    /// build collections without going through `Db`. To rebuild a live
    /// collection's index (e.g. after replication promotion) call
    /// [`rebuild_ttl_index`](Self::rebuild_ttl_index) instead: this one skips the
    /// `H::EXPIRES` check and does not clear the promotion requirement.
    #[doc(hidden)]
    pub fn build_ttl_index(&self) -> DbResult<()> {
        self.inner.build_ttl_index()
    }

    #[doc(hidden)]
    pub fn ttl_debug_len(&self) -> usize {
        self.inner.ttl_debug_len()
    }

    #[doc(hidden)]
    pub fn ttl_debug_deadline(&self, key: &K) -> Option<u64> {
        self.inner.ttl_debug_deadline(key)
    }

    pub fn sweep_expired(&self, now_ms: u64, budget: usize) -> DbResult<crate::ttl::SweepOutcome> {
        self.inner.sweep_expired(now_ms, budget)
    }

    pub fn set_ttl_sweep(&self, enabled: bool) -> DbResult<()> {
        self.inner.set_ttl_sweep(enabled)
    }

    pub fn rebuild_ttl_index(&self) -> DbResult<()> {
        self.inner.rebuild_ttl_index()
    }

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

    /// Get a value by key. O(1) lookup, zero disk I/O. Returns a copy of `T`.
    /// Takes a brief per-shard mutex to read the index (value is inline).
    pub fn get(&self, key: &K) -> Option<T> {
        let bytes = self.inner.get(key)?;
        Some(from_value_bytes::<V, T>(&bytes))
    }

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

    /// Check if a key exists.
    pub fn contains(&self, key: &K) -> bool {
        self.inner.contains(key)
    }

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

    /// Insert or update a key-value pair. Returns the old value if the key existed.
    pub fn put(&self, key: &K, value: &T) -> DbResult<Option<T>> {
        let bytes = to_bytes::<V, T>(value);
        self.inner
            .put(key, &bytes)
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
    }

    /// Insert a key-value pair only if the key does not exist.
    /// Returns `Err(KeyExists)` if the key is already present.
    pub fn insert(&self, key: &K, value: &T) -> DbResult<()> {
        let bytes = to_bytes::<V, T>(value);
        self.inner.insert(key, &bytes)
    }

    /// Delete a key. Returns the old value if the key existed.
    pub fn delete(&self, key: &K) -> DbResult<Option<T>> {
        self.inner
            .delete(key)
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
    }

    /// Compare-and-swap: if current value == expected, replace with new_value.
    /// Returns `Ok(())` on success, `Err(CasMismatch)` if current != expected,
    /// `Err(KeyNotFound)` if key doesn't exist.
    pub fn cas(&self, key: &K, expected: &T, new_value: &T) -> DbResult<()> {
        let exp_bytes = to_bytes::<V, T>(expected);
        let new_bytes = to_bytes::<V, T>(new_value);
        self.inner.cas(key, &exp_bytes, &new_bytes)
    }

    /// Compare-and-delete: if the current value == `expected`, delete the key.
    /// 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 exp_bytes = to_bytes::<V, T>(expected);
        self.inner.compare_delete(key, &exp_bytes)
    }

    /// Atomically read-modify-write. Returns `Some(T)` (the **new** value)
    /// if key existed, `None` otherwise.
    /// The closure must not be heavy (shard lock is held).
    pub fn update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        self.inner
            .update(key, |bytes| {
                let val = from_value_bytes::<V, T>(bytes);
                let new_val = f(&val);
                to_bytes::<V, T>(&new_val)
            })
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
    }

    /// Like [`update()`](Self::update), but returns `Some(T)` with the **old** value.
    pub fn fetch_update(&self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        self.inner
            .fetch_update(key, |bytes| {
                let val = from_value_bytes::<V, T>(bytes);
                let new_val = f(&val);
                to_bytes::<V, T>(&new_val)
            })
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
    }

    /// Decide-and-write: insert, update, keep, or delete based on the closure's
    /// [`Upsert`](crate::Upsert) decision. The closure must be short — the shard
    /// lock is held across it, so it must not panic and must not call back into
    /// this collection (the shard mutex is not reentrant).
    pub fn upsert<U: crate::IntoUpsert<T>>(
        &self,
        key: &K,
        f: impl FnOnce(Option<&T>) -> U,
    ) -> crate::DbResult<U::Out> {
        self.atomic(key, |s| crate::upsert::upsert_via_held(s, key, f))
    }

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

    /// Atomically execute multiple operations on a single shard.
    /// All keys must route to the same shard as `shard_key`.
    /// The closure must be short — shard lock is held for its duration.
    pub fn atomic<R>(
        &self,
        shard_key: &K,
        f: impl FnOnce(&mut ZeroMapShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>) -> DbResult<R>,
    ) -> DbResult<R> {
        self.inner.atomic(shard_key, |const_shard| {
            // SAFETY: ZeroMapShard is #[repr(transparent)] over ConstMapShard with the SAME
            // hook type (ZeroHookAdapter<K, T, H>). PhantomData<T> is ZST.
            let shard = unsafe {
                &mut *(const_shard as *mut ConstMapShard<'_, K, V, ZeroHookAdapter<K, T, H>, D>
                    as *mut ZeroMapShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>)
            };
            f(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)
    }

    /// Flush all shard data to disk.
    pub fn flush(&self) -> DbResult<()> {
        self.inner.flush()
    }

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

    /// Iterate all entries and optionally mutate them. Call once at startup.
    ///
    /// The callback receives each (key, &T) and returns `MigrateAction`:
    /// - `Keep` — no change (fires `on_init` if `NEEDS_INIT`)
    /// - `Update(new_value)` — replace value (hook-free write, fires `on_init`)
    /// - `Delete` — remove entry (hook-free tombstone, no `on_init`)
    ///
    /// `on_write` is **never** fired during migration.
    /// Returns the number of mutated entries.
    pub fn migrate(&self, f: impl Fn(&K, &T) -> crate::MigrateAction<T>) -> DbResult<usize> {
        self.migrate_inner(f, true)
    }

    pub(crate) fn migrate_inner(
        &self,
        f: impl Fn(&K, &T) -> crate::MigrateAction<T>,
        fire_init: bool,
    ) -> DbResult<usize> {
        self.inner.migrate_inner(
            |key, bytes| {
                let val: T = from_value_bytes(bytes);
                match f(key, &val) {
                    crate::MigrateAction::Keep => crate::MigrateAction::Keep,
                    crate::MigrateAction::Update(new) => {
                        crate::MigrateAction::Update(to_bytes::<V, T>(&new))
                    }
                    crate::MigrateAction::Delete => crate::MigrateAction::Delete,
                }
            },
            fire_init,
        )
    }

    /// Read-only pass over all live entries (see `ConstMap::for_each`).
    pub fn for_each(&self, mut f: impl FnMut(&K, &T)) {
        self.inner.for_each(|key, bytes| {
            let val: T = from_value_bytes(bytes);
            f(key, &val);
        });
    }

    /// Unordered snapshot visitor for inspection; stops when `f` returns `Break`.
    pub(crate) fn for_each_while(
        &self,
        mut f: impl FnMut(&K, &T) -> ControlFlow<()>,
    ) -> ControlFlow<()> {
        self.inner.for_each_while(|key, bytes| {
            let val: T = from_value_bytes::<V, T>(bytes);
            f(key, &val)
        })
    }

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

    pub(crate) fn reconcile_projections(&self) -> DbResult<()> {
        self.inner.reconcile_projections()
    }

    pub(crate) fn effective_projection_limits(&self) -> (usize, usize) {
        self.inner.effective_projection_limits()
    }

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

    /// Ordered scan view. `None` unless the collection was opened with
    /// `iterable(true)`. Direction follows `Config::reversed` or `FixedConfig::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. Map and Tree
    /// scans are both weakly consistent; the Map implementation releases shard locks
    /// between batches, so concurrent writes may be observed partially.
    pub fn iter_view(&self) -> Option<ZeroMapIterView<'_, K, V, T, H, D>> {
        self.inner.iter_view()?;
        Some(ZeroMapIterView {
            inner: MapIterView::new(self, dir_from(self.inner.reversed())),
        })
    }

    /// Field-granular prefix / range / page builder. `None` unless opened `iterable(true)`.
    pub fn scope<const N: usize>(
        &self,
        key: &K,
    ) -> Option<crate::scope::Scope<'_, crate::scope::MapScan<&Self>, N>>
    where
        T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    {
        crate::scope::map_scope::<_, _, N>(
            self,
            key,
            dir_from(self.reversed()),
            self.iter_view().is_some(),
        )
    }

    /// Arc-owned prefix/range builder for [`crate::Tx::delete_scope`].
    /// Returns `None` unless this map was opened iterable.
    #[cfg(feature = "armour")]
    pub fn tx_scope<const N: usize>(
        self: &Arc<Self>,
        key: &K,
    ) -> Option<crate::TxScope<Self, K, N>> {
        self.iter_view().is_some().then(|| {
            crate::txn::scope::TxScope::from_parts(
                self.clone(),
                key,
                dir_from(self.reversed()),
                Self::tx_scope_keys,
            )
        })
    }

    fn tx_scope_keys<'a>(
        collection: &'a Self,
        start: Bound<&'a K>,
        end: Bound<&'a K>,
        dir: Dir,
    ) -> Box<dyn Iterator<Item = K> + 'a> {
        Box::new(MapIterView::new(collection, dir).keys_range_bounds(start, end))
    }

    /// First `(key, value)` in configured iteration order, or `None` if empty.
    ///
    /// Map scans are weakly consistent: shard locks are released between batches, so
    /// concurrent writes may be observed partially.
    pub fn first(&self) -> DbResult<Option<(K, T)>> {
        let view = self.iter_view().ok_or(DbError::NotImplemented)?;
        Ok(view.iter().next())
    }

    /// Last `(key, value)` in configured iteration order, or `None` if empty.
    ///
    /// Map scans are weakly consistent: shard locks are released between batches, so
    /// concurrent writes may be observed partially.
    pub fn last(&self) -> DbResult<Option<(K, T)>> {
        self.iter_view().ok_or(DbError::NotImplemented)?;
        Ok(
            MapIterView::new(self, dir_from(self.inner.reversed()).flip())
                .iter()
                .next(),
        )
    }

    /// Whether iterable scan yields descending key order.
    pub(crate) fn reversed(&self) -> bool {
        self.inner.reversed()
    }

    /// Shard count for limit-derived inspection batch sizing.
    pub(crate) fn inspection_shard_count(&self) -> usize {
        ShardSource::shard_count(self)
    }

    /// Strict bounded ordered scan for inspection (not the public `iter_view`).
    pub(crate) fn inspection_range_bounds(
        &self,
        start: Bound<&K>,
        end: Bound<&K>,
        batch: usize,
    ) -> TryMapIter<'_, ZeroMap<K, V, T, H, D>>
    where
        D: Durability,
    {
        MapIterView::with_batch(self, dir_from(self.inner.reversed()), batch)
            .try_range_bounds(start, end)
    }
}

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
    D: Durability,
> ShardSource for ZeroMap<K, V, T, H, D>
{
    type Key = K;
    type Val = T;

    fn shard_count(&self) -> usize {
        ShardSource::shard_count(self.inner.as_ref())
    }

    fn next_batch(
        &self,
        shard: usize,
        after: Option<&K>,
        dir: Dir,
        n: usize,
        range_start: Bound<crate::map_index::OrdKey<K>>,
        range_end: Bound<crate::map_index::OrdKey<K>>,
    ) -> Vec<(K, T)> {
        ShardSource::next_batch(
            self.inner.as_ref(),
            shard,
            after,
            dir,
            n,
            range_start,
            range_end,
        )
        .into_iter()
        .map(|(k, bytes)| (k, from_value_bytes::<V, T>(&bytes)))
        .collect()
    }

    fn next_keys_batch(
        &self,
        shard: usize,
        after: Option<&K>,
        dir: Dir,
        n: usize,
        range_start: Bound<crate::map_index::OrdKey<K>>,
        range_end: Bound<crate::map_index::OrdKey<K>>,
    ) -> Vec<K> {
        ShardSource::next_keys_batch(
            self.inner.as_ref(),
            shard,
            after,
            dir,
            n,
            range_start,
            range_end,
        )
    }
}

/// View over an iterable [`ZeroMap`]; exposes ordered iteration over `(K, T)`
/// entries and cheap keys-only iteration. Obtained via [`ZeroMap::iter_view`].
pub struct ZeroMapIterView<'a, K, const V: usize, T, H = NoHook, D = Bitcask>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
    D: Durability,
{
    inner: MapIterView<'a, ZeroMap<K, V, T, H, D>>,
}

impl<K, const V: usize, T, H, D> ZeroMapIterView<'_, K, V, T, H, D>
where
    K: Key + Send + Sync + Hash + Eq,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
    D: Durability,
{
    /// Full ordered scan in the collection's configured direction.
    pub fn iter(&self) -> impl Iterator<Item = (K, T)> + '_ {
        self.inner.iter()
    }

    /// Iterate entries in `[start, end)` — start inclusive, end exclusive.
    pub fn range(&self, start: &K, end: &K) -> impl Iterator<Item = (K, T)> + '_ {
        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>,
    ) -> impl Iterator<Item = (K, T)> + '_ {
        self.inner.range_bounds(start, end)
    }

    /// Iterate entries whose keys start with `prefix`.
    pub fn prefix_iter(&self, prefix: &[u8]) -> impl Iterator<Item = (K, T)> + '_ {
        self.inner.prefix_iter(prefix)
    }

    /// Keyset pagination in the collection's configured direction.
    pub fn paginate(&self, after: Option<&K>, limit: usize) -> Page<K, T> {
        self.inner.paginate(after, limit)
    }

    /// Full ordered **keys-only** scan — no value materialized.
    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, materializes no values.
    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
    /// [`ZeroMap::compare_delete`] using the value observed during iteration.
    ///
    /// Not atomic across the collection; a genuine storage error aborts the scan
    /// and is returned. Concurrent value/key changes (`CasMismatch`/`KeyNotFound`)
    /// are skipped.
    pub fn retain(&self, mut f: impl FnMut(&K, &T) -> bool) -> DbResult<()> {
        for (key, value) in self.inner.iter() {
            if f(&key, &value) {
                continue;
            }
            match self.inner.src().compare_delete(&key, &value) {
                Ok(()) => {}
                Err(DbError::CasMismatch) | Err(DbError::KeyNotFound) => {
                    tracing::debug!("retain: compare_delete skipped (concurrent change)");
                }
                Err(e) => return Err(e),
            }
        }
        Ok(())
    }
}

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
    H: TypedWriteHook<K, T>,
> CompactionIndex<K> for ZeroMap<K, V, T, H, Bitcask>
{
    fn update_if_match(
        &self,
        key: &K,
        old_loc: crate::disk_loc::DiskLoc,
        new_loc: crate::disk_loc::DiskLoc,
    ) -> bool {
        self.inner.update_if_match(key, old_loc, new_loc)
    }

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

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

// ---------------------------------------------------------------------------
// ZeroMapShard
// ---------------------------------------------------------------------------

/// Handle for atomic multi-key operations within a single shard.
/// Obtained via [`ZeroMap::atomic`].
#[repr(transparent)]
pub struct ZeroMapShard<
    'a,
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
    D: Durability = Bitcask,
    WH: WriteHook<K> = NoHook,
> {
    inner: ConstMapShard<'a, K, V, WH, D>,
    _marker: PhantomData<T>,
}

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    D: Durability,
    WH: WriteHook<K>,
> ZeroMapShard<'_, K, V, T, D, WH>
{
    pub fn put(&mut self, key: &K, value: &T) -> DbResult<Option<T>> {
        let bytes = to_bytes::<V, T>(value);
        self.inner
            .put(key, &bytes)
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
    }

    pub fn insert(&mut self, key: &K, value: &T) -> DbResult<()> {
        let bytes = to_bytes::<V, T>(value);
        self.inner.insert(key, &bytes)
    }

    pub fn delete(&mut self, key: &K) -> DbResult<Option<T>> {
        self.inner
            .delete(key)
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
    }

    pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
        Ok(self
            .inner
            .try_get(key)?
            .map(|bytes| from_value_bytes::<V, T>(&bytes)))
    }

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

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

    /// Atomically read-modify-write. Returns `Some(T)` (the **new** value)
    /// if key existed, `None` otherwise.
    pub fn update(&mut self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        self.inner
            .update(key, |bytes| {
                let val = from_value_bytes::<V, T>(bytes);
                to_bytes::<V, T>(&f(&val))
            })
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
    }

    /// Like [`update()`](Self::update), but returns `Some(T)` with the **old** value.
    pub fn fetch_update(&mut self, key: &K, f: impl FnOnce(&T) -> T) -> DbResult<Option<T>> {
        self.inner
            .fetch_update(key, |bytes| {
                let val = from_value_bytes::<V, T>(bytes);
                to_bytes::<V, T>(&f(&val))
            })
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
    }

    /// Decide-and-write within the shard. The closure must be short — shard lock is held.
    pub fn upsert<U: crate::IntoUpsert<T>>(
        &mut self,
        key: &K,
        f: impl FnOnce(Option<&T>) -> U,
    ) -> crate::DbResult<U::Out> {
        crate::upsert::upsert_via_held(self, key, f)
    }
}

#[cfg(feature = "armour")]
impl<T, const V: usize, H> crate::armour::collection::Collection
    for ZeroMap<T::SelfId, V, T, H, Bitcask>
where
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync,
    H: crate::hook::TypedWriteHook<T::SelfId, T>,
    T::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
{
    fn name(&self) -> &str {
        T::NAME
    }
    fn len(&self) -> usize {
        self.len()
    }
    fn compact(&self) -> crate::DbResult<usize> {
        self.compact()
    }
    fn flush(&self) -> crate::DbResult<()> {
        crate::projection::collection_flush_fault()?;
        self.flush_buffers()?;
        if self.config().hints.unwrap_or(false) {
            self.sync_hints()?;
        }
        Ok(())
    }
    fn periodic_flush(&self) -> crate::DbResult<()> {
        self.inner.durable_flush()
    }
    fn durable_flush(&self) -> crate::DbResult<()> {
        self.inner.durable_flush()
    }
    fn clean_shutdown(&self) -> crate::DbResult<()> {
        ZeroMap::<T::SelfId, V, T, H, Bitcask>::clean_shutdown(self)
    }
    fn sweep_expired(
        &self,
        now_ms: u64,
        budget: usize,
    ) -> crate::DbResult<crate::ttl::SweepOutcome> {
        self.sweep_expired(now_ms, budget)
    }
    fn set_ttl_sweep(&self, enabled: bool) -> crate::DbResult<()> {
        self.set_ttl_sweep(enabled)
    }
    fn rebuild_ttl_index(&self) -> crate::DbResult<()> {
        self.rebuild_ttl_index()
    }
    fn reset_counter_dirty_marker(&self) -> bool {
        self.tx_inner()
            .counter_dirty
            .get()
            .is_some_and(|d| d.take_marked())
    }
    fn reconcile_counters(&self) -> crate::DbResult<u64> {
        self.tx_inner().reconcile_counters()
    }
    fn counter_target_names(&self) -> Vec<&'static str> {
        self.tx_inner().counter_target_names()
    }
    fn close_watches(&self) {
        if let Some(channel) = self.inner.watch_channel() {
            channel.close_terminal();
        }
    }
}

#[cfg(feature = "armour")]
impl<T, const V: usize, H> crate::armour::collection::Collection
    for ZeroMap<T::SelfId, V, T, H, Fixed>
where
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync,
    H: crate::hook::TypedWriteHook<T::SelfId, T>,
    T::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
{
    fn name(&self) -> &str {
        T::NAME
    }
    fn len(&self) -> usize {
        self.len()
    }
    fn compact(&self) -> crate::DbResult<usize> {
        Ok(0)
    }
    fn flush(&self) -> crate::DbResult<()> {
        crate::projection::collection_flush_fault()?;
        // FixedStore writes in-place; no hint file. flush() makes pending
        // writes durable.
        self.flush()
    }
    fn periodic_flush(&self) -> crate::DbResult<()> {
        // FixedStore: fdatasync (data already in the page cache via pwrite).
        // Resolves to the inherent ZeroMap<Fixed>::flush(), same as the
        // Collection::flush() impl above.
        self.flush()
    }
    fn durable_flush(&self) -> crate::DbResult<()> {
        self.flush()
    }
    fn clean_shutdown(&self) -> crate::DbResult<()> {
        ZeroMap::<T::SelfId, V, T, H, Fixed>::clean_shutdown(self)
    }
    fn sweep_expired(
        &self,
        now_ms: u64,
        budget: usize,
    ) -> crate::DbResult<crate::ttl::SweepOutcome> {
        self.sweep_expired(now_ms, budget)
    }
    fn set_ttl_sweep(&self, enabled: bool) -> crate::DbResult<()> {
        self.set_ttl_sweep(enabled)
    }
    fn rebuild_ttl_index(&self) -> crate::DbResult<()> {
        self.rebuild_ttl_index()
    }
    fn reset_counter_dirty_marker(&self) -> bool {
        self.tx_inner()
            .counter_dirty
            .get()
            .is_some_and(|d| d.take_marked())
    }
    fn reconcile_counters(&self) -> crate::DbResult<u64> {
        self.tx_inner().reconcile_counters()
    }
    fn counter_target_names(&self) -> Vec<&'static str> {
        self.tx_inner().counter_target_names()
    }
    fn close_watches(&self) {
        if let Some(channel) = self.inner.watch_channel() {
            channel.close_terminal();
        }
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::sealed::Sealed for ZeroMap<K, V, T, H, Bitcask>
{
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::ProjectableIndex for ZeroMap<K, V, T, H, Bitcask>
{
    fn collection_name(&self) -> &'static str {
        T::NAME
    }

    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        let v: T = zerocopy::FromBytes::read_from_bytes(value)
            .map_err(|_| DbError::FormatMismatch("zero index value decode".into()))?;
        self.put(&k, &v)?;
        Ok(())
    }

    fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        self.delete(&k)?;
        Ok(())
    }

    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
        Ok(a == b)
    }

    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
        let mut err: Option<DbError> = None;
        self.for_each(|k, v| {
            if err.is_some() {
                return;
            }
            if let Err(e) = f(k.as_bytes(), zerocopy::IntoBytes::as_bytes(v)) {
                err = Some(e);
            }
        });
        match err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    fn durable_flush(&self) -> DbResult<()> {
        ZeroMap::flush(self)
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::ProjectableIndexTx for ZeroMap<K, V, T, H, Bitcask>
{
    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
        self
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::sealed::Sealed for ZeroMap<K, V, T, H, Fixed>
{
}

impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::ProjectableIndex for ZeroMap<K, V, T, H, Fixed>
{
    fn collection_name(&self) -> &'static str {
        T::NAME
    }

    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        let v: T = zerocopy::FromBytes::read_from_bytes(value)
            .map_err(|_| DbError::FormatMismatch("zero index value decode".into()))?;
        self.put(&k, &v)?;
        Ok(())
    }

    fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        self.delete(&k)?;
        Ok(())
    }

    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
        Ok(a == b)
    }

    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
        let mut err: Option<DbError> = None;
        self.for_each(|k, v| {
            if err.is_some() {
                return;
            }
            if let Err(e) = f(k.as_bytes(), zerocopy::IntoBytes::as_bytes(v)) {
                err = Some(e);
            }
        });
        match err {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    fn durable_flush(&self) -> DbResult<()> {
        ZeroMap::flush(self)
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::ProjectableIndexTx for ZeroMap<K, V, T, H, Fixed>
{
    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
        self
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::IndexKeyed for ZeroMap<K, V, T, H, Bitcask>
{
    type K = K;
    type V = T;

    fn encode_value(&self, v: &T) -> DbResult<Vec<u8>> {
        Ok(zerocopy::IntoBytes::as_bytes(v).to_vec())
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::counter::CounterCollection for ZeroMap<K, V, T, H, Bitcask>
{
    fn decode_value(&self, bytes: &[u8]) -> DbResult<T> {
        zerocopy::FromBytes::read_from_bytes(bytes)
            .map_err(|_| DbError::FormatMismatch("zero counter value decode".into()))
    }

    /// Read path — no shard write lock: `get` is an O(1) index lookup.
    fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        self.get(&k).map(|v| self.encode_value(&v)).transpose()
    }

    fn contains_bytes(&self, key: &[u8]) -> DbResult<bool> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        Ok(self.contains(&k))
    }

    fn upsert_counter(
        &self,
        key: &[u8],
        f: &mut dyn FnMut(Option<&T>) -> Upsert<T>,
    ) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        self.upsert(&k, f)?;
        Ok(())
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::projection::IndexKeyed for ZeroMap<K, V, T, H, Fixed>
{
    type K = K;
    type V = T;

    fn encode_value(&self, v: &T) -> DbResult<Vec<u8>> {
        Ok(zerocopy::IntoBytes::as_bytes(v).to_vec())
    }
}

#[cfg(feature = "typed-tree")]
impl<
    K: Key + Send + Sync + std::hash::Hash + Eq + 'static,
    const V: usize,
    T: crate::CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Default
        + PartialEq
        + Send
        + Sync
        + 'static,
    H: TypedWriteHook<K, T> + 'static,
> crate::counter::CounterCollection for ZeroMap<K, V, T, H, Fixed>
{
    fn decode_value(&self, bytes: &[u8]) -> DbResult<T> {
        zerocopy::FromBytes::read_from_bytes(bytes)
            .map_err(|_| DbError::FormatMismatch("zero counter value decode".into()))
    }

    /// Read path — no shard write lock: `get` is an O(1) index lookup.
    fn read_bytes(&self, key: &[u8]) -> DbResult<Option<Vec<u8>>> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        self.get(&k).map(|v| self.encode_value(&v)).transpose()
    }

    fn contains_bytes(&self, key: &[u8]) -> DbResult<bool> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        Ok(self.contains(&k))
    }

    fn upsert_counter(
        &self,
        key: &[u8],
        f: &mut dyn FnMut(Option<&T>) -> Upsert<T>,
    ) -> DbResult<()> {
        if key.len() != std::mem::size_of::<K>() {
            return Err(DbError::CorruptedEntry { offset: 0 });
        }
        let k = K::from_bytes(key);
        self.upsert(&k, f)?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Config;
    use crate::hook::TypedWriteHook;
    use std::collections::HashMap as StdHashMap;
    use std::mem::size_of;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering as AO};
    use tempfile::tempdir;
    use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};

    #[derive(
        FromBytes, IntoBytes, Immutable, KnownLayout, Clone, Copy, Debug, PartialEq, Eq, Default,
    )]
    #[repr(C)]
    struct Counters {
        views: u64,
        likes: u64,
    }

    type Map = ZeroMap<[u8; 8], { size_of::<Counters>() }, Counters>;

    fn open(dir: &std::path::Path) -> Map {
        ZeroMap::<[u8; 8], { size_of::<Counters>() }, Counters>::open(dir, Config::test()).unwrap()
    }

    fn key_on_other_shard(map: &Map, anchor: &[u8; 8]) -> [u8; 8] {
        let anchor_shard = map.shard_for(anchor);
        (2u64..1000)
            .map(|i| i.to_be_bytes())
            .find(|k| map.shard_for(k) != anchor_shard)
            .expect("key on different shard")
    }

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

        let key_a = 1u64.to_be_bytes();
        let key_b = key_on_other_shard(&map, &key_a);
        let key_c = key_on_other_shard(&map, &key_b);

        let expected = StdHashMap::from([
            (
                key_a,
                Counters {
                    views: 10,
                    likes: 1,
                },
            ),
            (
                key_b,
                Counters {
                    views: 20,
                    likes: 2,
                },
            ),
            (
                key_c,
                Counters {
                    views: 30,
                    likes: 3,
                },
            ),
        ]);
        for (key, value) in &expected {
            map.put(key, value).unwrap();
        }

        let mut seen = StdHashMap::new();
        map.for_each(|key, value| {
            seen.insert(*key, *value);
        });

        assert_eq!(seen, expected);
        assert_ne!(map.shard_for(&key_a), map.shard_for(&key_b));
    }

    struct ZHook {
        writes: Arc<AtomicUsize>,
        last_new: Arc<crate::sync::Mutex<Option<u32>>>,
    }

    impl ZHook {
        fn new() -> Self {
            Self {
                writes: Arc::new(AtomicUsize::new(0)),
                last_new: Arc::new(crate::sync::Mutex::new(None)),
            }
        }
    }

    impl TypedWriteHook<[u8; 8], u32> for ZHook {
        fn on_write(&self, _k: &[u8; 8], _old: Option<&u32>, new: Option<&u32>) {
            self.writes.fetch_add(1, AO::Relaxed);
            *crate::sync::lock(&self.last_new) = new.copied();
        }
    }

    #[test]
    fn zero_map_atomic_fires_typed_hook() {
        let dir = tempfile::tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let hook = ZHook::new();
        let writes = Arc::clone(&hook.writes);
        let last_new = Arc::clone(&hook.last_new);
        let map = ZeroMap::<[u8; 8], 4, u32, _>::open_hooked(dir.path(), cfg, hook).unwrap();
        let k = 1u64.to_be_bytes();
        map.atomic(&k, |s| {
            s.put(&k, &42u32)?;
            Ok(())
        })
        .unwrap();
        assert_eq!(writes.load(AO::Relaxed), 1);
        assert_eq!(*crate::sync::lock(&last_new), Some(42));
    }

    #[test]
    fn zero_map_first_last_contract_both_backends() {
        let bitcask_dir = tempdir().unwrap();
        let bitcask_cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .reversed(false)
            .build();
        let bitcask = ZeroMap::<[u8; 8], 8, u64>::open(bitcask_dir.path(), bitcask_cfg).unwrap();
        assert!(bitcask.first().unwrap().is_none());
        assert!(bitcask.last().unwrap().is_none());
        for id in [3_u64, 1, 2] {
            bitcask.put(&id.to_be_bytes(), &(id * 10)).unwrap();
        }
        assert_eq!(
            bitcask
                .first()
                .unwrap()
                .map(|(k, v)| (u64::from_be_bytes(k), v)),
            Some((1, 10))
        );
        assert_eq!(
            bitcask
                .last()
                .unwrap()
                .map(|(k, v)| (u64::from_be_bytes(k), v)),
            Some((3, 30))
        );

        let fixed_dir = tempdir().unwrap();
        let fixed_cfg = crate::FixedConfig {
            iterable: true,
            reversed: false,
            ..crate::FixedConfig::test()
        };
        let fixed =
            ZeroMap::<[u8; 8], 8, u64, NoHook, Fixed>::open(fixed_dir.path(), fixed_cfg).unwrap();
        assert!(fixed.first().unwrap().is_none());
        assert!(fixed.last().unwrap().is_none());
        for id in [3_u64, 1, 2] {
            fixed.put(&id.to_be_bytes(), &(id * 10)).unwrap();
        }
        assert_eq!(
            fixed
                .first()
                .unwrap()
                .map(|(k, v)| (u64::from_be_bytes(k), v)),
            Some((1, 10))
        );
        assert_eq!(
            fixed
                .last()
                .unwrap()
                .map(|(k, v)| (u64::from_be_bytes(k), v)),
            Some((3, 30))
        );

        let no_index_dir = tempdir().unwrap();
        let no_index =
            ZeroMap::<[u8; 8], 8, u64>::open(no_index_dir.path(), Config::test()).unwrap();
        assert!(matches!(no_index.first(), Err(DbError::NotImplemented)));
        assert!(matches!(no_index.last(), Err(DbError::NotImplemented)));
    }

    #[test]
    fn zero_map_fixed_direction() {
        #[derive(Clone, Copy, FromBytes, IntoBytes, Immutable)]
        #[repr(C)]
        struct V {
            x: u32,
        }
        let dir = tempdir().unwrap();
        let cfg = crate::FixedConfig {
            iterable: true,
            reversed: false,
            ..crate::FixedConfig::test()
        };
        let m = ZeroMap::<[u8; 8], { size_of::<V>() }, V, NoHook, Fixed>::open(dir.path(), cfg)
            .unwrap();
        for i in 1u64..=3 {
            m.put(&i.to_be_bytes(), &V { x: i as u32 }).unwrap();
        }
        let keys: Vec<u64> = m
            .iter_view()
            .unwrap()
            .keys()
            .map(u64::from_be_bytes)
            .collect();
        assert_eq!(keys, [1, 2, 3], "ZeroMap<Fixed> reversed=false ASC");
    }

    #[test]
    fn zero_map_iter_yields_decoded_values_desc() {
        #[derive(Clone, Copy, FromBytes, IntoBytes, Immutable)]
        #[repr(C)]
        struct Val {
            x: u32,
        }
        let dir = tempdir().unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .build();
        let m = ZeroMap::<[u8; 2], { size_of::<Val>() }, Val>::open(dir.path(), cfg).unwrap();
        m.put(&[1, 0], &Val { x: 11 }).unwrap();
        m.put(&[2, 0], &Val { x: 22 }).unwrap();
        let got: Vec<(u8, u32)> = m
            .iter_view()
            .unwrap()
            .iter()
            .map(|(k, v)| (k[0], v.x))
            .collect();
        assert_eq!(got, [(2, 22), (1, 11)]);
        // keys-only: no zerocopy value decode.
        let keys: Vec<u8> = m.iter_view().unwrap().keys().map(|k| k[0]).collect();
        assert_eq!(keys, [2, 1]);
    }

    #[test]
    fn zero_map_shard_update_and_fetch_update() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        let map = ZeroMap::<[u8; 8], 4, u32>::open(dir.path(), cfg).unwrap();
        let key = 1u64.to_be_bytes();
        map.put(&key, &100u32).unwrap();

        let (upd, fetched, missing) = map
            .atomic(&key, |s| {
                let upd = s.update(&key, |old| old + 1)?; // new -> 101
                let fetched = s.fetch_update(&key, |old| old + 1)?; // old -> 101
                let missing = s.update(&2u64.to_be_bytes(), |old| *old)?;
                Ok((upd, fetched, missing))
            })
            .expect("atomic");

        assert_eq!(upd, Some(101));
        assert_eq!(fetched, Some(101));
        assert_eq!(missing, None);
        assert_eq!(map.get(&key), Some(102));
    }

    #[test]
    fn for_each_while_stops_after_break_without_visiting_later_shards() {
        use std::ops::ControlFlow;
        use std::sync::atomic::{AtomicUsize, Ordering};

        let dir = tempdir().unwrap();
        let map = open(dir.path());

        let key_a = 1u64.to_be_bytes();
        let key_b = key_on_other_shard(&map, &key_a);
        assert_ne!(map.shard_for(&key_a), map.shard_for(&key_b));

        map.put(&key_a, &Counters { views: 1, likes: 0 }).unwrap();
        map.put(&key_b, &Counters { views: 2, likes: 0 }).unwrap();

        let seen = AtomicUsize::new(0);
        let flow = map.for_each_while(|_k, _v| {
            seen.fetch_add(1, Ordering::Relaxed);
            ControlFlow::Break(())
        });
        assert_eq!(seen.load(Ordering::Relaxed), 1);
        assert!(matches!(flow, ControlFlow::Break(())));
    }
}