heaplet 0.1.0

//! In-memory store and internal plumbing.
//!
//! [`Store`] keeps a single key-space (`HashMap<String, Entry>`) behind an `RwLock`,
//! and provides typed facades for Redis-inspired data structures.
//!
//! # Key behaviors
//! - **Lazy expiration**: TTL is enforced on access (see key ops like `exists`, `ttl`, `get`).
//! - **List blocking ops**: `List` uses an inner `Mutex + Condvar` so blocking pops don't hold
//!   the global map lock.
//!
//! Most users will start from [`Store`] and call `kv()`, `keys()`, `hash()`, `set()`,
//! `list()`, `deque()`, `zset()`.

use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::sync::{Arc, Condvar, Mutex, RwLock};
use std::time::Instant;

use crate::codec::{BincodeCodec, Codec};
use crate::entry::{
    DequeEntry, Entry, HashEntry, ListEntry, ListInner, Meta, SetEntry, StringEntry, ValueType,
    ZSetEntry,
};
use crate::error::Error;

/// An in-memory Redis-inspired store.
///
/// Values live in a single key-space (`HashMap<String, Entry>`) guarded by an [`RwLock`].
/// Higher-level typed APIs (kv/hash/set/list/deque/zset/keys) are thin facades over internal helpers.
///
/// # Design notes
/// - **Lazy expiration**: TTL is checked on access via `purge_if_expired` (internal helper). There is no background
///   sweeper thread.
/// - **Blocking lists**: `List` uses an inner `Mutex + Condvar` so blocking pops do not hold the
///   global map write-lock.
/// - **Typed ergonomic APIs**: all data structures store encoded bytes via the store's [`Codec`],
///   with typed wrappers that encode/decode for you.
///
/// # Example
///
/// ```rust
/// use heaplet::Store;
///
/// let store = Store::new();
///
/// // String / KV
/// store.kv().set("k", &123_i64).unwrap();
/// let v: Option<i64> = store.kv().get("k").unwrap();
/// assert_eq!(v, Some(123));
///
/// // Hash
/// let h = store.hash("h");
/// h.hset("field", &"value").unwrap();
/// let hv: Option<String> = h.hget("field").unwrap();
/// assert_eq!(hv.as_deref(), Some("value"));
/// ```
pub struct Store<C: Codec = BincodeCodec> {
    codec: C,
    map: RwLock<HashMap<String, Entry>>,
}

impl Default for Store<BincodeCodec> {
    fn default() -> Self {
        Self::new()
    }
}

impl Store<BincodeCodec> {
    /// Create a store with the default [`BincodeCodec`].
    pub fn new() -> Self {
        Self::with_codec(BincodeCodec)
    }
}

impl<C: Codec> Store<C> {
    /// Create a store with a custom codec.
    pub fn with_codec(codec: C) -> Self {
        Self {
            codec,
            map: RwLock::new(HashMap::new()),
        }
    }

    /// Return the codec used by this store.
    pub fn codec(&self) -> &C {
        &self.codec
    }

    /// KV operations facade (string values).
    pub fn kv(&self) -> crate::kv::KvOps<'_, C> {
        crate::kv::KvOps::new(self)
    }

    /// Key-space operations (exists/rename/expire/scan).
    pub fn keys(&self) -> crate::keys::KeysOps<'_, C> {
        crate::keys::KeysOps::new(self)
    }

    /// Hash reference bound to `key`.
    pub fn hash<'a>(&'a self, key: &'a str) -> crate::hash::HashRef<'a, C> {
        crate::hash::HashRef::new(self, key)
    }

    /// Set reference bound to `key`.
    pub fn set<'a>(&'a self, key: &'a str) -> crate::set::SetRef<'a, C> {
        crate::set::SetRef::new(self, key)
    }

    /// List reference bound to `key`.
    pub fn list<'a>(&'a self, key: &'a str) -> crate::list::ListRef<'a, C> {
        crate::list::ListRef::new(self, key)
    }

    /// Deque reference bound to `key`.
    pub fn deque<'a>(&'a self, key: &'a str) -> crate::deque::DequeRef<'a, C> {
        crate::deque::DequeRef::new(self, key)
    }

    /// Sorted-set reference bound to `key`.
    pub fn zset<'a>(&'a self, key: &'a str) -> crate::zset::ZSetRef<'a, C> {
        crate::zset::ZSetRef::new(self, key)
    }

    // -------- internal helpers --------

    /// Read-only access to the raw key-space map.
    pub(crate) fn with_map_read<R>(&self, f: impl FnOnce(&HashMap<String, Entry>) -> R) -> R {
        let guard = self.map.read().expect("store lock poisoned");
        f(&guard)
    }

    /// Mutable access to the raw key-space map.
    pub(crate) fn with_map_write<R>(&self, f: impl FnOnce(&mut HashMap<String, Entry>) -> R) -> R {
        let mut guard = self.map.write().expect("store lock poisoned");
        f(&mut guard)
    }

    /// Returns whether the key exists (no TTL purge here).
    pub(crate) fn contains_key(&self, key: &str) -> bool {
        self.with_map_read(|m| m.contains_key(key))
    }

    /// Snapshot all current keys (no TTL purge here).
    pub(crate) fn snapshot_keys(&self) -> Vec<String> {
        self.with_map_read(|m| m.keys().cloned().collect())
    }

    /// Get a cloned entry (no TTL purge here).
    pub(crate) fn get_entry(&self, key: &str) -> Option<Entry> {
        self.with_map_read(|m| m.get(key).cloned())
    }

    /// Get entry meta (no TTL purge here).
    pub(crate) fn get_entry_meta(&self, key: &str) -> Option<Meta> {
        self.with_map_read(|m| m.get(key).map(|e| *e.meta()))
    }

    /// Overwrite `key` with a raw string entry.
    pub(crate) fn put_string_entry(&self, key: &str, bytes: crate::codec::Bytes) {
        self.with_map_write(|m| {
            let entry = Entry::String(StringEntry {
                meta: Meta::new(ValueType::String),
                bytes,
            });
            m.insert(key.to_string(), entry);
        });
    }

    /// Remove a key. Returns `true` if it existed.
    pub(crate) fn remove_entry(&self, key: &str) -> bool {
        self.with_map_write(|m| m.remove(key).is_some())
    }

    /// Set `expire_at` for an existing key.
    ///
    /// Returns `false` if the key does not exist.
    pub(crate) fn set_expire_at(&self, key: &str, when: Option<Instant>) -> bool {
        self.with_map_write(|m| {
            let Some(entry) = m.get_mut(key) else {
                return false;
            };
            entry.meta_mut().expire_at = when;
            true
        })
    }

    /// Purge the key if it is expired **at the time of access** (lazy TTL).
    ///
    /// MVP semantics:
    /// - We do a quick read check first.
    /// - If it looks expired, we take the write lock and re-check before removing.
    ///
    /// This avoids deleting a key that was refreshed between the two phases.
    pub(crate) fn purge_if_expired(&self, key: &str) {
        // Phase 1: read check.
        let expired = self.with_map_read(|m| {
            let Some(entry) = m.get(key) else {
                return false;
            };
            match entry.meta().expire_at {
                None => false,
                Some(t) => t <= Instant::now(),
            }
        });

        if !expired {
            return;
        }

        // Phase 2: write lock + re-check.
        self.with_map_write(|m| {
            let now = Instant::now();
            let should_remove = match m.get(key) {
                None => false,
                Some(entry) => match entry.meta().expire_at {
                    None => false,
                    Some(t) => t <= now,
                },
            };

            if should_remove {
                m.remove(key);
            }
        });
    }

    /// Rename a key.
    ///
    /// - If `nx` is `true`, the destination must not exist.
    /// - Returns `Ok(false)` when `nx` prevents overwriting.
    pub(crate) fn rename_internal(&self, from: &str, to: &str, nx: bool) -> Result<bool, Error> {
        if from == to {
            return Ok(true);
        }

        self.with_map_write(|m| {
            let Some(entry) = m.remove(from) else {
                return Err(Error::NotFound);
            };

            if nx && m.contains_key(to) {
                // Put it back so the operation behaves atomically for callers.
                m.insert(from.to_string(), entry);
                return Ok(false);
            }

            m.insert(to.to_string(), entry);
            Ok(true)
        })
    }

    // --- hash ---

    /// Ensure `key` is a hash and pass its inner map to `f` (create-on-demand).
    pub(crate) fn with_hash_mut<R>(
        &self,
        key: &str,
        f: impl FnOnce(&mut HashMap<crate::codec::Bytes, crate::codec::Bytes>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        self.with_map_write(|m| {
            if !m.contains_key(key) {
                m.insert(
                    key.to_string(),
                    Entry::Hash(HashEntry {
                        meta: Meta::new(ValueType::Hash),
                        map: HashMap::new(),
                    }),
                );
            }

            match m.get_mut(key).expect("just inserted or existed") {
                Entry::Hash(he) => f(&mut he.map),
                other => Err(Error::WrongType {
                    expected: ValueType::Hash.as_str(),
                    got: other.value_type().as_str(),
                }),
            }
        })
    }

    /// Read-only hash access (missing => `None`).
    pub(crate) fn with_hash_read<R>(
        &self,
        key: &str,
        f: impl FnOnce(Option<&HashMap<crate::codec::Bytes, crate::codec::Bytes>>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        self.with_map_read(|m| match m.get(key) {
            None => f(None),
            Some(entry) => match entry {
                Entry::Hash(he) => f(Some(&he.map)),
                other => Err(Error::WrongType {
                    expected: ValueType::Hash.as_str(),
                    got: other.value_type().as_str(),
                }),
            },
        })
    }

    // --- set ---

    /// Ensure `key` is a set and pass it to `f` (create-on-demand).
    pub(crate) fn with_set_mut<R>(
        &self,
        key: &str,
        f: impl FnOnce(&mut HashSet<crate::codec::Bytes>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        self.with_map_write(|m| {
            if !m.contains_key(key) {
                m.insert(
                    key.to_string(),
                    Entry::Set(SetEntry {
                        meta: Meta::new(ValueType::Set),
                        set: HashSet::new(),
                    }),
                );
            }

            match m.get_mut(key).expect("just inserted or existed") {
                Entry::Set(se) => f(&mut se.set),
                other => Err(Error::WrongType {
                    expected: ValueType::Set.as_str(),
                    got: other.value_type().as_str(),
                }),
            }
        })
    }

    /// Read-only set access (missing => `None`).
    pub(crate) fn with_set_read<R>(
        &self,
        key: &str,
        f: impl FnOnce(Option<&HashSet<crate::codec::Bytes>>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        self.with_map_read(|m| match m.get(key) {
            None => f(None),
            Some(entry) => match entry {
                Entry::Set(se) => f(Some(&se.set)),
                other => Err(Error::WrongType {
                    expected: ValueType::Set.as_str(),
                    got: other.value_type().as_str(),
                }),
            },
        })
    }

    // --- list ---

    /// Ensure `key` is a list and pass its deque to `f` (create-on-demand).
    ///
    /// List data lives behind an inner `Mutex`, so the global map lock is not held while
    /// user code runs in `f`.
    pub(crate) fn with_list_mut<R>(
        &self,
        key: &str,
        f: impl FnOnce(&mut VecDeque<crate::codec::Bytes>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        let inner = self.with_map_write(|m| {
            match m.get(key) {
                None => {
                    let entry = Entry::List(ListEntry {
                        meta: Meta::new(ValueType::List),
                        inner: Arc::new((
                            Mutex::new(ListInner {
                                deque: VecDeque::new(),
                            }),
                            Condvar::new(),
                        )),
                    });
                    m.insert(key.to_string(), entry);
                }
                Some(e) if e.value_type() != ValueType::List => {
                    return Err(Error::WrongType {
                        expected: ValueType::List.as_str(),
                        got: e.value_type().as_str(),
                    });
                }
                Some(_) => {}
            }

            match m.get(key).expect("ensured above") {
                Entry::List(le) => Ok(le.inner.clone()),
                other => Err(Error::WrongType {
                    expected: ValueType::List.as_str(),
                    got: other.value_type().as_str(),
                }),
            }
        })?;

        let (lock, _cv) = &*inner;
        let mut guard = lock.lock().expect("list mutex poisoned");
        f(&mut guard.deque)
    }

    /// Read-only list access (missing => `None`).
    pub(crate) fn with_list_read<R>(
        &self,
        key: &str,
        f: impl FnOnce(Option<&VecDeque<crate::codec::Bytes>>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        let inner = self.with_map_read(|m| match m.get(key) {
            None => Ok::<_, Error>(None),
            Some(entry) => match entry {
                Entry::List(le) => Ok(Some(le.inner.clone())),
                other => Err(Error::WrongType {
                    expected: ValueType::List.as_str(),
                    got: other.value_type().as_str(),
                }),
            },
        })?;

        let Some(inner) = inner else {
            return f(None);
        };

        let (lock, _cv) = &*inner;
        let guard = lock.lock().expect("list mutex poisoned");
        f(Some(&guard.deque))
    }

    /// Notify waiters on the list's condition variable (no-op if missing).
    pub(crate) fn list_notify(&self, key: &str) -> Result<(), Error> {
        self.purge_if_expired(key);

        let inner = self.with_map_read(|m| match m.get(key) {
            None => Ok::<_, Error>(None),
            Some(entry) => match entry {
                Entry::List(le) => Ok(Some(le.inner.clone())),
                other => Err(Error::WrongType {
                    expected: ValueType::List.as_str(),
                    got: other.value_type().as_str(),
                }),
            },
        })?;

        if let Some(inner) = inner {
            let (_lock, cv) = &*inner;
            cv.notify_all();
        }
        Ok(())
    }

    // --- deque ---

    /// Ensure `key` is a deque and pass its deque to `f` (create-on-demand).
    pub(crate) fn with_deque_mut<R>(
        &self,
        key: &str,
        f: impl FnOnce(&mut VecDeque<crate::codec::Bytes>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        self.with_map_write(|m| {
            let entry = m.entry(key.to_string()).or_insert_with(|| {
                Entry::Deque(DequeEntry {
                    meta: Meta::new(ValueType::Deque),
                    deque: VecDeque::new(),
                })
            });

            match entry {
                Entry::Deque(de) => f(&mut de.deque),
                other => Err(Error::WrongType {
                    expected: ValueType::Deque.as_str(),
                    got: other.value_type().as_str(),
                }),
            }
        })
    }

    /// Read-only deque access (missing => `None`).
    pub(crate) fn with_deque_read<R>(
        &self,
        key: &str,
        f: impl FnOnce(Option<&VecDeque<crate::codec::Bytes>>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        self.with_map_read(|m| match m.get(key) {
            None => f(None),
            Some(entry) => match entry {
                Entry::Deque(de) => f(Some(&de.deque)),
                other => Err(Error::WrongType {
                    expected: ValueType::Deque.as_str(),
                    got: other.value_type().as_str(),
                }),
            },
        })
    }

    // --- zset ---

    /// Ensure `key` is a zset and pass it to `f` (create-on-demand).
    pub(crate) fn with_zset_mut<R>(
        &self,
        key: &str,
        f: impl FnOnce(&mut ZSetEntry) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        self.with_map_write(|m| {
            let entry = m.entry(key.to_string()).or_insert_with(|| {
                Entry::ZSet(ZSetEntry {
                    meta: Meta::new(ValueType::ZSet),
                    member_to_score: HashMap::new(),
                    score_to_members: BTreeMap::new(),
                })
            });

            match entry {
                Entry::ZSet(ze) => f(ze),
                other => Err(Error::WrongType {
                    expected: ValueType::ZSet.as_str(),
                    got: other.value_type().as_str(),
                }),
            }
        })
    }

    /// Read-only zset access (missing => `None`).
    pub(crate) fn with_zset_read<R>(
        &self,
        key: &str,
        f: impl FnOnce(Option<&ZSetEntry>) -> Result<R, Error>,
    ) -> Result<R, Error> {
        self.purge_if_expired(key);

        self.with_map_read(|m| match m.get(key) {
            None => f(None),
            Some(entry) => match entry {
                Entry::ZSet(ze) => f(Some(ze)),
                other => Err(Error::WrongType {
                    expected: ValueType::ZSet.as_str(),
                    got: other.value_type().as_str(),
                }),
            },
        })
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use crate::Store;

    #[test]
    fn rename_preserves_structure() {
        let store = Store::new();
        let h = store.hash("old");

        h.hset("a", &1_i64).unwrap();
        store.keys().rename("old", "new").unwrap();

        let h2 = store.hash("new");
        let v: Option<i64> = h2.hget("a").unwrap();
        assert_eq!(v, Some(1));
    }

    #[test]
    fn ttl_applies_to_hash() {
        let store = Store::new();
        let h = store.hash("h");

        h.hset("a", &1_i64).unwrap();
        store.keys().pexpire("h", 1);

        std::thread::sleep(Duration::from_millis(3));

        let v: Option<i64> = h.hget("a").unwrap();
        assert_eq!(v, None);
        assert!(!store.keys().exists("h"));
    }
}