armdb 0.1.10

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

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, ZeroHookAdapter};
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`/`update`.
/// Does **not** fire inside `atomic()`. 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.
///
/// No ordered iteration — use [`ZeroTree`](crate::ZeroTree) if you need 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::default(),
/// )?;
/// map.put(&key, &Session { user_id: 42, expires: 0 })?;
/// if let Some(s) = map.get(&key) {
///     println!("user: {}", s.user_id);
/// }
/// ```
pub struct ZeroMap<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy = [u8; V],
    H: TypedWriteHook<K, T> = NoHook,
    D: Durability = Bitcask,
> {
    inner: ConstMap<K, V, ZeroHookAdapter<K, T, H>, D>,
    _marker: PhantomData<T>,
}

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

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T: Copy> ZeroMap<K, V, T, NoHook, Bitcask> {
    /// Open or create a `ZeroMap` 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: ConstMap::open_hooked(path, config, adapter)?,
            _marker: PhantomData,
        })
    }
}

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T: Copy, H: TypedWriteHook<K, T>>
    ZeroMap<K, V, T, H, Bitcask>
{
    /// Open or create a `ZeroMap` 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: ConstMap::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()
    }

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

    /// 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> 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.
    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: ConstMap::open_with_hook(path, config, adapter)?,
            _marker: PhantomData,
        })
    }
}

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T: Copy, H: TypedWriteHook<K, T>>
    ZeroMap<K, V, T, H, Fixed>
{
    /// Open or create a `ZeroMap` 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: ConstMap::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 + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy,
    H: TypedWriteHook<K, T>,
    D: Durability,
> ZeroMap<K, V, T, H, D>
{
    // -- Reads ----------------------------------------------------------------

    /// Get a value by key. O(1) lookup, zero disk I/O. Returns a copy of `T`.
    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)
    }

    /// 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 ZeroMapShard<'_, K, V, T, D>) -> DbResult<R>,
    ) -> DbResult<R> {
        self.inner.atomic(shard_key, |const_shard| {
            // SAFETY: ZeroMapShard is a transparent wrapper over ConstMapShard.
            // The hook type parameter doesn't affect ConstMapShard layout
            // (hook is stored in the parent ConstMap, not in the shard).
            let shard = unsafe {
                &mut *(const_shard as *mut ConstMapShard<'_, K, V, ZeroHookAdapter<K, T, H>, D>
                    as *mut ZeroMapShard<'_, K, V, T, D>)
            };
            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()
    }
}

impl<
    K: Key + Send + Sync + Hash + Eq,
    const V: usize,
    T: Copy + 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)
    }
}

// ---------------------------------------------------------------------------
// 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 = [u8; V],
    D: Durability = Bitcask,
> {
    // The actual hook type is ZeroHookAdapter, but ZeroMapShard is accessed via
    // unsafe pointer cast in atomic(). ConstMapShard layout doesn't depend on H
    // (hook is stored in the parent ConstMap, not in the shard).
    inner: ConstMapShard<'a, K, V, NoHook, D>,
    _marker: PhantomData<T>,
}

impl<K: Key + Send + Sync + Hash + Eq, const V: usize, T: Copy, D: Durability>
    ZeroMapShard<'_, K, V, T, D>
{
    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)
    }
}

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