heaplet 0.1.0

//! Key-space operations (lifecycle, TTL, renaming, and key enumeration).
//!
//! This module provides Redis-like key APIs via [`KeysOps`]:
//! - existence & deletion (`exists`, `del`)
//! - TTL management (`expire`, `pexpire`, `ttl`, `pttl`, `persist`)
//! - rename (`rename`, `renamenx`)
//! - enumeration (`scan`, `iter_keys`)
//!
//! Notes:
//! - Expiration is checked lazily on access.
//! - `scan`-style APIs are snapshot-based for deterministic paging (MVP semantics).

use std::time::{Duration, Instant};

use crate::codec::Codec;
use crate::entry::ValueType;
use crate::store::Store;

/// A cursor for incremental scans.
///
/// `ScanCursor(0)` means "start" or "done", consistent with Redis scan cursors.
/// Higher values represent an offset into a snapshot in this MVP implementation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ScanCursor(pub u64);

/// A single page of results returned by `scan`-style APIs.
#[derive(Debug, Clone)]
pub struct ScanPage<T> {
    /// The cursor to use for the next page. `ScanCursor(0)` indicates completion.
    pub cursor: ScanCursor,
    /// Items in this page.
    pub items: Vec<T>,
}

/// Key-space operations (lifecycle, TTL, renaming, and key enumeration).
///
/// Notes:
/// - Expiration is checked lazily on access (`exists`, `ttl`, `type`, etc.).
/// - `scan` and `iter_keys` use a snapshot of current keys (MVP implementation).
///
/// # Example
///
/// ```rust
/// use heaplet::{keys::ScanCursor, Store};
///
/// let store = Store::new();
/// store.kv().set("user:1", &1_i32).unwrap();
/// store.kv().set("user:2", &2_i32).unwrap();
/// store.kv().set("other", &0_i32).unwrap();
///
/// assert!(store.keys().exists("user:1"));
/// assert_eq!(store.keys().r#type("user:1"), Some(heaplet::entry::ValueType::String));
///
/// // Snapshot scan + paging.
/// let p1 = store.keys().scan(ScanCursor(0), Some("user:*"), 1);
/// assert_eq!(p1.items.len(), 1);
///
/// let p2 = store.keys().scan(p1.cursor, Some("user:*"), 10);
/// assert_eq!(p2.cursor.0, 0); // done
///
/// // Rename and delete.
/// store.keys().rename("other", "other2").unwrap();
/// assert!(store.keys().del("other2"));
/// ```
pub struct KeysOps<'a, C: Codec> {
    store: &'a Store<C>,
}

impl<'a, C: Codec> KeysOps<'a, C> {
    pub(crate) fn new(store: &'a Store<C>) -> Self {
        Self { store }
    }

    /// Returns `true` if the key exists.
    ///
    /// Expired keys may be removed during this call.
    #[must_use]
    pub fn exists(&self, key: &str) -> bool {
        self.store.purge_if_expired(key);
        self.store.contains_key(key)
    }

    /// Deletes the key.
    ///
    /// Returns `true` if the key existed.
    pub fn del(&self, key: &str) -> bool {
        self.store.remove_entry(key)
    }

    /// Sets the key expiration in seconds.
    ///
    /// Returns `false` if the key does not exist.
    pub fn expire(&self, key: &str, seconds: u64) -> bool {
        self.pexpire(key, seconds.saturating_mul(1000))
    }

    /// Sets the key expiration in milliseconds.
    ///
    /// Returns `false` if the key does not exist.
    pub fn pexpire(&self, key: &str, ms: u64) -> bool {
        self.store.purge_if_expired(key);
        if self.store.get_entry_meta(key).is_none() {
            return false;
        }

        let when = Instant::now() + Duration::from_millis(ms);
        self.store.set_expire_at(key, Some(when))
    }

    /// Returns the TTL (time-to-live) in seconds.
    ///
    /// Semantics:
    /// - `None`: key does not exist
    /// - `Some(-1)`: key exists but has no expiration
    /// - `Some(n >= 0)`: remaining TTL in seconds
    #[must_use]
    pub fn ttl(&self, key: &str) -> Option<i64> {
        self.store.purge_if_expired(key);
        let meta = self.store.get_entry_meta(key)?;

        match meta.expire_at {
            None => Some(-1),
            Some(t) => {
                let now = Instant::now();
                if t <= now {
                    // Should not happen due to purge, but keep it safe.
                    self.store.remove_entry(key);
                    None
                } else {
                    Some((t - now).as_secs() as i64)
                }
            }
        }
    }

    /// Returns the PTTL (time-to-live) in milliseconds.
    ///
    /// Semantics are the same as [`ttl`](Self::ttl).
    #[must_use]
    pub fn pttl(&self, key: &str) -> Option<i64> {
        self.store.purge_if_expired(key);
        let meta = self.store.get_entry_meta(key)?;

        match meta.expire_at {
            None => Some(-1),
            Some(t) => {
                let now = Instant::now();
                if t <= now {
                    self.store.remove_entry(key);
                    None
                } else {
                    Some((t - now).as_millis() as i64)
                }
            }
        }
    }

    /// Removes expiration from the key.
    ///
    /// Returns `false` if the key does not exist.
    pub fn persist(&self, key: &str) -> bool {
        self.store.purge_if_expired(key);
        self.store.set_expire_at(key, None)
    }

    /// Returns the stored value type of the key.
    ///
    /// Returns `None` if the key does not exist.
    #[must_use]
    pub fn r#type(&self, key: &str) -> Option<ValueType> {
        self.store.purge_if_expired(key);
        self.store.get_entry_meta(key).map(|m| m.value_type)
    }

    /// Renames `from` to `to`, overwriting `to` if it exists.
    pub fn rename(&self, from: &str, to: &str) -> Result<(), crate::error::Error> {
        self.store.purge_if_expired(from);
        self.store.purge_if_expired(to);
        self.store
            .rename_internal(from, to, /* nx = */ false)
            .map(|_| ())
    }

    /// Renames `from` to `to` if `to` does not exist.
    ///
    /// Returns:
    /// - `Ok(true)` if the rename succeeded
    /// - `Ok(false)` if the destination already existed
    pub fn renamenx(&self, from: &str, to: &str) -> Result<bool, crate::error::Error> {
        self.store.purge_if_expired(from);
        self.store.purge_if_expired(to);
        self.store.rename_internal(from, to, /* nx = */ true)
    }

    /// Cursor-based scan over keys.
    ///
    /// Parameters:
    /// - `cursor`: start from `ScanCursor(0)`
    /// - `pattern`: optional glob-like pattern (MVP supports `*`)
    /// - `count`: page size hint (minimum is 1)
    ///
    /// This MVP implementation snapshots all keys, sorts them, applies an optional
    /// pattern filter, then returns a slice of results.
    #[must_use]
    pub fn scan(
        &self,
        cursor: ScanCursor,
        pattern: Option<&str>,
        count: usize,
    ) -> ScanPage<String> {
        let mut keys = self.store.snapshot_keys();
        keys.sort();

        let filtered: Vec<String> = match pattern {
            None => keys,
            Some(pat) => keys
                .into_iter()
                .filter(|k| matches_pattern(k, pat))
                .collect(),
        };

        if filtered.is_empty() {
            return ScanPage {
                cursor: ScanCursor(0),
                items: vec![],
            };
        }

        let start = cursor.0 as usize;
        if start >= filtered.len() {
            return ScanPage {
                cursor: ScanCursor(0),
                items: vec![],
            };
        }

        let take = count.max(1);
        let end = (start + take).min(filtered.len());
        let items = filtered[start..end].to_vec();

        let next = if end >= filtered.len() { 0 } else { end as u64 };
        ScanPage {
            cursor: ScanCursor(next),
            items,
        }
    }

    /// Returns an iterator over keys (snapshot-based).
    ///
    /// This is a convenience wrapper around key snapshot + optional filtering.
    pub fn iter_keys(&self, pattern: Option<&str>) -> impl Iterator<Item = String> {
        let mut keys = self.store.snapshot_keys();
        keys.sort();

        let v: Vec<String> = match pattern {
            None => keys,
            Some(pat) => keys
                .into_iter()
                .filter(|k| matches_pattern(k, pat))
                .collect(),
        };

        v.into_iter()
    }
}

/// A minimal glob-like matcher used by scan-style APIs.
///
/// Supported patterns:
/// - `"*"` matches everything
/// - `"prefix*"` matches keys starting with `prefix`
/// - `"*suffix"` matches keys ending with `suffix`
/// - `"*mid*"` matches keys containing `mid`
///
/// For other patterns with multiple `*`, this falls back to checking that all
/// non-empty segments appear in order.
fn matches_pattern(s: &str, pat: &str) -> bool {
    if pat == "*" {
        return true;
    }

    let parts: Vec<&str> = pat.split('*').collect();
    match parts.as_slice() {
        // No wildcard.
        [one] => s == *one,
        // prefix*
        [prefix, ""] => s.starts_with(prefix),
        // *suffix
        ["", suffix] => s.ends_with(suffix),
        // *mid*
        ["", mid, ""] => s.contains(mid),
        _ => {
            // Fallback: naive ordered contains for all non-empty parts.
            let mut idx = 0usize;
            for part in parts.into_iter().filter(|p| !p.is_empty()) {
                if let Some(pos) = s[idx..].find(part) {
                    idx += pos + part.len();
                } else {
                    return false;
                }
            }
            true
        }
    }
}

/// Internal-only re-export so other modules can reuse the same matcher.
pub(crate) fn matches_pattern_for_internal_use(s: &str, pat: &str) -> bool {
    matches_pattern(s, pat)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Store;

    #[test]
    fn keys_exists_del_type() {
        let store = Store::new();
        assert!(!store.keys().exists("k"));
        store.kv().set("k", &123_i32).unwrap();
        assert!(store.keys().exists("k"));
        assert_eq!(store.keys().r#type("k"), Some(ValueType::String));
        assert!(store.keys().del("k"));
        assert!(!store.keys().exists("k"));
        assert_eq!(store.keys().r#type("k"), None);
    }

    #[test]
    fn keys_ttl_persist_expire() {
        let store = Store::new();
        store.kv().set("k", &"v").unwrap();

        assert_eq!(store.keys().ttl("k"), Some(-1));
        assert!(store.keys().expire("k", 1));
        let t = store.keys().ttl("k").unwrap();
        assert!(t >= 0);

        assert!(store.keys().persist("k"));
        assert_eq!(store.keys().ttl("k"), Some(-1));
    }

    #[test]
    fn keys_rename_renamenx() {
        let store = Store::new();
        store.kv().set("a", &1_i32).unwrap();

        store.keys().rename("a", "b").unwrap();
        let v: Option<i32> = store.kv().get("b").unwrap();
        assert_eq!(v, Some(1));

        store.kv().set("x", &1_i32).unwrap();
        store.kv().set("y", &2_i32).unwrap();
        let ok = store.keys().renamenx("x", "y").unwrap();
        assert!(!ok); // y exists
    }

    #[test]
    fn keys_scan_iter_keys() {
        let store = Store::new();
        store.kv().set("user:1", &1_i32).unwrap();
        store.kv().set("user:2", &2_i32).unwrap();
        store.kv().set("other", &0_i32).unwrap();

        let page = store.keys().scan(ScanCursor(0), Some("user:*"), 10);
        assert_eq!(page.cursor, ScanCursor(0));
        assert_eq!(page.items.len(), 2);

        let all: Vec<String> = store.keys().iter_keys(Some("*")).collect();
        assert!(all.len() >= 3);
    }

    #[test]
    fn keys_edge_cases() {
        let store = Store::new();

        // rename src missing
        assert!(store.keys().rename("missing", "x").is_err());

        store.kv().set("a", &1_i64).unwrap();
        store.kv().set("b", &2_i64).unwrap();

        // renamenx should fail if dest exists
        assert!(!store.keys().renamenx("a", "b").unwrap());

        // expire on missing
        assert!(!store.keys().expire("missing", 1));

        // ttl on existing without expire
        assert_eq!(store.keys().ttl("a"), Some(-1));

        // persist on key without ttl
        assert!(store.keys().persist("a"));

        // scan cursor out of range should return empty and cursor=0
        let p = store.keys().scan(ScanCursor(10_000), None, 10);
        assert_eq!(p.cursor.0, 0);
    }
}