armdb 0.6.0

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

use crate::Key;
use crate::batch::{Applied, BatchWrite};
use crate::compaction::CompactionIndex;
use crate::config::Config;
use crate::const_tree::{ConstIter, ConstShard, ConstTree};
use crate::disk_loc::DiskLoc;
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::key::Location;

/// A tree with fixed-size keys and zerocopy-compatible values.
///
/// Thin wrapper around [`ConstTree<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: `ZeroTree::<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.
///
/// For complex types (String, Vec, nested structs) use [`TypedTree`](crate::TypedTree) instead.
///
/// # Usage
///
/// ```ignore
/// use zerocopy::{FromBytes, IntoBytes, Immutable, KnownLayout};
///
/// #[derive(FromBytes, IntoBytes, Immutable, KnownLayout, Clone, Copy)]
/// #[repr(C)]
/// struct Counters {
///     views: u64,
///     likes: u64,
/// }
///
/// let tree = ZeroTree::<[u8; 16], { size_of::<Counters>() }, Counters>::open(
///     "data/counters",
///     Config::balanced().build(),
/// )?;
/// tree.put(&key, &Counters { views: 1, likes: 0 })?;
/// if let Some(c) = tree.get(&key) {
///     println!("views: {}", c.views);
/// }
/// ```
///
/// # Iteration
///
/// `iter()`, `range()`, and `prefix_iter()` return [`ZeroIter`] which
/// implements `Iterator + DoubleEndedIterator` with `Item = (K, T)`.
///
/// # Batch operations
///
/// [`get_many`](Self::get_many) does one sorted finger-walk over the index for a
/// list of keys; [`update_many`](Self::update_many) applies a per-key closure
/// grouped by shard (per-shard isolation, no global atomicity). See
/// `docs/superpowers/specs/26-06-24-armdb-batch-tree-ops.md`.
pub struct ZeroTree<
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable = [u8; V],
    H: TypedWriteHook<K, T> = NoHook,
    D: Durability = Bitcask,
> {
    inner: ConstTree<K, V, ZeroHookAdapter<K, T, H>, D>,
    _marker: PhantomData<T>,
}

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

impl<
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
> ZeroTree<K, V, T, NoHook, Bitcask>
{
    /// Open or create a `ZeroTree` at the given path.
    /// Recovers the index from existing data files on disk.
    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: ConstTree::open_hooked(path, config, adapter)?,
            _marker: PhantomData,
        })
    }
}

impl<
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
> ZeroTree<K, V, T, H, Bitcask>
{
    /// Open or create a `ZeroTree` with a write hook for secondary index maintenance.
    pub fn open_hooked(
        path: impl AsRef<std::path::Path>,
        config: Config,
        hook: H,
    ) -> DbResult<Self> {
        const { assert!(size_of::<T>() == V) }
        let adapter = ZeroHookAdapter {
            inner: hook,
            _marker: PhantomData,
        };
        Ok(Self {
            inner: ConstTree::open_hooked(path, config, adapter)?,
            _marker: PhantomData,
        })
    }

    /// Graceful shutdown: write hint files, 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()
    }

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

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

impl<
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
> ZeroTree<K, V, T, NoHook, Fixed>
{
    /// Open or create a `ZeroTree` with Fixed (fixed-slot) backend.
    /// Recovers the index from existing data files on disk.
    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: ConstTree::open_with_hook(path, config, adapter)?,
            _marker: PhantomData,
        })
    }
}

impl<
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
> ZeroTree<K, V, T, H, Fixed>
{
    /// Open or create a `ZeroTree` with a write hook, using Fixed (fixed-slot) backend.
    pub fn open_with_hook(
        path: impl AsRef<std::path::Path>,
        config: FixedConfig,
        hook: H,
    ) -> DbResult<Self> {
        const { assert!(size_of::<T>() == V) }
        let adapter = ZeroHookAdapter {
            inner: hook,
            _marker: PhantomData,
        };
        Ok(Self {
            inner: ConstTree::open_with_hook(path, config, adapter)?,
            _marker: PhantomData,
        })
    }

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

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

impl<
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
    D: Durability,
> ZeroTree<K, V, T, H, D>
{
    // -- Reads ----------------------------------------------------------------

    /// Get a value by key. Lock-free, zero disk I/O. Returns a copy of `T`.
    ///
    /// **O(log n)** — SkipList point lookup.
    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)
    }

    /// Return the first entry in index order, or `None` if empty.
    /// With `reversed=true` (default): the entry with the largest key.
    /// O(1) — follows head's level-0 pointer.
    pub fn first(&self) -> Option<(K, T)> {
        self.inner
            .first()
            .map(|(k, v)| (k, from_value_bytes::<V, T>(&v)))
    }

    /// Return the last entry in index order, or `None` if empty.
    /// With `reversed=true` (default): the entry with the smallest key.
    pub fn last(&self) -> Option<(K, T)> {
        self.inner
            .last()
            .map(|(k, v)| (k, from_value_bytes::<V, T>(&v)))
    }

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

    // -- 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 ZeroShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>) -> DbResult<R>,
    ) -> DbResult<R> {
        self.inner.atomic(shard_key, |const_shard| {
            // SAFETY: ZeroShard is #[repr(transparent)] over ConstShard with the SAME
            // hook type (ZeroHookAdapter<K, T, H>). PhantomData<T> is ZST.
            let shard = unsafe {
                &mut *(const_shard as *mut ConstShard<'_, K, V, ZeroHookAdapter<K, T, H>, D>
                    as *mut ZeroShard<'_, K, V, T, D, ZeroHookAdapter<K, T, H>>)
            };
            f(shard)
        })
    }

    // -- Iteration ------------------------------------------------------------

    /// Iterate entries whose keys start with `prefix`.
    ///
    /// `reversed=true` (default): yields matching keys in DESC order.
    /// `next()` is O(1), `next_back()` is O(log n).
    pub fn prefix_iter(&self, prefix: &[u8]) -> ZeroIter<'_, K, V, T, D::Loc> {
        ZeroIter {
            inner: self.inner.prefix_iter(prefix),
            _marker: PhantomData,
        }
    }

    /// Iterate all entries in index order.
    ///
    /// `reversed=true` (default): DESC. `reversed=false`: ASC.
    /// `next()` is O(1), `next_back()` is O(log n).
    pub fn iter(&self) -> ZeroIter<'_, K, V, T, D::Loc> {
        ZeroIter {
            inner: self.inner.iter(),
            _marker: PhantomData,
        }
    }

    /// Iterate entries in `[start, end)` — start inclusive, end exclusive.
    ///
    /// `reversed=true` (default): DESC within range. `reversed=false`: ASC.
    /// `next()` is O(1), `next_back()` is O(log n).
    pub fn range(&self, start: &K, end: &K) -> ZeroIter<'_, K, V, T, D::Loc> {
        self.range_bounds(Bound::Included(start), Bound::Excluded(end))
    }

    /// Iterate entries in range defined by `start` and `end` bounds.
    ///
    /// Unlike [`range()`](Self::range), allows `Included`, `Excluded`, or `Unbounded`
    /// for each bound independently.
    ///
    /// `reversed=true` (default): DESC within range. `reversed=false`: ASC.
    /// `next()` is O(1), `next_back()` is O(log n).
    pub fn range_bounds(&self, start: Bound<&K>, end: Bound<&K>) -> ZeroIter<'_, K, V, T, D::Loc> {
        ZeroIter {
            inner: self.inner.range_bounds(start, end),
            _marker: PhantomData,
        }
    }

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

    // -- Batch ----------------------------------------------------------------

    /// Batch point-read returning copies of `T`, aligned to `keys`.
    pub fn get_many(&self, keys: &[K]) -> Vec<Option<T>> {
        self.inner
            .get_many(keys)
            .into_iter()
            .map(|opt| opt.map(|b| from_value_bytes::<V, T>(&b)))
            .collect()
    }

    /// Batch read-modify-write over typed values. See [`ConstTree::update_many`].
    pub fn update_many<P>(
        &self,
        items: Vec<(K, P)>,
        f: impl Fn(&K, Option<&T>, &P) -> BatchWrite<T>,
    ) -> DbResult<Vec<(K, Applied<T>)>> {
        let inner_out = self.inner.update_many(items, |k, old_bytes, p| {
            let old_t: Option<T> = old_bytes.map(|b| from_value_bytes::<V, T>(b));
            match f(k, old_t.as_ref(), p) {
                BatchWrite::Set(v) => BatchWrite::Set(to_bytes::<V, T>(&v)),
                BatchWrite::Keep => BatchWrite::Keep,
                BatchWrite::Delete => BatchWrite::Delete,
            }
        })?;
        Ok(inner_out
            .into_iter()
            .map(|(k, applied)| {
                let applied_t = match applied {
                    Applied::Written { old, new } => Applied::Written {
                        old: old.map(|b| from_value_bytes::<V, T>(&b)),
                        new: from_value_bytes::<V, T>(&new),
                    },
                    Applied::Kept => Applied::Kept,
                    Applied::Deleted(b) => Applied::Deleted(from_value_bytes::<V, T>(&b)),
                };
                (k, applied_t)
            })
            .collect())
    }

    // -- 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.inner.migrate(|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,
            }
        })
    }

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

impl<
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable + Send + Sync,
    H: TypedWriteHook<K, T>,
> CompactionIndex<K> for ZeroTree<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)
    }
}

// ---------------------------------------------------------------------------
// ZeroIter
// ---------------------------------------------------------------------------

/// Iterator over entries in a `ZeroTree`. Wraps [`ConstIter`] and converts
/// `[u8; V]` values to `T` via zerocopy.
pub struct ZeroIter<'a, K: Key, const V: usize, T = [u8; V], L: Location = DiskLoc> {
    inner: ConstIter<'a, K, V, L>,
    _marker: PhantomData<T>,
}

impl<
    'a,
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    L: Location,
> Iterator for ZeroIter<'a, K, V, T, L>
{
    type Item = (K, T);

    fn next(&mut self) -> Option<Self::Item> {
        self.inner
            .next()
            .map(|(k, v)| (k, from_value_bytes::<V, T>(&v)))
    }
}

impl<
    'a,
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    L: Location,
> DoubleEndedIterator for ZeroIter<'a, K, V, T, L>
{
    fn next_back(&mut self) -> Option<Self::Item> {
        self.inner
            .next_back()
            .map(|(k, v)| (k, from_value_bytes::<V, T>(&v)))
    }
}

// ---------------------------------------------------------------------------
// ZeroShard
// ---------------------------------------------------------------------------

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

impl<
    K: Key,
    const V: usize,
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    D: Durability,
    WH: WriteHook<K>,
> ZeroShard<'_, 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 get(&self, key: &K) -> Option<T> {
        let bytes = self.inner.get(key)?;
        Some(from_value_bytes::<V, T>(&bytes))
    }

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

    pub fn contains(&self, key: &K) -> bool {
        self.inner.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)))
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

#[inline(always)]
pub(crate) fn to_bytes<const V: usize, T: Copy + zerocopy::IntoBytes + zerocopy::Immutable>(
    value: &T,
) -> [u8; V] {
    debug_assert_eq!(size_of::<T>(), V);
    let mut buf = [0u8; V];
    buf.copy_from_slice(zerocopy::IntoBytes::as_bytes(value));
    buf
}

#[inline(always)]
pub(crate) fn from_value_bytes<const V: usize, T: Copy + zerocopy::FromBytes>(
    bytes: &[u8; V],
) -> T {
    debug_assert_eq!(V, size_of::<T>());
    zerocopy::FromBytes::read_from_bytes(bytes.as_slice())
        .expect("size validated by const V == size_of::<T>()")
}

#[cfg(feature = "armour")]
impl<T, const V: usize, H> crate::armour::collection::Collection
    for ZeroTree<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 + Ord,
{
    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<()> {
        self.flush_buffers()?;
        self.sync_hints()?;
        Ok(())
    }
    fn periodic_flush(&self) -> crate::DbResult<()> {
        self.flush_buffers()
    }
}

#[cfg(feature = "armour")]
impl<T, const V: usize, H> crate::armour::collection::Collection
    for ZeroTree<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 + Ord,
{
    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<()> {
        // 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 ZeroTree<Fixed>::flush(), same as the
        // Collection::flush() impl above.
        self.flush()
    }
}

// ==========================================================================
// MultiTx — cross-collection transaction support (feature `armour`)
//
// ZeroTree wraps a ConstTree whose hook is ZeroHookAdapter; the multi-tx handle
// wraps ConstTx and layers the zerocopy `T <-> [u8; V]` codec, delegating every
// lock/teardown step to the inner ConstTree's MultiTx impl.
// ==========================================================================

/// Multi-shard transaction handle for [`ZeroTree`] inside `Db::atomicN`.
#[cfg(feature = "armour")]
pub struct ZeroTx<'a, K: Key, const V: usize, T, H, D: Durability>
where
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
{
    inner: crate::const_tree::ConstTx<'a, K, V, ZeroHookAdapter<K, T, H>, D>,
}

#[cfg(feature = "armour")]
impl<K: Key, const V: usize, T, H, D: Durability> ZeroTx<'_, K, V, T, H, D>
where
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
{
    pub fn try_get(&self, key: &K) -> DbResult<Option<T>> {
        Ok(self
            .inner
            .try_get(key)?
            .map(|b| from_value_bytes::<V, T>(&b)))
    }

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

#[cfg(feature = "armour")]
impl<K: Key, const V: usize, T, H, D: Durability> crate::armour::multi_tx::MultiTx
    for ZeroTree<K, V, T, H, D>
where
    T: Copy + zerocopy::IntoBytes + zerocopy::FromBytes + zerocopy::Immutable,
    H: TypedWriteHook<K, T>,
{
    type Key = K;
    type Tx<'a>
        = ZeroTx<'a, K, V, T, H, D>
    where
        Self: 'a;

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

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

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

    fn release_locks(
        &self,
        tx: &mut ZeroTx<'_, K, V, T, H, D>,
    ) -> 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: ZeroTx<'_, K, V, T, H, D>) {
        crate::armour::multi_tx::MultiTx::replay_hooks(&self.inner, tx.inner)
    }
}

#[cfg(test)]
mod tests {
    use super::ZeroTree;
    use crate::Config;
    use crate::hook::TypedWriteHook;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering as AO};

    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_tree_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 tree = ZeroTree::<[u8; 8], 4, u32, _>::open_hooked(dir.path(), cfg, hook).unwrap();
        let k = 1u64.to_be_bytes();
        tree.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_get_many_and_update_many_typed() {
        use crate::{Applied, BatchWrite};
        let dir = tempfile::tempdir().unwrap();
        let cfg = Config::test();
        let tree = ZeroTree::<[u8; 8], 8, u64>::open(dir.path(), cfg).unwrap();
        tree.put(&1u64.to_be_bytes(), &100).unwrap();
        tree.put(&2u64.to_be_bytes(), &200).unwrap();

        let keys: Vec<[u8; 8]> = vec![2u64.to_be_bytes(), 9u64.to_be_bytes(), 1u64.to_be_bytes()];
        assert_eq!(tree.get_many(&keys), vec![Some(200u64), None, Some(100u64)]);

        let items: Vec<([u8; 8], u64)> = vec![(1u64.to_be_bytes(), 5), (3u64.to_be_bytes(), 42)];
        let out = tree
            .update_many(items, |_k, cur, d| {
                BatchWrite::Set(cur.copied().unwrap_or(0) + *d)
            })
            .unwrap();
        assert_eq!(
            out[0].1,
            Applied::Written {
                old: Some(100u64),
                new: 105u64
            }
        );
        assert_eq!(
            out[1].1,
            Applied::Written {
                old: None,
                new: 42u64
            }
        );
        assert_eq!(tree.get(&1u64.to_be_bytes()), Some(105));
        assert_eq!(tree.get(&3u64.to_be_bytes()), Some(42));
    }

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

        let (upd, fetched, missing) = tree
            .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!(tree.get(&key), Some(102));
    }

    #[test]
    fn zero_tree_fixed_direction() {
        use crate::FixedConfig;
        use crate::durability::Fixed;
        use crate::hook::NoHook;
        let dir = tempfile::tempdir().unwrap();
        let tree =
            ZeroTree::<[u8; 8], 4, u32, NoHook, Fixed>::open(dir.path(), FixedConfig::test())
                .unwrap();
        for i in 1u32..=3 {
            tree.put(&(i as u64).to_be_bytes(), &i).unwrap();
        }
        let desc: Vec<u64> = tree.iter().map(|(k, _)| u64::from_be_bytes(k)).collect();
        assert_eq!(desc, [3, 2, 1], "ZeroTree<Fixed> default DESC");

        let dir2 = tempfile::tempdir().unwrap();
        let cfg = FixedConfig {
            reversed: false,
            ..FixedConfig::test()
        };
        let tree2 = ZeroTree::<[u8; 8], 4, u32, NoHook, Fixed>::open(dir2.path(), cfg).unwrap();
        for i in 1u32..=3 {
            tree2.put(&(i as u64).to_be_bytes(), &i).unwrap();
        }
        let asc: Vec<u64> = tree2.iter().map(|(k, _)| u64::from_be_bytes(k)).collect();
        assert_eq!(asc, [1, 2, 3], "ZeroTree<Fixed> reversed=false ASC");
    }
}