armdb 0.9.0

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

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

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

#[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(())
    }
}

/// 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 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;
    const PROJECTS: bool = H::PROJECTS;

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

/// 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 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;
    const PROJECTS: bool = H::PROJECTS;

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