armdb 0.14.4

use crate::Key;
use crate::error::{DbError, DbResult};

#[cfg(feature = "typed-tree")]
use crate::projection::ProjectionSink;

use std::mem::size_of;

/// Trait for receiving write notifications from tree/map operations.
///
/// Used for maintaining secondary indexes, audit logs, or any side-effect
/// that must observe every mutation synchronously.
///
/// # Zero-overhead default
///
/// The default implementation [`NoHook`] has `NEEDS_OLD_VALUE = false`,
/// `NEEDS_WRITE = false`, and an empty `on_write`. The compiler eliminates
/// all hook-related branches and calls when `H = NoHook`.
///
/// # Associated constants
///
/// - `NEEDS_OLD_VALUE`: when `false`, skips disk I/O for the old value in
///   VarTree/VarMap — `old` is `None` in `on_write`.
/// - `NEEDS_INIT`: when `true`, `on_init` is called for every live entry
///   during collection open.
/// - `NEEDS_WRITE`: when `false`, `on_write` may be skipped entirely by
///   the collection or its adapter. Set to `false` on `NoHook`.
pub trait WriteHook<K: Key>: Send + Sync {
    const WATCHES: bool = false;

    fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
        None
    }

    /// Whether the watch carrier requests the previous value in published
    /// events. Kept separate from `NEEDS_OLD_VALUE`, which may be true only
    /// for an unrelated write hook.
    fn watch_old(&self) -> bool {
        false
    }

    const NEEDS_OLD_VALUE: bool = true;
    const NEEDS_INIT: bool = false;
    const NEEDS_WRITE: bool = true;
    const EXPIRES: bool = false;
    const TTL_REVISION: u32 = 0;

    fn deadline(&self, _key: &K, _value: &[u8]) -> Option<u64> {
        None
    }

    /// True when `value` bytes cannot be decoded for TTL stage-2 (Var families).
    /// Distinguishes decode errors from a `None` policy deadline.
    fn ttl_decode_failed(&self, _value: &[u8]) -> bool {
        false
    }

    /// Called after a successful write operation.
    ///
    /// - `key` — the affected key
    /// - `old` — previous value (`Some` on update/delete when `NEEDS_OLD_VALUE`, else `None`)
    /// - `new` — new value (`Some` on put/insert/update, `None` on delete)
    ///
    /// # Reentrancy
    ///
    /// The shard lock is NOT held during this call — implementations may
    /// safely read from or write to the same collection, including the
    /// same shard.
    ///
    /// Inside `atomic()`, `on_write` is buffered per mutation and replayed after the shard
    /// lock is released, in application order; on a closure `Err`, hooks still fire for
    /// the operations that were applied.
    fn on_write(&self, key: &K, old: Option<&[u8]>, new: Option<&[u8]>);

    /// Called once per live entry during collection open.
    /// Fires after recovery and migration are complete, with the final value.
    ///
    /// # Reentrancy
    ///
    /// Like `on_write`, the shard lock is NOT held during this call —
    /// implementations may safely read from or write to the same collection,
    /// including the same shard.
    fn on_init(&self, _key: &K, _value: &[u8]) {}

    #[cfg(feature = "typed-tree")]
    const PROJECTS: bool = false;

    #[cfg(feature = "typed-tree")]
    fn project(
        &self,
        _key: &K,
        _old: Option<&[u8]>,
        _new: Option<&[u8]>,
        _sink: &mut ProjectionSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    #[cfg(feature = "typed-tree")]
    fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
        None
    }

    #[cfg(feature = "typed-tree")]
    fn projection_limits(&self) -> (usize, usize) {
        (
            crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
            crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
        )
    }

    /// Whether this hook maintains engine counters. Compile-time, so an
    /// uncounted collection's write path is unchanged.
    #[cfg(feature = "typed-tree")]
    const COUNTS: bool = false;

    /// Byte-snapshot counter projection — mirrors [`WriteHook::project`]'s
    /// shape (key, old snapshot, new snapshot, sink).
    #[cfg(feature = "typed-tree")]
    fn counter_project(
        &self,
        _key: &K,
        _old: Option<&[u8]>,
        _new: Option<&[u8]>,
        _sink: &mut crate::counter::CounterSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    #[cfg(feature = "typed-tree")]
    fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
        None
    }

    /// Whether this hook declares claims into a unique namespace (T10).
    /// Compile-time, so an unclaimed collection's write path is unchanged.
    #[cfg(feature = "typed-tree")]
    const CLAIMS: bool = false;

    /// Byte-snapshot claim projection — mirrors [`WriteHook::project`]'s shape.
    #[cfg(feature = "typed-tree")]
    fn claim_project(
        &self,
        _key: &K,
        _old: Option<&[u8]>,
        _new: Option<&[u8]>,
        _sink: &mut crate::unique::ClaimSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    #[cfg(feature = "typed-tree")]
    fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
        None
    }

    #[cfg(all(feature = "typed-tree", feature = "armour"))]
    const EMITS: bool = false;

    #[cfg(all(feature = "typed-tree", feature = "armour"))]
    fn stream_project(
        &self,
        _key: &K,
        _old: Option<&[u8]>,
        _new: Option<&[u8]>,
        _sink: &mut crate::EmitSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    #[cfg(all(feature = "typed-tree", feature = "armour"))]
    fn stream_targets(&self) -> Option<&crate::QueueTargets> {
        None
    }
}

/// Default no-op hook. All branches are eliminated at compile time.
pub struct NoHook;

impl<K: Key> WriteHook<K> for NoHook {
    const NEEDS_OLD_VALUE: bool = false;
    const NEEDS_INIT: bool = false;
    const NEEDS_WRITE: bool = false;

    #[inline(always)]
    fn on_write(&self, _key: &K, _old: Option<&[u8]>, _new: Option<&[u8]>) {}

    #[inline(always)]
    fn on_init(&self, _key: &K, _value: &[u8]) {}

    #[cfg(feature = "typed-tree")]
    const PROJECTS: bool = false;

    #[cfg(feature = "typed-tree")]
    fn project(
        &self,
        _key: &K,
        _old: Option<&[u8]>,
        _new: Option<&[u8]>,
        _sink: &mut ProjectionSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    #[cfg(feature = "typed-tree")]
    const COUNTS: bool = false;

    #[cfg(feature = "typed-tree")]
    const CLAIMS: bool = false;

    #[cfg(all(feature = "typed-tree", feature = "armour"))]
    const EMITS: bool = false;
}

/// Typed write hook for [`TypedTree`](crate::TypedTree).
///
/// Receives `&T` directly instead of raw bytes — no encode/decode needed.
/// Called after the in-memory commit. The hook arguments carry the
/// mutation's (old, new) pair; a reentrant `get()` from the hook
/// observes the committed value (or a later concurrent write).
///
/// Inside `atomic()`, `on_write` is buffered per mutation and replayed after the shard
/// lock is released, in application order; on a closure `Err`, hooks still fire for
/// the operations that were applied.
///
/// **Exception — `LsmTree`** (feature `fjall`): inside `LsmTree::atomic` hooks fire
/// *synchronously, under the held stripe lock*, not buffered-and-replayed. A hook
/// that re-enters the same collection there will deadlock. Tracked in
/// libfunc/armour#119.
#[cfg(feature = "typed-tree")]
pub trait TypedWriteHook<K: Key, T>: Send + Sync {
    const WATCHES: bool = false;

    fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
        None
    }

    fn watch_old(&self) -> bool {
        false
    }

    /// Unused for TypedTree/TypedMap/ZeroTree/ZeroMap — old value is always
    /// provided (it lives in memory). Kept for trait uniformity with `WriteHook`.
    const NEEDS_OLD_VALUE: bool = true;
    const NEEDS_INIT: bool = false;
    const NEEDS_WRITE: bool = true;
    const EXPIRES: bool = false;
    const TTL_REVISION: u32 = 0;

    fn deadline(&self, _key: &K, _value: &T) -> Option<u64> {
        None
    }

    fn on_write(&self, key: &K, old: Option<&T>, new: Option<&T>);

    /// See [`WriteHook::on_init`] for reentrancy guarantees.
    fn on_init(&self, _key: &K, _value: &T) {}

    const PROJECTS: bool = false;

    fn project(&self, _key: &K, _value: &T, _sink: &mut ProjectionSink<'_>) -> DbResult<()> {
        Ok(())
    }

    fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
        None
    }

    fn projection_limits(&self) -> (usize, usize) {
        (
            crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
            crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
        )
    }

    /// Whether this hook maintains engine counters. Compile-time, so an
    /// uncounted collection's write path is unchanged.
    const COUNTS: bool = false;

    fn counter_project(
        &self,
        _key: &K,
        _value: &T,
        _sink: &mut crate::counter::CounterSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
        None
    }

    /// Whether this hook declares claims into a unique namespace (T10).
    /// Compile-time, so an unclaimed collection's write path is unchanged.
    const CLAIMS: bool = false;

    fn claim_project(
        &self,
        _key: &K,
        _value: &T,
        _sink: &mut crate::unique::ClaimSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
        None
    }

    #[cfg(feature = "armour")]
    const EMITS: bool = false;

    #[cfg(feature = "armour")]
    fn stream_project(
        &self,
        _key: &K,
        _old: Option<&T>,
        _new: Option<&T>,
        _sink: &mut crate::EmitSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    #[cfg(feature = "armour")]
    fn stream_targets(&self) -> Option<&crate::QueueTargets> {
        None
    }
}

#[cfg(feature = "typed-tree")]
impl<K: Key, T> TypedWriteHook<K, T> for NoHook {
    const NEEDS_OLD_VALUE: bool = false;
    const NEEDS_INIT: bool = false;
    const NEEDS_WRITE: bool = false;

    #[inline(always)]
    fn on_write(&self, _key: &K, _old: Option<&T>, _new: Option<&T>) {}

    #[inline(always)]
    fn on_init(&self, _key: &K, _value: &T) {}

    const PROJECTS: bool = false;

    fn project(&self, _key: &K, _value: &T, _sink: &mut ProjectionSink<'_>) -> DbResult<()> {
        Ok(())
    }

    const COUNTS: bool = false;

    const CLAIMS: bool = false;
}

/// Adapter: wraps a [`TypedWriteHook<K, T>`] and implements [`WriteHook<K>`]
/// by converting raw bytes to `&T` via zerocopy. Used by ZeroTree/ZeroMap
/// so their users get typed hook callbacks.
#[cfg(feature = "typed-tree")]
pub struct ZeroHookAdapter<K, T, H> {
    pub(crate) inner: H,
    pub(crate) _marker: std::marker::PhantomData<fn() -> (K, T)>,
}

#[cfg(feature = "typed-tree")]
impl<K, T, H> ZeroHookAdapter<K, T, H> {
    pub fn new(inner: H) -> Self {
        Self {
            inner,
            _marker: std::marker::PhantomData,
        }
    }
}

// SAFETY: The adapter delegates to H which is Send+Sync (via TypedWriteHook bound).
// PhantomData<fn() -> (K, T)> is always Send+Sync.
#[cfg(feature = "typed-tree")]
unsafe impl<K, T, H: Send> Send for ZeroHookAdapter<K, T, H> {}
#[cfg(feature = "typed-tree")]
unsafe impl<K, T, H: Sync> Sync for ZeroHookAdapter<K, T, H> {}

#[cfg(feature = "typed-tree")]
impl<K: Key, T: Copy + zerocopy::FromBytes, H: TypedWriteHook<K, T>> WriteHook<K>
    for ZeroHookAdapter<K, T, H>
{
    const WATCHES: bool = H::WATCHES;
    const NEEDS_OLD_VALUE: bool = H::NEEDS_OLD_VALUE;
    const NEEDS_INIT: bool = H::NEEDS_INIT;
    const NEEDS_WRITE: bool = H::NEEDS_WRITE;
    const EXPIRES: bool = H::EXPIRES;
    const TTL_REVISION: u32 = H::TTL_REVISION;

    fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
        self.inner.watch_channel()
    }
    fn watch_old(&self) -> bool {
        self.inner.watch_old()
    }
    const PROJECTS: bool = H::PROJECTS;
    const COUNTS: bool = H::COUNTS;
    const CLAIMS: bool = H::CLAIMS;
    #[cfg(feature = "armour")]
    const EMITS: bool = H::EMITS;

    fn deadline(&self, key: &K, value: &[u8]) -> Option<u64> {
        debug_assert_eq!(value.len(), size_of::<T>());
        let val: T = zerocopy::FromBytes::read_from_bytes(value).ok()?;
        self.inner.deadline(key, &val)
    }

    #[inline]
    fn on_write(&self, key: &K, old: Option<&[u8]>, new: Option<&[u8]>) {
        if !H::NEEDS_WRITE {
            return;
        }
        let read = |b: &[u8]| -> T {
            debug_assert_eq!(b.len(), size_of::<T>());
            zerocopy::FromBytes::read_from_bytes(b).expect("slice len == size_of::<T>()")
        };
        let old_val = old.map(read);
        let new_val = new.map(read);
        self.inner.on_write(key, old_val.as_ref(), new_val.as_ref());
    }

    #[inline]
    fn on_init(&self, key: &K, value: &[u8]) {
        debug_assert_eq!(value.len(), size_of::<T>());
        let val: T =
            zerocopy::FromBytes::read_from_bytes(value).expect("slice len == size_of::<T>()");
        self.inner.on_init(key, &val);
    }

    fn project(
        &self,
        key: &K,
        _old: Option<&[u8]>,
        new: Option<&[u8]>,
        sink: &mut ProjectionSink<'_>,
    ) -> DbResult<()> {
        if !H::PROJECTS {
            return Ok(());
        }
        if let Some(b) = new {
            debug_assert_eq!(b.len(), size_of::<T>());
            let val: T = zerocopy::FromBytes::read_from_bytes(b).map_err(|_| {
                DbError::FormatMismatch("zero hook adapter project decode failed".into())
            })?;
            self.inner.project(key, &val, sink)?;
        }
        Ok(())
    }

    fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
        self.inner.projection_targets()
    }

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

    fn counter_project(
        &self,
        key: &K,
        _old: Option<&[u8]>,
        new: Option<&[u8]>,
        sink: &mut crate::counter::CounterSink<'_>,
    ) -> DbResult<()> {
        if !H::COUNTS {
            return Ok(());
        }
        if let Some(b) = new {
            debug_assert_eq!(b.len(), size_of::<T>());
            let val: T = zerocopy::FromBytes::read_from_bytes(b).map_err(|_| {
                DbError::FormatMismatch("zero hook adapter counter_project decode failed".into())
            })?;
            self.inner.counter_project(key, &val, sink)?;
        }
        Ok(())
    }

    fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
        self.inner.counter_targets()
    }

    fn claim_project(
        &self,
        key: &K,
        _old: Option<&[u8]>,
        new: Option<&[u8]>,
        sink: &mut crate::unique::ClaimSink<'_>,
    ) -> DbResult<()> {
        if !H::CLAIMS {
            return Ok(());
        }
        if let Some(b) = new {
            debug_assert_eq!(b.len(), size_of::<T>());
            let val: T = zerocopy::FromBytes::read_from_bytes(b).map_err(|_| {
                DbError::FormatMismatch("zero hook adapter claim_project decode failed".into())
            })?;
            self.inner.claim_project(key, &val, sink)?;
        }
        Ok(())
    }

    fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
        self.inner.claim_targets()
    }

    #[cfg(feature = "armour")]
    fn stream_project(
        &self,
        key: &K,
        old: Option<&[u8]>,
        new: Option<&[u8]>,
        sink: &mut crate::EmitSink<'_>,
    ) -> DbResult<()> {
        if !H::EMITS {
            return Ok(());
        }
        let read = |b: &[u8]| -> Result<T, DbError> {
            zerocopy::FromBytes::read_from_bytes(b).map_err(|_| {
                DbError::FormatMismatch("zero hook adapter stream decode failed".into())
            })
        };
        let old = old.map(read).transpose()?;
        let new = new.map(read).transpose()?;
        self.inner
            .stream_project(key, old.as_ref(), new.as_ref(), sink)
    }

    #[cfg(feature = "armour")]
    fn stream_targets(&self) -> Option<&crate::QueueTargets> {
        self.inner.stream_targets()
    }
}

/// Adapter: wraps a [`TypedWriteHook<K, T>`] and implements [`WriteHook<K>`]
/// by decoding raw bytes to `T` via a [`Codec`](crate::codec::Codec).
/// Used by `VarTypedTree` / `VarTypedMap` so their users get typed hook callbacks.
///
/// Decode errors are logged at `warn` level. `on_write` passes `None` for the
/// failed side; `on_init` skips the entry entirely.
///
/// When `H::NEEDS_WRITE` is `false` (e.g. `NoHook`), `on_write` is a no-op —
/// no decode is performed.
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
pub struct VarTypedHookAdapter<K, T, C, H> {
    pub(crate) inner: H,
    pub(crate) codec: C,
    pub(crate) _marker: std::marker::PhantomData<fn() -> (K, T)>,
}

#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
impl<K, T, C, H> VarTypedHookAdapter<K, T, C, H> {
    pub fn new(inner: H, codec: C) -> Self {
        Self {
            inner,
            codec,
            _marker: std::marker::PhantomData,
        }
    }
}

#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
impl<K, T, C, H> VarTypedHookAdapter<K, T, C, H>
where
    K: Key,
    T: Send + Sync,
    H: TypedWriteHook<K, T>,
{
    pub(crate) fn typed_deadline(&self, key: &K, value: &T) -> Option<u64> {
        if H::EXPIRES {
            self.inner.deadline(key, value)
        } else {
            None
        }
    }
}

// SAFETY: delegates to H (Send+Sync via TypedWriteHook) and C (Send+Sync via Codec).
// PhantomData<fn() -> (K, T)> is always Send+Sync.
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
unsafe impl<K, T, C: Send, H: Send> Send for VarTypedHookAdapter<K, T, C, H> {}
#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
unsafe impl<K, T, C: Sync, H: Sync> Sync for VarTypedHookAdapter<K, T, C, H> {}

#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
impl<K, T, C, H> WriteHook<K> for VarTypedHookAdapter<K, T, C, H>
where
    K: Key,
    T: Send + Sync,
    C: crate::codec::Codec<T>,
    H: TypedWriteHook<K, T>,
{
    const WATCHES: bool = H::WATCHES;
    const NEEDS_OLD_VALUE: bool = H::NEEDS_OLD_VALUE;
    const NEEDS_INIT: bool = H::NEEDS_INIT;
    // The raw Var family consults `watch_channel().has_subscribers()` at the
    // mutation linearization point.  Do not make a dormant watch force it to
    // retain disk bytes or allocate an owned event.
    const NEEDS_WRITE: bool = H::NEEDS_WRITE;
    const EXPIRES: bool = H::EXPIRES;
    const TTL_REVISION: u32 = H::TTL_REVISION;

    fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
        self.inner.watch_channel()
    }
    fn watch_old(&self) -> bool {
        self.inner.watch_old()
    }
    const PROJECTS: bool = H::PROJECTS;
    const COUNTS: bool = H::COUNTS;
    const CLAIMS: bool = H::CLAIMS;
    #[cfg(feature = "armour")]
    const EMITS: bool = H::EMITS;

    fn deadline(&self, key: &K, value: &[u8]) -> Option<u64> {
        self.codec
            .decode_from(value)
            .ok()
            .and_then(|val| self.inner.deadline(key, &val))
    }

    fn ttl_decode_failed(&self, value: &[u8]) -> bool {
        self.codec.decode_from(value).is_err()
    }

    #[inline]
    fn on_write(&self, key: &K, old: Option<&[u8]>, new: Option<&[u8]>) {
        if !H::NEEDS_WRITE {
            return;
        }
        let old_t = old.and_then(|b| {
            self.codec
                .decode_from(b)
                .map_err(|_| {
                    tracing::warn!(
                        side = "old",
                        value_len = b.len(),
                        "var_typed hook adapter: on_write decode failed"
                    );
                })
                .ok()
        });
        let new_t = new.and_then(|b| {
            self.codec
                .decode_from(b)
                .map_err(|_| {
                    tracing::warn!(
                        side = "new",
                        value_len = b.len(),
                        "var_typed hook adapter: on_write decode failed"
                    );
                })
                .ok()
        });
        self.inner.on_write(key, old_t.as_ref(), new_t.as_ref());
    }

    #[inline]
    fn on_init(&self, key: &K, value: &[u8]) {
        match self.codec.decode_from(value) {
            Ok(val) => self.inner.on_init(key, &val),
            Err(_) => {
                tracing::warn!(
                    key_len = std::mem::size_of::<K>(),
                    value_len = value.len(),
                    "var_typed hook adapter: on_init decode failed, entry skipped"
                );
            }
        }
    }

    fn project(
        &self,
        key: &K,
        _old: Option<&[u8]>,
        new: Option<&[u8]>,
        sink: &mut ProjectionSink<'_>,
    ) -> DbResult<()> {
        if !H::PROJECTS {
            return Ok(());
        }
        if let Some(b) = new {
            let val = self.codec.decode_from(b)?;
            self.inner.project(key, &val, sink)?;
        }
        Ok(())
    }

    fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
        self.inner.projection_targets()
    }

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

    fn counter_project(
        &self,
        key: &K,
        _old: Option<&[u8]>,
        new: Option<&[u8]>,
        sink: &mut crate::counter::CounterSink<'_>,
    ) -> DbResult<()> {
        if !H::COUNTS {
            return Ok(());
        }
        if let Some(b) = new {
            let val = self.codec.decode_from(b)?;
            self.inner.counter_project(key, &val, sink)?;
        }
        Ok(())
    }

    fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
        self.inner.counter_targets()
    }

    fn claim_project(
        &self,
        key: &K,
        _old: Option<&[u8]>,
        new: Option<&[u8]>,
        sink: &mut crate::unique::ClaimSink<'_>,
    ) -> DbResult<()> {
        if !H::CLAIMS {
            return Ok(());
        }
        if let Some(b) = new {
            let val = self.codec.decode_from(b)?;
            self.inner.claim_project(key, &val, sink)?;
        }
        Ok(())
    }

    fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
        self.inner.claim_targets()
    }

    #[cfg(feature = "armour")]
    fn stream_project(
        &self,
        key: &K,
        old: Option<&[u8]>,
        new: Option<&[u8]>,
        sink: &mut crate::EmitSink<'_>,
    ) -> DbResult<()> {
        if !H::EMITS {
            return Ok(());
        }
        let old = old.map(|bytes| self.codec.decode_from(bytes)).transpose()?;
        let new = new.map(|bytes| self.codec.decode_from(bytes)).transpose()?;
        self.inner
            .stream_project(key, old.as_ref(), new.as_ref(), sink)
    }

    #[cfg(feature = "armour")]
    fn stream_targets(&self) -> Option<&crate::QueueTargets> {
        self.inner.stream_targets()
    }
}

#[cfg(all(test, feature = "typed-tree", feature = "armour"))]
mod tests {
    use super::*;
    use crate::projection::Projected;
    use std::sync::Arc;

    /// T9 Task 5 forwarding fix: a wrapper hook must forward `COUNTS` from its
    /// inner `TypedWriteHook`, exactly like it already forwards `PROJECTS`.
    /// Every Zero-family collection's hook type is a `ZeroHookAdapter` — a
    /// missing forward would silently fall back to the trait's `false`
    /// default and counters would never fire, with nothing failing to
    /// compile. `COUNTS` is `true` on any `Projected`, regardless of whether
    /// `.and_counters` was called (mirrors `PROJECTS`, which is `true`
    /// regardless of whether `IndexTargets` is empty) — no instance
    /// construction is needed to observe the constant.
    #[test]
    fn zero_hook_adapter_forwards_counts() {
        type Inner = Projected<[u8; 8], u64>;
        assert!(<ZeroHookAdapter<[u8; 8], u64, Inner> as WriteHook<[u8; 8]>>::COUNTS);
    }

    /// Same forwarding proof for `VarTypedHookAdapter` (Var-typed families).
    #[cfg(all(feature = "var-collections", feature = "rapira-codec"))]
    #[test]
    fn var_typed_hook_adapter_forwards_counts() {
        type Inner = Projected<[u8; 8], u64>;
        assert!(
            <VarTypedHookAdapter<[u8; 8], u64, crate::codec::RapiraCodec, Inner> as WriteHook<
                [u8; 8],
            >>::COUNTS
        );
    }

    /// T10 Task 3 twin of `zero_hook_adapter_forwards_counts`: a wrapper hook
    /// must forward `CLAIMS` from its inner `TypedWriteHook` exactly like it
    /// already forwards `PROJECTS`/`COUNTS`. `CLAIMS` is `true` on any
    /// `Projected`, regardless of whether `.and_unique` was called (mirrors
    /// `COUNTS`) — no instance construction is needed to observe the constant.
    #[test]
    fn zero_hook_adapter_forwards_claims() {
        type Inner = Projected<[u8; 8], u64>;
        assert!(<ZeroHookAdapter<[u8; 8], u64, Inner> as WriteHook<[u8; 8]>>::CLAIMS);
    }

    /// Same forwarding proof for `VarTypedHookAdapter` (Var-typed families).
    #[cfg(all(feature = "var-collections", feature = "rapira-codec"))]
    #[test]
    fn var_typed_hook_adapter_forwards_claims() {
        type Inner = Projected<[u8; 8], u64>;
        assert!(
            <VarTypedHookAdapter<[u8; 8], u64, crate::codec::RapiraCodec, Inner> as WriteHook<
                [u8; 8],
            >>::CLAIMS
        );
    }

    #[derive(Clone)]
    struct SnapshotHook(Arc<std::sync::Mutex<Vec<(Option<u64>, Option<u64>)>>>);

    impl TypedWriteHook<[u8; 8], u64> for SnapshotHook {
        const EMITS: bool = true;

        fn on_write(&self, _: &[u8; 8], _: Option<&u64>, _: Option<&u64>) {}

        fn stream_project(
            &self,
            _: &[u8; 8],
            old: Option<&u64>,
            new: Option<&u64>,
            _: &mut crate::EmitSink<'_>,
        ) -> DbResult<()> {
            self.0.lock().unwrap().push((old.copied(), new.copied()));
            Ok(())
        }
    }

    #[test]
    fn zero_hook_adapter_decodes_and_forwards_each_stream_snapshot_once() {
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let hook = ZeroHookAdapter::new(SnapshotHook(Arc::clone(&seen)));
        let targets = crate::QueueTargets::new();
        let mut sink = crate::EmitSink::new(&targets, 1, 1);
        let old = 7u64.to_ne_bytes();
        let new = 9u64.to_ne_bytes();
        hook.stream_project(&[0; 8], Some(&old), Some(&new), &mut sink)
            .unwrap();
        assert_eq!(*seen.lock().unwrap(), vec![(Some(7), Some(9))]);
    }

    #[cfg(all(feature = "var-collections", feature = "rapira-codec"))]
    #[test]
    fn var_hook_adapter_decodes_and_forwards_each_stream_snapshot_once() {
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let hook =
            VarTypedHookAdapter::new(SnapshotHook(Arc::clone(&seen)), crate::codec::RapiraCodec);
        let targets = crate::QueueTargets::new();
        let mut sink = crate::EmitSink::new(&targets, 1, 1);
        let old = rapira::serialize(&7u64);
        let new = rapira::serialize(&9u64);
        hook.stream_project(&[0; 8], Some(&old), Some(&new), &mut sink)
            .unwrap();
        assert_eq!(*seen.lock().unwrap(), vec![(Some(7), Some(9))]);
    }

    /// T10 Task 3 review fix F5: the `CLAIMS` **constant** alone cannot tell
    /// "forwards the inner hook's claims" apart from "reports `CLAIMS = true`
    /// but keeps the trait-default `claim_project` (`Ok(())`, silently
    /// swallowing every claim)" — both pass `zero_hook_adapter_forwards_claims`
    /// above. This drives a real claim through `ZeroHookAdapter::claim_project`
    /// and checks it actually reached the sink; reverting the adapter's
    /// `claim_project` body to the trait default makes this fail with 0
    /// claims instead of 1.
    #[cfg(feature = "armour")]
    mod claim_forwarding {
        use std::sync::Arc;

        use crate::projection::IndexTargets;
        use crate::unique::{ClaimSink, UniqueTargets};
        use crate::unique_tests::{ClaimFwdKey, ClaimFwdRow, open_claim_ns};

        use super::*;

        #[test]
        fn zero_hook_adapter_forwards_a_claim_end_to_end() {
            let (_dir, _db, ns) = open_claim_ns();
            let targets = UniqueTargets::new().with(&ns);
            let ns_for_project = Arc::clone(&ns);
            let inner =
                Projected::<ClaimFwdKey, ClaimFwdRow>::new(IndexTargets::new(), |_, _, _| Ok(()))
                    .and_unique(targets.clone(), move |_k, _v: &ClaimFwdRow, sink| {
                        sink.claim(&ns_for_project, ClaimFwdKey::new(7))
                    });
            let adapter = ZeroHookAdapter::new(inner);
            let bytes = zerocopy::IntoBytes::as_bytes(&ClaimFwdRow(1)).to_vec();
            let mut sink = ClaimSink::new(&targets, 16, 1024);
            adapter
                .claim_project(&ClaimFwdKey::new(0), None, Some(&bytes), &mut sink)
                .expect("claim_project");
            assert_eq!(
                sink.claims().len(),
                1,
                "the inner Projected's claim must reach the adapter's caller"
            );
        }

        /// Same proof for `VarTypedHookAdapter` (Var-typed families).
        #[cfg(all(feature = "var-collections", feature = "rapira-codec"))]
        #[test]
        fn var_typed_hook_adapter_forwards_a_claim_end_to_end() {
            let (_dir, _db, ns) = open_claim_ns();
            let targets = UniqueTargets::new().with(&ns);
            let ns_for_project = Arc::clone(&ns);
            let inner =
                Projected::<ClaimFwdKey, ClaimFwdRow>::new(IndexTargets::new(), |_, _, _| Ok(()))
                    .and_unique(targets.clone(), move |_k, _v: &ClaimFwdRow, sink| {
                        sink.claim(&ns_for_project, ClaimFwdKey::new(7))
                    });
            let adapter = VarTypedHookAdapter::new(inner, crate::RapiraCodec::default());
            let bytes = rapira::serialize(&ClaimFwdRow(1));
            let mut sink = ClaimSink::new(&targets, 16, 1024);
            adapter
                .claim_project(&ClaimFwdKey::new(0), None, Some(&bytes), &mut sink)
                .expect("claim_project");
            assert_eq!(
                sink.claims().len(),
                1,
                "the inner Projected's claim must reach the adapter's caller"
            );
        }
    }
}