ember_core/

keyspace.rs

//! The keyspace: Ember's core key-value store.
//!
//! A `Keyspace` owns a flat `HashMap<String, Entry>` and handles
//! get, set, delete, existence checks, and TTL management. Expired
//! keys are removed lazily on access. Memory usage is tracked on
//! every mutation for eviction and stats reporting.

use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};

use bytes::Bytes;
use rand::seq::IteratorRandom;

use crate::memory::{self, MemoryTracker};
use crate::types::sorted_set::{SortedSet, ZAddFlags};
use crate::types::{self, normalize_range, Value};

/// Error returned when a command is used against a key holding the wrong type.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WrongType;

impl std::fmt::Display for WrongType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "WRONGTYPE Operation against a key holding the wrong kind of value"
        )
    }
}

impl std::error::Error for WrongType {}

/// Error returned by write operations that may fail due to type mismatch
/// or memory limits.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WriteError {
    /// The key holds a different type than expected.
    WrongType,
    /// Memory limit reached and eviction couldn't free enough space.
    OutOfMemory,
}

impl From<WrongType> for WriteError {
    fn from(_: WrongType) -> Self {
        WriteError::WrongType
    }
}

/// Errors that can occur during INCR/DECR operations.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IncrError {
    /// Key holds a non-string type.
    WrongType,
    /// Value is not a valid integer.
    NotAnInteger,
    /// Increment or decrement would overflow i64.
    Overflow,
    /// Memory limit reached.
    OutOfMemory,
}

impl std::fmt::Display for IncrError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            IncrError::WrongType => write!(
                f,
                "WRONGTYPE Operation against a key holding the wrong kind of value"
            ),
            IncrError::NotAnInteger => write!(f, "ERR value is not an integer or out of range"),
            IncrError::Overflow => write!(f, "ERR increment or decrement would overflow"),
            IncrError::OutOfMemory => {
                write!(f, "OOM command not allowed when used memory > 'maxmemory'")
            }
        }
    }
}

impl std::error::Error for IncrError {}
/// Result of a ZADD operation, containing both the client-facing count
/// and the list of members that were actually applied (for AOF correctness).
#[derive(Debug, Clone)]
pub struct ZAddResult {
    /// Number of members added (or added+updated if CH flag was set).
    pub count: usize,
    /// Members that were actually inserted or had their score updated.
    /// Only these should be persisted to the AOF.
    pub applied: Vec<(f64, String)>,
}

/// How the keyspace should handle writes when the memory limit is reached.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum EvictionPolicy {
    /// Return an error on writes when memory is full.
    #[default]
    NoEviction,
    /// Evict the least-recently-used key (approximated via random sampling).
    AllKeysLru,
}

/// Configuration for a single keyspace / shard.
#[derive(Debug, Clone)]
pub struct ShardConfig {
    /// Maximum memory in bytes. `None` means unlimited.
    pub max_memory: Option<usize>,
    /// What to do when memory is full.
    pub eviction_policy: EvictionPolicy,
    /// Numeric identifier for this shard (used for persistence file naming).
    pub shard_id: u16,
}

impl Default for ShardConfig {
    fn default() -> Self {
        Self {
            max_memory: None,
            eviction_policy: EvictionPolicy::NoEviction,
            shard_id: 0,
        }
    }
}

/// Result of a set operation that may fail under memory pressure.
#[derive(Debug, PartialEq, Eq)]
pub enum SetResult {
    /// The key was stored successfully.
    Ok,
    /// Memory limit reached and eviction policy is NoEviction.
    OutOfMemory,
}

/// A single entry in the keyspace: a value plus optional expiration
/// and last access time for LRU approximation.
#[derive(Debug, Clone)]
pub(crate) struct Entry {
    pub(crate) value: Value,
    pub(crate) expires_at: Option<Instant>,
    pub(crate) last_access: Instant,
}

impl Entry {
    fn new(value: Value, expires_at: Option<Instant>) -> Self {
        Self {
            value,
            expires_at,
            last_access: Instant::now(),
        }
    }

    /// Returns `true` if this entry has passed its expiration time.
    fn is_expired(&self) -> bool {
        match self.expires_at {
            Some(deadline) => Instant::now() >= deadline,
            None => false,
        }
    }

    /// Marks this entry as accessed right now.
    fn touch(&mut self) {
        self.last_access = Instant::now();
    }
}

/// Result of a TTL query, matching Redis semantics.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TtlResult {
    /// Key exists and has a TTL. Returns remaining seconds.
    Seconds(u64),
    /// Key exists and has a TTL. Returns remaining milliseconds.
    Milliseconds(u64),
    /// Key exists but has no expiration set.
    NoExpiry,
    /// Key does not exist.
    NotFound,
}

/// Aggregated statistics for a keyspace.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct KeyspaceStats {
    /// Number of live keys.
    pub key_count: usize,
    /// Estimated memory usage in bytes.
    pub used_bytes: usize,
    /// Number of keys with an expiration set.
    pub keys_with_expiry: usize,
}

/// Number of random keys to sample when looking for an eviction candidate.
///
/// Eviction uses sampling-based approximate LRU — we randomly select this many
/// keys and evict the least-recently-accessed among them. This trades perfect
/// LRU accuracy for O(1) eviction (no sorted structure to maintain).
///
/// Larger sample sizes give better LRU approximation but cost more per eviction.
/// 16 is a reasonable balance — similar to Redis's default sample size. With
/// 16 samples, we statistically find a good eviction candidate while keeping
/// eviction overhead low even at millions of keys.
const EVICTION_SAMPLE_SIZE: usize = 16;

/// The core key-value store.
///
/// All operations are single-threaded per shard — no internal locking.
/// Memory usage is tracked incrementally on every mutation.
pub struct Keyspace {
    entries: HashMap<String, Entry>,
    memory: MemoryTracker,
    config: ShardConfig,
    /// Number of entries that currently have an expiration set.
    expiry_count: usize,
}

impl Keyspace {
    /// Creates a new, empty keyspace with default config (no memory limit).
    pub fn new() -> Self {
        Self::with_config(ShardConfig::default())
    }

    /// Creates a new, empty keyspace with the given config.
    pub fn with_config(config: ShardConfig) -> Self {
        Self {
            entries: HashMap::new(),
            memory: MemoryTracker::new(),
            config,
            expiry_count: 0,
        }
    }

    /// Retrieves the string value for `key`, or `None` if missing/expired.
    ///
    /// Returns `Err(WrongType)` if the key holds a non-string value.
    /// Expired keys are removed lazily on access. Successful reads update
    /// the entry's last access time for LRU tracking.
    pub fn get(&mut self, key: &str) -> Result<Option<Value>, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(None);
        }
        match self.entries.get_mut(key) {
            Some(e) => match &e.value {
                Value::String(_) => {
                    e.touch();
                    Ok(Some(e.value.clone()))
                }
                _ => Err(WrongType),
            },
            None => Ok(None),
        }
    }

    /// Returns the type name of the value at `key`, or "none" if missing.
    pub fn value_type(&mut self, key: &str) -> &'static str {
        if self.remove_if_expired(key) {
            return "none";
        }
        match self.entries.get(key) {
            Some(e) => types::type_name(&e.value),
            None => "none",
        }
    }

    /// Stores a key-value pair. If the key already existed, the old entry
    /// (including any TTL) is replaced entirely.
    ///
    /// `expire` sets an optional TTL as a duration from now.
    ///
    /// Returns `SetResult::OutOfMemory` if the memory limit is reached
    /// and the eviction policy is `NoEviction`. With `AllKeysLru`, this
    /// will evict keys to make room before inserting.
    pub fn set(&mut self, key: String, value: Bytes, expire: Option<Duration>) -> SetResult {
        let expires_at = expire.map(|d| Instant::now() + d);
        let new_value = Value::String(value);

        // check memory limit — for overwrites, only the net increase matters
        let new_size = memory::entry_size(&key, &new_value);
        let old_size = self
            .entries
            .get(&key)
            .map(|e| memory::entry_size(&key, &e.value))
            .unwrap_or(0);
        let net_increase = new_size.saturating_sub(old_size);

        if !self.enforce_memory_limit(net_increase) {
            return SetResult::OutOfMemory;
        }

        if let Some(old_entry) = self.entries.get(&key) {
            self.memory.replace(&key, &old_entry.value, &new_value);
            // adjust expiry count if the TTL status changed
            let had_expiry = old_entry.expires_at.is_some();
            let has_expiry = expires_at.is_some();
            match (had_expiry, has_expiry) {
                (false, true) => self.expiry_count += 1,
                (true, false) => self.expiry_count = self.expiry_count.saturating_sub(1),
                _ => {}
            }
        } else {
            self.memory.add(&key, &new_value);
            if expires_at.is_some() {
                self.expiry_count += 1;
            }
        }

        self.entries.insert(key, Entry::new(new_value, expires_at));
        SetResult::Ok
    }

    /// Tries to evict one key using LRU approximation.
    ///
    /// Randomly samples `EVICTION_SAMPLE_SIZE` keys and removes the one
    /// with the oldest `last_access` time. Returns `true` if a key was
    /// evicted, `false` if the keyspace is empty.
    fn try_evict(&mut self) -> bool {
        if self.entries.is_empty() {
            return false;
        }

        let mut rng = rand::rng();

        // randomly sample keys and find the least recently accessed one
        let victim = self
            .entries
            .iter()
            .choose_multiple(&mut rng, EVICTION_SAMPLE_SIZE)
            .into_iter()
            .min_by_key(|(_, entry)| entry.last_access)
            .map(|(k, _)| k.clone());

        if let Some(key) = victim {
            if let Some(entry) = self.entries.remove(&key) {
                self.memory.remove(&key, &entry.value);
                if entry.expires_at.is_some() {
                    self.expiry_count = self.expiry_count.saturating_sub(1);
                }
                return true;
            }
        }
        false
    }

    /// Checks whether the memory limit allows a write that would increase
    /// usage by `estimated_increase` bytes. Attempts eviction if the
    /// policy allows it. Returns `true` if the write can proceed.
    fn enforce_memory_limit(&mut self, estimated_increase: usize) -> bool {
        if let Some(max) = self.config.max_memory {
            while self.memory.used_bytes() + estimated_increase > max {
                match self.config.eviction_policy {
                    EvictionPolicy::NoEviction => return false,
                    EvictionPolicy::AllKeysLru => {
                        if !self.try_evict() {
                            return false;
                        }
                    }
                }
            }
        }
        true
    }

    /// Removes a key. Returns `true` if the key existed (and wasn't expired).
    pub fn del(&mut self, key: &str) -> bool {
        if self.remove_if_expired(key) {
            return false;
        }
        if let Some(entry) = self.entries.remove(key) {
            self.memory.remove(key, &entry.value);
            if entry.expires_at.is_some() {
                self.expiry_count = self.expiry_count.saturating_sub(1);
            }
            true
        } else {
            false
        }
    }

    /// Returns `true` if the key exists and hasn't expired.
    pub fn exists(&mut self, key: &str) -> bool {
        if self.remove_if_expired(key) {
            return false;
        }
        self.entries.contains_key(key)
    }

    /// Sets an expiration on an existing key. Returns `true` if the key
    /// exists (and the TTL was set), `false` if the key doesn't exist.
    pub fn expire(&mut self, key: &str, seconds: u64) -> bool {
        if self.remove_if_expired(key) {
            return false;
        }
        match self.entries.get_mut(key) {
            Some(entry) => {
                if entry.expires_at.is_none() {
                    self.expiry_count += 1;
                }
                entry.expires_at = Some(Instant::now() + Duration::from_secs(seconds));
                true
            }
            None => false,
        }
    }

    /// Returns the TTL status for a key, following Redis semantics:
    /// - `Seconds(n)` if the key has a TTL
    /// - `NoExpiry` if the key exists without a TTL
    /// - `NotFound` if the key doesn't exist
    pub fn ttl(&mut self, key: &str) -> TtlResult {
        if self.remove_if_expired(key) {
            return TtlResult::NotFound;
        }
        match self.entries.get(key) {
            Some(entry) => match entry.expires_at {
                Some(deadline) => {
                    let remaining = deadline.saturating_duration_since(Instant::now());
                    TtlResult::Seconds(remaining.as_secs())
                }
                None => TtlResult::NoExpiry,
            },
            None => TtlResult::NotFound,
        }
    }

    /// Removes the expiration from a key.
    ///
    /// Returns `true` if the key existed and had a timeout that was removed.
    /// Returns `false` if the key doesn't exist or has no expiration.
    pub fn persist(&mut self, key: &str) -> bool {
        if self.remove_if_expired(key) {
            return false;
        }
        match self.entries.get_mut(key) {
            Some(entry) => {
                if entry.expires_at.is_some() {
                    entry.expires_at = None;
                    self.expiry_count = self.expiry_count.saturating_sub(1);
                    true
                } else {
                    false
                }
            }
            None => false,
        }
    }

    /// Returns the TTL status for a key in milliseconds, following Redis semantics:
    /// - `Milliseconds(n)` if the key has a TTL
    /// - `NoExpiry` if the key exists without a TTL
    /// - `NotFound` if the key doesn't exist
    pub fn pttl(&mut self, key: &str) -> TtlResult {
        if self.remove_if_expired(key) {
            return TtlResult::NotFound;
        }
        match self.entries.get(key) {
            Some(entry) => match entry.expires_at {
                Some(deadline) => {
                    let remaining = deadline.saturating_duration_since(Instant::now());
                    TtlResult::Milliseconds(remaining.as_millis() as u64)
                }
                None => TtlResult::NoExpiry,
            },
            None => TtlResult::NotFound,
        }
    }

    /// Sets an expiration on an existing key in milliseconds.
    ///
    /// Returns `true` if the key exists (and the TTL was set),
    /// `false` if the key doesn't exist.
    pub fn pexpire(&mut self, key: &str, millis: u64) -> bool {
        if self.remove_if_expired(key) {
            return false;
        }
        match self.entries.get_mut(key) {
            Some(entry) => {
                if entry.expires_at.is_none() {
                    self.expiry_count += 1;
                }
                entry.expires_at = Some(Instant::now() + Duration::from_millis(millis));
                true
            }
            None => false,
        }
    }

    /// Increments the integer value of a key by 1.
    ///
    /// If the key doesn't exist, it's initialized to 0 before incrementing.
    /// Returns the new value after the operation.
    pub fn incr(&mut self, key: &str) -> Result<i64, IncrError> {
        self.incr_by(key, 1)
    }

    /// Decrements the integer value of a key by 1.
    ///
    /// If the key doesn't exist, it's initialized to 0 before decrementing.
    /// Returns the new value after the operation.
    pub fn decr(&mut self, key: &str) -> Result<i64, IncrError> {
        self.incr_by(key, -1)
    }

    /// Shared implementation for INCR/DECR. Adds `delta` to the current
    /// integer value of the key, creating it if necessary.
    ///
    /// Preserves the existing TTL when updating an existing key.
    fn incr_by(&mut self, key: &str, delta: i64) -> Result<i64, IncrError> {
        self.remove_if_expired(key);

        // read current value and TTL
        let (current, existing_expire) = match self.entries.get(key) {
            Some(entry) => {
                let val = match &entry.value {
                    Value::String(data) => {
                        let s = std::str::from_utf8(data).map_err(|_| IncrError::NotAnInteger)?;
                        s.parse::<i64>().map_err(|_| IncrError::NotAnInteger)?
                    }
                    _ => return Err(IncrError::WrongType),
                };
                let expire = entry
                    .expires_at
                    .map(|deadline| deadline.saturating_duration_since(Instant::now()));
                (val, expire)
            }
            None => (0, None),
        };

        let new_val = current.checked_add(delta).ok_or(IncrError::Overflow)?;
        let new_bytes = Bytes::from(new_val.to_string());

        match self.set(key.to_owned(), new_bytes, existing_expire) {
            SetResult::Ok => Ok(new_val),
            SetResult::OutOfMemory => Err(IncrError::OutOfMemory),
        }
    }

    /// Returns aggregated stats for this keyspace.
    ///
    /// All fields are tracked incrementally — this is O(1).
    pub fn stats(&self) -> KeyspaceStats {
        KeyspaceStats {
            key_count: self.memory.key_count(),
            used_bytes: self.memory.used_bytes(),
            keys_with_expiry: self.expiry_count,
        }
    }

    /// Returns the number of live keys.
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Removes all keys from the keyspace.
    pub fn clear(&mut self) {
        self.entries.clear();
        self.memory.reset();
        self.expiry_count = 0;
    }

    /// Returns `true` if the keyspace has no entries.
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }

    /// Scans keys starting from a cursor position.
    ///
    /// Returns the next cursor (0 if scan complete) and a batch of keys.
    /// The `pattern` argument supports glob-style matching (*, ?, [abc]).
    pub fn scan_keys(
        &self,
        cursor: u64,
        count: usize,
        pattern: Option<&str>,
    ) -> (u64, Vec<String>) {
        let mut keys = Vec::with_capacity(count);
        let mut position = 0u64;
        let target_count = if count == 0 { 10 } else { count };

        for (key, entry) in self.entries.iter() {
            // skip expired entries
            if entry.is_expired() {
                continue;
            }

            // skip entries before cursor
            if position < cursor {
                position += 1;
                continue;
            }

            // pattern matching
            if let Some(pat) = pattern {
                if !glob_match(pat, key) {
                    position += 1;
                    continue;
                }
            }

            keys.push(key.clone());
            position += 1;

            if keys.len() >= target_count {
                // return position as next cursor
                return (position, keys);
            }
        }

        // scan complete
        (0, keys)
    }

    /// Iterates over all live (non-expired) entries, yielding the key, a
    /// clone of the value, and the remaining TTL in milliseconds (-1 for
    /// entries with no expiration). Used by snapshot and AOF rewrite.
    pub fn iter_entries(&self) -> impl Iterator<Item = (&str, &Value, i64)> {
        let now = Instant::now();
        self.entries.iter().filter_map(move |(key, entry)| {
            if entry.is_expired() {
                return None;
            }
            let ttl_ms = match entry.expires_at {
                Some(deadline) => {
                    let remaining = deadline.saturating_duration_since(now);
                    remaining.as_millis() as i64
                }
                None => -1,
            };
            Some((key.as_str(), &entry.value, ttl_ms))
        })
    }

    /// Restores an entry during recovery, bypassing memory limits.
    ///
    /// If `expires_at` is in the past, the entry is silently skipped.
    /// This is used only during shard startup when loading from
    /// snapshot/AOF — normal writes should go through `set()`.
    pub fn restore(&mut self, key: String, value: Value, expires_at: Option<Instant>) {
        // skip entries that already expired
        if let Some(deadline) = expires_at {
            if Instant::now() >= deadline {
                return;
            }
        }

        // if replacing an existing entry, adjust memory tracking
        if let Some(old) = self.entries.get(&key) {
            self.memory.replace(&key, &old.value, &value);
            let had_expiry = old.expires_at.is_some();
            let has_expiry = expires_at.is_some();
            match (had_expiry, has_expiry) {
                (false, true) => self.expiry_count += 1,
                (true, false) => self.expiry_count = self.expiry_count.saturating_sub(1),
                _ => {}
            }
        } else {
            self.memory.add(&key, &value);
            if expires_at.is_some() {
                self.expiry_count += 1;
            }
        }

        self.entries.insert(key, Entry::new(value, expires_at));
    }

    // -- list operations --

    /// Pushes one or more values to the head (left) of a list.
    ///
    /// Creates the list if the key doesn't exist. Returns `Err(WriteError::WrongType)`
    /// if the key exists but holds a non-list value, or
    /// `Err(WriteError::OutOfMemory)` if the memory limit is reached.
    /// Returns the new length on success.
    pub fn lpush(&mut self, key: &str, values: &[Bytes]) -> Result<usize, WriteError> {
        self.list_push(key, values, true)
    }

    /// Pushes one or more values to the tail (right) of a list.
    ///
    /// Creates the list if the key doesn't exist. Returns `Err(WriteError::WrongType)`
    /// if the key exists but holds a non-list value, or
    /// `Err(WriteError::OutOfMemory)` if the memory limit is reached.
    /// Returns the new length on success.
    pub fn rpush(&mut self, key: &str, values: &[Bytes]) -> Result<usize, WriteError> {
        self.list_push(key, values, false)
    }

    /// Pops a value from the head (left) of a list.
    ///
    /// Returns `Ok(None)` if the key doesn't exist. Removes the key if
    /// the list becomes empty. Returns `Err(WrongType)` on type mismatch.
    pub fn lpop(&mut self, key: &str) -> Result<Option<Bytes>, WrongType> {
        self.list_pop(key, true)
    }

    /// Pops a value from the tail (right) of a list.
    ///
    /// Returns `Ok(None)` if the key doesn't exist. Removes the key if
    /// the list becomes empty. Returns `Err(WrongType)` on type mismatch.
    pub fn rpop(&mut self, key: &str) -> Result<Option<Bytes>, WrongType> {
        self.list_pop(key, false)
    }

    /// Returns a range of elements from a list by index.
    ///
    /// Supports negative indices (e.g. -1 = last element). Out-of-bounds
    /// indices are clamped to the list boundaries. Returns `Err(WrongType)`
    /// on type mismatch. Missing keys return an empty vec.
    pub fn lrange(&mut self, key: &str, start: i64, stop: i64) -> Result<Vec<Bytes>, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(vec![]);
        }
        match self.entries.get_mut(key) {
            None => Ok(vec![]),
            Some(entry) => {
                let result = match &entry.value {
                    Value::List(deque) => {
                        let len = deque.len() as i64;
                        let (s, e) = normalize_range(start, stop, len);
                        if s > e {
                            return Ok(vec![]);
                        }
                        Ok(deque
                            .iter()
                            .skip(s as usize)
                            .take((e - s + 1) as usize)
                            .cloned()
                            .collect())
                    }
                    _ => Err(WrongType),
                };
                if result.is_ok() {
                    entry.touch();
                }
                result
            }
        }
    }

    /// Returns the length of a list, or 0 if the key doesn't exist.
    ///
    /// Returns `Err(WrongType)` on type mismatch.
    pub fn llen(&mut self, key: &str) -> Result<usize, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(0);
        }
        match self.entries.get(key) {
            None => Ok(0),
            Some(entry) => match &entry.value {
                Value::List(deque) => Ok(deque.len()),
                _ => Err(WrongType),
            },
        }
    }

    /// Internal push implementation shared by lpush/rpush.
    fn list_push(&mut self, key: &str, values: &[Bytes], left: bool) -> Result<usize, WriteError> {
        self.remove_if_expired(key);

        let is_new = !self.entries.contains_key(key);

        if !is_new && !matches!(self.entries[key].value, Value::List(_)) {
            return Err(WriteError::WrongType);
        }

        // estimate the memory increase and enforce the limit before mutating
        let element_increase: usize = values
            .iter()
            .map(|v| memory::VECDEQUE_ELEMENT_OVERHEAD + v.len())
            .sum();
        let estimated_increase = if is_new {
            memory::ENTRY_OVERHEAD + key.len() + memory::VECDEQUE_BASE_OVERHEAD + element_increase
        } else {
            element_increase
        };
        if !self.enforce_memory_limit(estimated_increase) {
            return Err(WriteError::OutOfMemory);
        }

        if is_new {
            let value = Value::List(VecDeque::new());
            self.memory.add(key, &value);
            self.entries.insert(key.to_owned(), Entry::new(value, None));
        }

        let entry = self
            .entries
            .get_mut(key)
            .expect("just inserted or verified");
        let old_entry_size = memory::entry_size(key, &entry.value);

        if let Value::List(ref mut deque) = entry.value {
            for val in values {
                if left {
                    deque.push_front(val.clone());
                } else {
                    deque.push_back(val.clone());
                }
            }
        }
        entry.touch();

        let new_entry_size = memory::entry_size(key, &entry.value);
        self.memory.adjust(old_entry_size, new_entry_size);

        let len = match &entry.value {
            Value::List(d) => d.len(),
            _ => unreachable!(),
        };
        Ok(len)
    }

    /// Internal pop implementation shared by lpop/rpop.
    fn list_pop(&mut self, key: &str, left: bool) -> Result<Option<Bytes>, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(None);
        }

        match self.entries.get(key) {
            None => return Ok(None),
            Some(e) => {
                if !matches!(e.value, Value::List(_)) {
                    return Err(WrongType);
                }
            }
        };

        let old_entry_size = memory::entry_size(key, &self.entries[key].value);
        let entry = self.entries.get_mut(key).expect("verified above");
        let popped = if let Value::List(ref mut deque) = entry.value {
            if left {
                deque.pop_front()
            } else {
                deque.pop_back()
            }
        } else {
            unreachable!()
        };
        entry.touch();

        // check if list is now empty — if so, delete the key entirely
        let is_empty = matches!(&entry.value, Value::List(d) if d.is_empty());
        if is_empty {
            let removed = self.entries.remove(key).expect("verified above");
            // use remove_with_size since the value was already mutated
            self.memory.remove_with_size(old_entry_size);
            if removed.expires_at.is_some() {
                self.expiry_count = self.expiry_count.saturating_sub(1);
            }
        } else {
            let new_entry_size = memory::entry_size(key, &self.entries[key].value);
            self.memory.adjust(old_entry_size, new_entry_size);
        }

        Ok(popped)
    }

    // -- sorted set operations --

    /// Adds members with scores to a sorted set, with optional ZADD flags.
    ///
    /// Creates the sorted set if the key doesn't exist. Returns a
    /// `ZAddResult` containing the count for the client response and the
    /// list of members that were actually applied (for AOF correctness).
    /// Returns `Err(WriteError::WrongType)` on type mismatch, or
    /// `Err(WriteError::OutOfMemory)` if the memory limit is reached.
    pub fn zadd(
        &mut self,
        key: &str,
        members: &[(f64, String)],
        flags: &ZAddFlags,
    ) -> Result<ZAddResult, WriteError> {
        self.remove_if_expired(key);

        let is_new = !self.entries.contains_key(key);
        if !is_new && !matches!(self.entries[key].value, Value::SortedSet(_)) {
            return Err(WriteError::WrongType);
        }

        // worst-case estimate: assume all members are new
        let member_increase: usize = members
            .iter()
            .map(|(_, m)| SortedSet::estimated_member_cost(m))
            .sum();
        let estimated_increase = if is_new {
            memory::ENTRY_OVERHEAD + key.len() + SortedSet::BASE_OVERHEAD + member_increase
        } else {
            member_increase
        };
        if !self.enforce_memory_limit(estimated_increase) {
            return Err(WriteError::OutOfMemory);
        }

        if is_new {
            let value = Value::SortedSet(SortedSet::new());
            self.memory.add(key, &value);
            self.entries.insert(key.to_owned(), Entry::new(value, None));
        }

        let entry = self
            .entries
            .get_mut(key)
            .expect("just inserted or verified");
        let old_entry_size = memory::entry_size(key, &entry.value);

        let mut count = 0;
        let mut applied = Vec::new();
        if let Value::SortedSet(ref mut ss) = entry.value {
            for (score, member) in members {
                let result = ss.add_with_flags(member.clone(), *score, flags);
                if result.added || result.updated {
                    applied.push((*score, member.clone()));
                }
                if flags.ch {
                    if result.added || result.updated {
                        count += 1;
                    }
                } else if result.added {
                    count += 1;
                }
            }
        }
        entry.touch();

        let new_entry_size = memory::entry_size(key, &entry.value);
        self.memory.adjust(old_entry_size, new_entry_size);

        // clean up if the set is still empty (e.g. XX on a new key)
        if let Value::SortedSet(ref ss) = entry.value {
            if ss.is_empty() {
                self.memory
                    .remove_with_size(memory::entry_size(key, &entry.value));
                self.entries.remove(key);
            }
        }

        Ok(ZAddResult { count, applied })
    }

    /// Removes members from a sorted set. Returns the names of members
    /// that were actually removed (for AOF correctness). Deletes the key
    /// if the set becomes empty.
    ///
    /// Returns `Err(WrongType)` if the key holds a non-sorted-set value.
    pub fn zrem(&mut self, key: &str, members: &[String]) -> Result<Vec<String>, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(vec![]);
        }

        match self.entries.get(key) {
            None => return Ok(vec![]),
            Some(e) => {
                if !matches!(e.value, Value::SortedSet(_)) {
                    return Err(WrongType);
                }
            }
        }

        let old_entry_size = memory::entry_size(key, &self.entries[key].value);
        let entry = self.entries.get_mut(key).expect("verified above");
        let mut removed = Vec::new();
        if let Value::SortedSet(ref mut ss) = entry.value {
            for member in members {
                if ss.remove(member) {
                    removed.push(member.clone());
                }
            }
        }
        entry.touch();

        let is_empty = matches!(&entry.value, Value::SortedSet(ss) if ss.is_empty());
        if is_empty {
            let removed_entry = self.entries.remove(key).expect("verified above");
            self.memory.remove_with_size(old_entry_size);
            if removed_entry.expires_at.is_some() {
                self.expiry_count = self.expiry_count.saturating_sub(1);
            }
        } else {
            let new_entry_size = memory::entry_size(key, &self.entries[key].value);
            self.memory.adjust(old_entry_size, new_entry_size);
        }

        Ok(removed)
    }

    /// Returns the score for a member in a sorted set.
    ///
    /// Returns `Ok(None)` if the key or member doesn't exist.
    /// Returns `Err(WrongType)` on type mismatch.
    pub fn zscore(&mut self, key: &str, member: &str) -> Result<Option<f64>, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(None);
        }
        match self.entries.get_mut(key) {
            None => Ok(None),
            Some(entry) => match &entry.value {
                Value::SortedSet(ss) => {
                    let score = ss.score(member);
                    entry.touch();
                    Ok(score)
                }
                _ => Err(WrongType),
            },
        }
    }

    /// Returns the 0-based rank of a member in a sorted set (lowest score = 0).
    ///
    /// Returns `Ok(None)` if the key or member doesn't exist.
    /// Returns `Err(WrongType)` on type mismatch.
    pub fn zrank(&mut self, key: &str, member: &str) -> Result<Option<usize>, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(None);
        }
        match self.entries.get_mut(key) {
            None => Ok(None),
            Some(entry) => match &entry.value {
                Value::SortedSet(ss) => {
                    let rank = ss.rank(member);
                    entry.touch();
                    Ok(rank)
                }
                _ => Err(WrongType),
            },
        }
    }

    /// Returns a range of members from a sorted set by rank.
    ///
    /// Supports negative indices. If `with_scores` is true, the result
    /// includes `(member, score)` pairs; otherwise just members.
    /// Returns `Err(WrongType)` on type mismatch.
    pub fn zrange(
        &mut self,
        key: &str,
        start: i64,
        stop: i64,
    ) -> Result<Vec<(String, f64)>, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(vec![]);
        }
        match self.entries.get_mut(key) {
            None => Ok(vec![]),
            Some(entry) => {
                let result = match &entry.value {
                    Value::SortedSet(ss) => {
                        let items = ss.range_by_rank(start, stop);
                        Ok(items.into_iter().map(|(m, s)| (m.to_owned(), s)).collect())
                    }
                    _ => Err(WrongType),
                };
                if result.is_ok() {
                    entry.touch();
                }
                result
            }
        }
    }

    /// Returns the number of members in a sorted set, or 0 if the key doesn't exist.
    ///
    /// Returns `Err(WrongType)` on type mismatch.
    pub fn zcard(&mut self, key: &str) -> Result<usize, WrongType> {
        if self.remove_if_expired(key) {
            return Ok(0);
        }
        match self.entries.get(key) {
            None => Ok(0),
            Some(entry) => match &entry.value {
                Value::SortedSet(ss) => Ok(ss.len()),
                _ => Err(WrongType),
            },
        }
    }

    /// Randomly samples up to `count` keys and removes any that have expired.
    ///
    /// Returns the number of keys actually removed. Used by the active
    /// expiration cycle to clean up keys that no one is reading.
    pub fn expire_sample(&mut self, count: usize) -> usize {
        if self.entries.is_empty() {
            return 0;
        }

        let mut rng = rand::rng();

        let keys_to_check: Vec<String> = self
            .entries
            .keys()
            .choose_multiple(&mut rng, count)
            .into_iter()
            .cloned()
            .collect();

        let mut removed = 0;
        for key in &keys_to_check {
            if self.remove_if_expired(key) {
                removed += 1;
            }
        }
        removed
    }

    /// Checks if a key is expired and removes it if so. Returns `true`
    /// if the key was removed (or didn't exist).
    fn remove_if_expired(&mut self, key: &str) -> bool {
        let expired = self
            .entries
            .get(key)
            .map(|e| e.is_expired())
            .unwrap_or(false);

        if expired {
            if let Some(entry) = self.entries.remove(key) {
                self.memory.remove(key, &entry.value);
                if entry.expires_at.is_some() {
                    self.expiry_count = self.expiry_count.saturating_sub(1);
                }
            }
        }
        expired
    }
}

impl Default for Keyspace {
    fn default() -> Self {
        Self::new()
    }
}

/// Glob-style pattern matching for SCAN's MATCH option.
///
/// Supports:
/// - `*` matches any sequence of characters (including empty)
/// - `?` matches exactly one character
/// - `[abc]` matches one character from the set
/// - `[^abc]` or `[!abc]` matches one character NOT in the set
///
/// Uses an iterative two-pointer algorithm with backtracking for O(n*m)
/// worst-case performance, where n is pattern length and m is text length.
fn glob_match(pattern: &str, text: &str) -> bool {
    let pat: Vec<char> = pattern.chars().collect();
    let txt: Vec<char> = text.chars().collect();

    let mut pi = 0; // pattern index
    let mut ti = 0; // text index

    // backtracking state for the most recent '*'
    let mut star_pi: Option<usize> = None;
    let mut star_ti: usize = 0;

    while ti < txt.len() || pi < pat.len() {
        if pi < pat.len() {
            match pat[pi] {
                '*' => {
                    // record star position and try matching zero chars first
                    star_pi = Some(pi);
                    star_ti = ti;
                    pi += 1;
                    continue;
                }
                '?' if ti < txt.len() => {
                    pi += 1;
                    ti += 1;
                    continue;
                }
                '[' if ti < txt.len() => {
                    // parse character class
                    let tc = txt[ti];
                    let mut j = pi + 1;
                    let mut negated = false;
                    let mut matched = false;

                    if j < pat.len() && (pat[j] == '^' || pat[j] == '!') {
                        negated = true;
                        j += 1;
                    }

                    while j < pat.len() && pat[j] != ']' {
                        if pat[j] == tc {
                            matched = true;
                        }
                        j += 1;
                    }

                    if negated {
                        matched = !matched;
                    }

                    if matched && j < pat.len() {
                        pi = j + 1; // skip past ']'
                        ti += 1;
                        continue;
                    }
                    // fall through to backtrack
                }
                c if ti < txt.len() && c == txt[ti] => {
                    pi += 1;
                    ti += 1;
                    continue;
                }
                _ => {}
            }
        }

        // mismatch or end of pattern — try backtracking to last '*'
        if let Some(sp) = star_pi {
            pi = sp + 1;
            star_ti += 1;
            ti = star_ti;
            if ti > txt.len() {
                return false;
            }
        } else {
            return false;
        }
    }

    // skip trailing '*' in pattern
    while pi < pat.len() && pat[pi] == '*' {
        pi += 1;
    }

    pi == pat.len()
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::thread;

    #[test]
    fn set_and_get() {
        let mut ks = Keyspace::new();
        ks.set("hello".into(), Bytes::from("world"), None);
        assert_eq!(
            ks.get("hello").unwrap(),
            Some(Value::String(Bytes::from("world")))
        );
    }

    #[test]
    fn get_missing_key() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.get("nope").unwrap(), None);
    }

    #[test]
    fn overwrite_replaces_value() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("first"), None);
        ks.set("key".into(), Bytes::from("second"), None);
        assert_eq!(
            ks.get("key").unwrap(),
            Some(Value::String(Bytes::from("second")))
        );
    }

    #[test]
    fn overwrite_clears_old_ttl() {
        let mut ks = Keyspace::new();
        ks.set(
            "key".into(),
            Bytes::from("v1"),
            Some(Duration::from_secs(100)),
        );
        // overwrite without TTL — should clear the old one
        ks.set("key".into(), Bytes::from("v2"), None);
        assert_eq!(ks.ttl("key"), TtlResult::NoExpiry);
    }

    #[test]
    fn del_existing() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("val"), None);
        assert!(ks.del("key"));
        assert_eq!(ks.get("key").unwrap(), None);
    }

    #[test]
    fn del_missing() {
        let mut ks = Keyspace::new();
        assert!(!ks.del("nope"));
    }

    #[test]
    fn exists_present_and_absent() {
        let mut ks = Keyspace::new();
        ks.set("yes".into(), Bytes::from("here"), None);
        assert!(ks.exists("yes"));
        assert!(!ks.exists("no"));
    }

    #[test]
    fn expired_key_returns_none() {
        let mut ks = Keyspace::new();
        ks.set(
            "temp".into(),
            Bytes::from("gone"),
            Some(Duration::from_millis(10)),
        );
        // wait for expiration
        thread::sleep(Duration::from_millis(30));
        assert_eq!(ks.get("temp").unwrap(), None);
        // should also be gone from exists
        assert!(!ks.exists("temp"));
    }

    #[test]
    fn ttl_no_expiry() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("val"), None);
        assert_eq!(ks.ttl("key"), TtlResult::NoExpiry);
    }

    #[test]
    fn ttl_not_found() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.ttl("missing"), TtlResult::NotFound);
    }

    #[test]
    fn ttl_with_expiry() {
        let mut ks = Keyspace::new();
        ks.set(
            "key".into(),
            Bytes::from("val"),
            Some(Duration::from_secs(100)),
        );
        match ks.ttl("key") {
            TtlResult::Seconds(s) => assert!(s >= 98 && s <= 100),
            other => panic!("expected Seconds, got {other:?}"),
        }
    }

    #[test]
    fn ttl_expired_key() {
        let mut ks = Keyspace::new();
        ks.set(
            "temp".into(),
            Bytes::from("val"),
            Some(Duration::from_millis(10)),
        );
        thread::sleep(Duration::from_millis(30));
        assert_eq!(ks.ttl("temp"), TtlResult::NotFound);
    }

    #[test]
    fn expire_existing_key() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("val"), None);
        assert!(ks.expire("key", 60));
        match ks.ttl("key") {
            TtlResult::Seconds(s) => assert!(s >= 58 && s <= 60),
            other => panic!("expected Seconds, got {other:?}"),
        }
    }

    #[test]
    fn expire_missing_key() {
        let mut ks = Keyspace::new();
        assert!(!ks.expire("nope", 60));
    }

    #[test]
    fn del_expired_key_returns_false() {
        let mut ks = Keyspace::new();
        ks.set(
            "temp".into(),
            Bytes::from("val"),
            Some(Duration::from_millis(10)),
        );
        thread::sleep(Duration::from_millis(30));
        // key is expired, del should return false (not found)
        assert!(!ks.del("temp"));
    }

    // -- memory tracking tests --

    #[test]
    fn memory_increases_on_set() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.stats().used_bytes, 0);
        ks.set("key".into(), Bytes::from("value"), None);
        assert!(ks.stats().used_bytes > 0);
        assert_eq!(ks.stats().key_count, 1);
    }

    #[test]
    fn memory_decreases_on_del() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("value"), None);
        let after_set = ks.stats().used_bytes;
        ks.del("key");
        assert_eq!(ks.stats().used_bytes, 0);
        assert!(after_set > 0);
    }

    #[test]
    fn memory_adjusts_on_overwrite() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("short"), None);
        let small = ks.stats().used_bytes;

        ks.set("key".into(), Bytes::from("a much longer value"), None);
        let large = ks.stats().used_bytes;

        assert!(large > small);
        assert_eq!(ks.stats().key_count, 1);
    }

    #[test]
    fn memory_decreases_on_expired_removal() {
        let mut ks = Keyspace::new();
        ks.set(
            "temp".into(),
            Bytes::from("data"),
            Some(Duration::from_millis(10)),
        );
        assert!(ks.stats().used_bytes > 0);
        thread::sleep(Duration::from_millis(30));
        // trigger lazy expiration
        let _ = ks.get("temp");
        assert_eq!(ks.stats().used_bytes, 0);
        assert_eq!(ks.stats().key_count, 0);
    }

    #[test]
    fn stats_tracks_expiry_count() {
        let mut ks = Keyspace::new();
        ks.set("a".into(), Bytes::from("1"), None);
        ks.set("b".into(), Bytes::from("2"), Some(Duration::from_secs(100)));
        ks.set("c".into(), Bytes::from("3"), Some(Duration::from_secs(200)));

        let stats = ks.stats();
        assert_eq!(stats.key_count, 3);
        assert_eq!(stats.keys_with_expiry, 2);
    }

    // -- eviction tests --

    #[test]
    fn noeviction_returns_oom_when_full() {
        // one entry with key "a" + value "val" = 1 + 3 + 96 = 100 bytes
        // set limit so one entry fits but two don't
        let config = ShardConfig {
            max_memory: Some(150),
            eviction_policy: EvictionPolicy::NoEviction,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        // first key should fit
        assert_eq!(ks.set("a".into(), Bytes::from("val"), None), SetResult::Ok);

        // second key should push us over the limit
        let result = ks.set("b".into(), Bytes::from("val"), None);
        assert_eq!(result, SetResult::OutOfMemory);

        // original key should still be there
        assert!(ks.exists("a"));
    }

    #[test]
    fn lru_eviction_makes_room() {
        let config = ShardConfig {
            max_memory: Some(150),
            eviction_policy: EvictionPolicy::AllKeysLru,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        assert_eq!(ks.set("a".into(), Bytes::from("val"), None), SetResult::Ok);

        // this should trigger eviction of "a" to make room
        assert_eq!(ks.set("b".into(), Bytes::from("val"), None), SetResult::Ok);

        // "a" should have been evicted
        assert!(!ks.exists("a"));
        assert!(ks.exists("b"));
    }

    #[test]
    fn overwrite_same_size_succeeds_at_limit() {
        let config = ShardConfig {
            max_memory: Some(150),
            eviction_policy: EvictionPolicy::NoEviction,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        assert_eq!(ks.set("a".into(), Bytes::from("val"), None), SetResult::Ok);

        // overwriting with same-size value should succeed — no net increase
        assert_eq!(ks.set("a".into(), Bytes::from("new"), None), SetResult::Ok);
        assert_eq!(
            ks.get("a").unwrap(),
            Some(Value::String(Bytes::from("new")))
        );
    }

    #[test]
    fn overwrite_larger_value_respects_limit() {
        let config = ShardConfig {
            max_memory: Some(150),
            eviction_policy: EvictionPolicy::NoEviction,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        assert_eq!(ks.set("a".into(), Bytes::from("val"), None), SetResult::Ok);

        // overwriting with a much larger value should fail if it exceeds limit
        let big_value = "x".repeat(200);
        let result = ks.set("a".into(), Bytes::from(big_value), None);
        assert_eq!(result, SetResult::OutOfMemory);

        // original value should still be intact
        assert_eq!(
            ks.get("a").unwrap(),
            Some(Value::String(Bytes::from("val")))
        );
    }

    // -- iter_entries tests --

    #[test]
    fn iter_entries_returns_live_entries() {
        let mut ks = Keyspace::new();
        ks.set("a".into(), Bytes::from("1"), None);
        ks.set("b".into(), Bytes::from("2"), Some(Duration::from_secs(100)));

        let entries: Vec<_> = ks.iter_entries().collect();
        assert_eq!(entries.len(), 2);
    }

    #[test]
    fn iter_entries_skips_expired() {
        let mut ks = Keyspace::new();
        ks.set(
            "dead".into(),
            Bytes::from("gone"),
            Some(Duration::from_millis(1)),
        );
        ks.set("alive".into(), Bytes::from("here"), None);
        thread::sleep(Duration::from_millis(10));

        let entries: Vec<_> = ks.iter_entries().collect();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].0, "alive");
    }

    #[test]
    fn iter_entries_ttl_for_no_expiry() {
        let mut ks = Keyspace::new();
        ks.set("permanent".into(), Bytes::from("val"), None);

        let entries: Vec<_> = ks.iter_entries().collect();
        assert_eq!(entries[0].2, -1);
    }

    // -- restore tests --

    #[test]
    fn restore_adds_entry() {
        let mut ks = Keyspace::new();
        ks.restore("restored".into(), Value::String(Bytes::from("data")), None);
        assert_eq!(
            ks.get("restored").unwrap(),
            Some(Value::String(Bytes::from("data")))
        );
        assert_eq!(ks.stats().key_count, 1);
    }

    #[test]
    fn restore_skips_past_deadline() {
        let mut ks = Keyspace::new();
        // deadline already passed
        let past = Instant::now() - Duration::from_secs(1);
        ks.restore(
            "expired".into(),
            Value::String(Bytes::from("old")),
            Some(past),
        );
        assert!(ks.is_empty());
    }

    #[test]
    fn restore_overwrites_existing() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("old"), None);
        ks.restore("key".into(), Value::String(Bytes::from("new")), None);
        assert_eq!(
            ks.get("key").unwrap(),
            Some(Value::String(Bytes::from("new")))
        );
        assert_eq!(ks.stats().key_count, 1);
    }

    #[test]
    fn restore_bypasses_memory_limit() {
        let config = ShardConfig {
            max_memory: Some(50), // very small
            eviction_policy: EvictionPolicy::NoEviction,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        // normal set would fail due to memory limit
        ks.restore(
            "big".into(),
            Value::String(Bytes::from("x".repeat(200))),
            None,
        );
        assert_eq!(ks.stats().key_count, 1);
    }

    #[test]
    fn no_limit_never_rejects() {
        // default config has no memory limit
        let mut ks = Keyspace::new();
        for i in 0..100 {
            assert_eq!(
                ks.set(format!("key:{i}"), Bytes::from("value"), None),
                SetResult::Ok
            );
        }
        assert_eq!(ks.len(), 100);
    }

    // -- list tests --

    #[test]
    fn lpush_creates_list() {
        let mut ks = Keyspace::new();
        let len = ks
            .lpush("list", &[Bytes::from("a"), Bytes::from("b")])
            .unwrap();
        assert_eq!(len, 2);
        // lpush pushes each to front, so order is b, a
        let items = ks.lrange("list", 0, -1).unwrap();
        assert_eq!(items, vec![Bytes::from("b"), Bytes::from("a")]);
    }

    #[test]
    fn rpush_creates_list() {
        let mut ks = Keyspace::new();
        let len = ks
            .rpush("list", &[Bytes::from("a"), Bytes::from("b")])
            .unwrap();
        assert_eq!(len, 2);
        let items = ks.lrange("list", 0, -1).unwrap();
        assert_eq!(items, vec![Bytes::from("a"), Bytes::from("b")]);
    }

    #[test]
    fn push_to_existing_list() {
        let mut ks = Keyspace::new();
        ks.rpush("list", &[Bytes::from("a")]).unwrap();
        let len = ks.rpush("list", &[Bytes::from("b")]).unwrap();
        assert_eq!(len, 2);
    }

    #[test]
    fn lpop_returns_front() {
        let mut ks = Keyspace::new();
        ks.rpush("list", &[Bytes::from("a"), Bytes::from("b")])
            .unwrap();
        assert_eq!(ks.lpop("list").unwrap(), Some(Bytes::from("a")));
        assert_eq!(ks.lpop("list").unwrap(), Some(Bytes::from("b")));
        assert_eq!(ks.lpop("list").unwrap(), None); // empty, key deleted
    }

    #[test]
    fn rpop_returns_back() {
        let mut ks = Keyspace::new();
        ks.rpush("list", &[Bytes::from("a"), Bytes::from("b")])
            .unwrap();
        assert_eq!(ks.rpop("list").unwrap(), Some(Bytes::from("b")));
    }

    #[test]
    fn pop_from_missing_key() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.lpop("nope").unwrap(), None);
        assert_eq!(ks.rpop("nope").unwrap(), None);
    }

    #[test]
    fn empty_list_auto_deletes_key() {
        let mut ks = Keyspace::new();
        ks.rpush("list", &[Bytes::from("only")]).unwrap();
        ks.lpop("list").unwrap();
        assert!(!ks.exists("list"));
        assert_eq!(ks.stats().key_count, 0);
        assert_eq!(ks.stats().used_bytes, 0);
    }

    #[test]
    fn lrange_negative_indices() {
        let mut ks = Keyspace::new();
        ks.rpush(
            "list",
            &[Bytes::from("a"), Bytes::from("b"), Bytes::from("c")],
        )
        .unwrap();
        // -2 to -1 => last two elements
        let items = ks.lrange("list", -2, -1).unwrap();
        assert_eq!(items, vec![Bytes::from("b"), Bytes::from("c")]);
    }

    #[test]
    fn lrange_out_of_bounds_clamps() {
        let mut ks = Keyspace::new();
        ks.rpush("list", &[Bytes::from("a"), Bytes::from("b")])
            .unwrap();
        let items = ks.lrange("list", -100, 100).unwrap();
        assert_eq!(items, vec![Bytes::from("a"), Bytes::from("b")]);
    }

    #[test]
    fn lrange_missing_key_returns_empty() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.lrange("nope", 0, -1).unwrap(), Vec::<Bytes>::new());
    }

    #[test]
    fn llen_returns_length() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.llen("nope").unwrap(), 0);
        ks.rpush("list", &[Bytes::from("a"), Bytes::from("b")])
            .unwrap();
        assert_eq!(ks.llen("list").unwrap(), 2);
    }

    #[test]
    fn list_memory_tracked_on_push_pop() {
        let mut ks = Keyspace::new();
        ks.rpush("list", &[Bytes::from("hello")]).unwrap();
        let after_push = ks.stats().used_bytes;
        assert!(after_push > 0);

        ks.rpush("list", &[Bytes::from("world")]).unwrap();
        let after_second = ks.stats().used_bytes;
        assert!(after_second > after_push);

        ks.lpop("list").unwrap();
        let after_pop = ks.stats().used_bytes;
        assert!(after_pop < after_second);
    }

    #[test]
    fn lpush_on_string_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("val"), None);
        assert!(ks.lpush("s", &[Bytes::from("nope")]).is_err());
    }

    #[test]
    fn lrange_on_string_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("val"), None);
        assert!(ks.lrange("s", 0, -1).is_err());
    }

    #[test]
    fn llen_on_string_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("val"), None);
        assert!(ks.llen("s").is_err());
    }

    // -- wrongtype tests --

    #[test]
    fn get_on_list_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        let mut list = std::collections::VecDeque::new();
        list.push_back(Bytes::from("item"));
        ks.restore("mylist".into(), Value::List(list), None);

        assert!(ks.get("mylist").is_err());
    }

    #[test]
    fn value_type_returns_correct_types() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.value_type("missing"), "none");

        ks.set("s".into(), Bytes::from("val"), None);
        assert_eq!(ks.value_type("s"), "string");

        let mut list = std::collections::VecDeque::new();
        list.push_back(Bytes::from("item"));
        ks.restore("l".into(), Value::List(list), None);
        assert_eq!(ks.value_type("l"), "list");

        ks.zadd("z", &[(1.0, "a".into())], &ZAddFlags::default())
            .unwrap();
        assert_eq!(ks.value_type("z"), "zset");
    }

    // -- sorted set tests --

    #[test]
    fn zadd_creates_sorted_set() {
        let mut ks = Keyspace::new();
        let result = ks
            .zadd(
                "board",
                &[(100.0, "alice".into()), (200.0, "bob".into())],
                &ZAddFlags::default(),
            )
            .unwrap();
        assert_eq!(result.count, 2);
        assert_eq!(result.applied.len(), 2);
        assert_eq!(ks.value_type("board"), "zset");
    }

    #[test]
    fn zadd_updates_existing_score() {
        let mut ks = Keyspace::new();
        ks.zadd("z", &[(100.0, "alice".into())], &ZAddFlags::default())
            .unwrap();
        // update score — default flags don't count updates
        let result = ks
            .zadd("z", &[(200.0, "alice".into())], &ZAddFlags::default())
            .unwrap();
        assert_eq!(result.count, 0);
        // score was updated, so applied should have the member
        assert_eq!(result.applied.len(), 1);
        assert_eq!(ks.zscore("z", "alice").unwrap(), Some(200.0));
    }

    #[test]
    fn zadd_ch_flag_counts_changes() {
        let mut ks = Keyspace::new();
        ks.zadd("z", &[(100.0, "alice".into())], &ZAddFlags::default())
            .unwrap();
        let flags = ZAddFlags {
            ch: true,
            ..Default::default()
        };
        let result = ks
            .zadd(
                "z",
                &[(200.0, "alice".into()), (50.0, "bob".into())],
                &flags,
            )
            .unwrap();
        // 1 updated + 1 added = 2
        assert_eq!(result.count, 2);
        assert_eq!(result.applied.len(), 2);
    }

    #[test]
    fn zadd_nx_skips_existing() {
        let mut ks = Keyspace::new();
        ks.zadd("z", &[(100.0, "alice".into())], &ZAddFlags::default())
            .unwrap();
        let flags = ZAddFlags {
            nx: true,
            ..Default::default()
        };
        let result = ks.zadd("z", &[(999.0, "alice".into())], &flags).unwrap();
        assert_eq!(result.count, 0);
        assert!(result.applied.is_empty());
        assert_eq!(ks.zscore("z", "alice").unwrap(), Some(100.0));
    }

    #[test]
    fn zadd_xx_skips_new() {
        let mut ks = Keyspace::new();
        let flags = ZAddFlags {
            xx: true,
            ..Default::default()
        };
        let result = ks.zadd("z", &[(100.0, "alice".into())], &flags).unwrap();
        assert_eq!(result.count, 0);
        assert!(result.applied.is_empty());
        // key should be cleaned up since nothing was added
        assert_eq!(ks.value_type("z"), "none");
    }

    #[test]
    fn zadd_gt_only_increases() {
        let mut ks = Keyspace::new();
        ks.zadd("z", &[(100.0, "alice".into())], &ZAddFlags::default())
            .unwrap();
        let flags = ZAddFlags {
            gt: true,
            ..Default::default()
        };
        ks.zadd("z", &[(50.0, "alice".into())], &flags).unwrap();
        assert_eq!(ks.zscore("z", "alice").unwrap(), Some(100.0));
        ks.zadd("z", &[(200.0, "alice".into())], &flags).unwrap();
        assert_eq!(ks.zscore("z", "alice").unwrap(), Some(200.0));
    }

    #[test]
    fn zadd_lt_only_decreases() {
        let mut ks = Keyspace::new();
        ks.zadd("z", &[(100.0, "alice".into())], &ZAddFlags::default())
            .unwrap();
        let flags = ZAddFlags {
            lt: true,
            ..Default::default()
        };
        ks.zadd("z", &[(200.0, "alice".into())], &flags).unwrap();
        assert_eq!(ks.zscore("z", "alice").unwrap(), Some(100.0));
        ks.zadd("z", &[(50.0, "alice".into())], &flags).unwrap();
        assert_eq!(ks.zscore("z", "alice").unwrap(), Some(50.0));
    }

    #[test]
    fn zrem_removes_members() {
        let mut ks = Keyspace::new();
        ks.zadd(
            "z",
            &[(1.0, "a".into()), (2.0, "b".into()), (3.0, "c".into())],
            &ZAddFlags::default(),
        )
        .unwrap();
        let removed = ks
            .zrem("z", &["a".into(), "c".into(), "nonexistent".into()])
            .unwrap();
        assert_eq!(removed.len(), 2);
        assert!(removed.contains(&"a".to_owned()));
        assert!(removed.contains(&"c".to_owned()));
        assert_eq!(ks.zscore("z", "a").unwrap(), None);
        assert_eq!(ks.zscore("z", "b").unwrap(), Some(2.0));
    }

    #[test]
    fn zrem_auto_deletes_empty() {
        let mut ks = Keyspace::new();
        ks.zadd("z", &[(1.0, "only".into())], &ZAddFlags::default())
            .unwrap();
        ks.zrem("z", &["only".into()]).unwrap();
        assert!(!ks.exists("z"));
        assert_eq!(ks.stats().key_count, 0);
    }

    #[test]
    fn zrem_missing_key() {
        let mut ks = Keyspace::new();
        assert!(ks.zrem("nope", &["a".into()]).unwrap().is_empty());
    }

    #[test]
    fn zscore_returns_score() {
        let mut ks = Keyspace::new();
        ks.zadd("z", &[(42.5, "member".into())], &ZAddFlags::default())
            .unwrap();
        assert_eq!(ks.zscore("z", "member").unwrap(), Some(42.5));
        assert_eq!(ks.zscore("z", "missing").unwrap(), None);
    }

    #[test]
    fn zscore_missing_key() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.zscore("nope", "m").unwrap(), None);
    }

    #[test]
    fn zrank_returns_rank() {
        let mut ks = Keyspace::new();
        ks.zadd(
            "z",
            &[
                (300.0, "c".into()),
                (100.0, "a".into()),
                (200.0, "b".into()),
            ],
            &ZAddFlags::default(),
        )
        .unwrap();
        assert_eq!(ks.zrank("z", "a").unwrap(), Some(0));
        assert_eq!(ks.zrank("z", "b").unwrap(), Some(1));
        assert_eq!(ks.zrank("z", "c").unwrap(), Some(2));
        assert_eq!(ks.zrank("z", "d").unwrap(), None);
    }

    #[test]
    fn zrange_returns_range() {
        let mut ks = Keyspace::new();
        ks.zadd(
            "z",
            &[(1.0, "a".into()), (2.0, "b".into()), (3.0, "c".into())],
            &ZAddFlags::default(),
        )
        .unwrap();

        let all = ks.zrange("z", 0, -1).unwrap();
        assert_eq!(
            all,
            vec![
                ("a".to_owned(), 1.0),
                ("b".to_owned(), 2.0),
                ("c".to_owned(), 3.0),
            ]
        );

        let middle = ks.zrange("z", 1, 1).unwrap();
        assert_eq!(middle, vec![("b".to_owned(), 2.0)]);

        let last_two = ks.zrange("z", -2, -1).unwrap();
        assert_eq!(last_two, vec![("b".to_owned(), 2.0), ("c".to_owned(), 3.0)]);
    }

    #[test]
    fn zrange_missing_key() {
        let mut ks = Keyspace::new();
        assert!(ks.zrange("nope", 0, -1).unwrap().is_empty());
    }

    #[test]
    fn zadd_on_string_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("val"), None);
        assert!(ks
            .zadd("s", &[(1.0, "m".into())], &ZAddFlags::default())
            .is_err());
    }

    #[test]
    fn zrem_on_string_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("val"), None);
        assert!(ks.zrem("s", &["m".into()]).is_err());
    }

    #[test]
    fn zscore_on_list_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.rpush("l", &[Bytes::from("item")]).unwrap();
        assert!(ks.zscore("l", "m").is_err());
    }

    #[test]
    fn zrank_on_string_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("val"), None);
        assert!(ks.zrank("s", "m").is_err());
    }

    #[test]
    fn zrange_on_string_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("val"), None);
        assert!(ks.zrange("s", 0, -1).is_err());
    }

    #[test]
    fn sorted_set_memory_tracked() {
        let mut ks = Keyspace::new();
        let before = ks.stats().used_bytes;
        ks.zadd("z", &[(1.0, "alice".into())], &ZAddFlags::default())
            .unwrap();
        let after_add = ks.stats().used_bytes;
        assert!(after_add > before);

        ks.zadd("z", &[(2.0, "bob".into())], &ZAddFlags::default())
            .unwrap();
        let after_second = ks.stats().used_bytes;
        assert!(after_second > after_add);

        ks.zrem("z", &["alice".into()]).unwrap();
        let after_remove = ks.stats().used_bytes;
        assert!(after_remove < after_second);
    }

    #[test]
    fn incr_new_key_defaults_to_zero() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.incr("counter").unwrap(), 1);
        // verify the stored value
        match ks.get("counter").unwrap() {
            Some(Value::String(data)) => assert_eq!(data, Bytes::from("1")),
            other => panic!("expected String(\"1\"), got {other:?}"),
        }
    }

    #[test]
    fn incr_existing_value() {
        let mut ks = Keyspace::new();
        ks.set("n".into(), Bytes::from("10"), None);
        assert_eq!(ks.incr("n").unwrap(), 11);
    }

    #[test]
    fn decr_new_key_defaults_to_zero() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.decr("counter").unwrap(), -1);
    }

    #[test]
    fn decr_existing_value() {
        let mut ks = Keyspace::new();
        ks.set("n".into(), Bytes::from("10"), None);
        assert_eq!(ks.decr("n").unwrap(), 9);
    }

    #[test]
    fn incr_non_integer_returns_error() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("notanum"), None);
        assert_eq!(ks.incr("s").unwrap_err(), IncrError::NotAnInteger);
    }

    #[test]
    fn incr_on_list_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.lpush("list", &[Bytes::from("a")]).unwrap();
        assert_eq!(ks.incr("list").unwrap_err(), IncrError::WrongType);
    }

    #[test]
    fn incr_overflow_returns_error() {
        let mut ks = Keyspace::new();
        ks.set("max".into(), Bytes::from(i64::MAX.to_string()), None);
        assert_eq!(ks.incr("max").unwrap_err(), IncrError::Overflow);
    }

    #[test]
    fn decr_overflow_returns_error() {
        let mut ks = Keyspace::new();
        ks.set("min".into(), Bytes::from(i64::MIN.to_string()), None);
        assert_eq!(ks.decr("min").unwrap_err(), IncrError::Overflow);
    }

    #[test]
    fn incr_preserves_ttl() {
        let mut ks = Keyspace::new();
        ks.set("n".into(), Bytes::from("5"), Some(Duration::from_secs(60)));
        ks.incr("n").unwrap();
        match ks.ttl("n") {
            TtlResult::Seconds(s) => assert!(s >= 58 && s <= 60),
            other => panic!("expected TTL preserved, got {other:?}"),
        }
    }

    #[test]
    fn zrem_returns_actually_removed_members() {
        let mut ks = Keyspace::new();
        ks.zadd(
            "z",
            &[(1.0, "a".into()), (2.0, "b".into())],
            &ZAddFlags::default(),
        )
        .unwrap();
        // "a" exists, "ghost" doesn't — only "a" should be in the result
        let removed = ks.zrem("z", &["a".into(), "ghost".into()]).unwrap();
        assert_eq!(removed, vec!["a".to_owned()]);
    }

    #[test]
    fn zcard_returns_count() {
        let mut ks = Keyspace::new();
        ks.zadd(
            "z",
            &[(1.0, "a".into()), (2.0, "b".into())],
            &ZAddFlags::default(),
        )
        .unwrap();
        assert_eq!(ks.zcard("z").unwrap(), 2);
    }

    #[test]
    fn zcard_missing_key_returns_zero() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.zcard("missing").unwrap(), 0);
    }

    #[test]
    fn zcard_on_string_key_returns_wrongtype() {
        let mut ks = Keyspace::new();
        ks.set("s".into(), Bytes::from("val"), None);
        assert!(ks.zcard("s").is_err());
    }

    #[test]
    fn persist_removes_expiry() {
        let mut ks = Keyspace::new();
        ks.set(
            "key".into(),
            Bytes::from("val"),
            Some(Duration::from_secs(60)),
        );
        assert!(matches!(ks.ttl("key"), TtlResult::Seconds(_)));

        assert!(ks.persist("key"));
        assert_eq!(ks.ttl("key"), TtlResult::NoExpiry);
        assert_eq!(ks.stats().keys_with_expiry, 0);
    }

    #[test]
    fn persist_returns_false_without_expiry() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("val"), None);
        assert!(!ks.persist("key"));
    }

    #[test]
    fn persist_returns_false_for_missing_key() {
        let mut ks = Keyspace::new();
        assert!(!ks.persist("missing"));
    }

    #[test]
    fn pttl_returns_milliseconds() {
        let mut ks = Keyspace::new();
        ks.set(
            "key".into(),
            Bytes::from("val"),
            Some(Duration::from_secs(60)),
        );
        match ks.pttl("key") {
            TtlResult::Milliseconds(ms) => assert!(ms > 59_000 && ms <= 60_000),
            other => panic!("expected Milliseconds, got {other:?}"),
        }
    }

    #[test]
    fn pttl_no_expiry() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("val"), None);
        assert_eq!(ks.pttl("key"), TtlResult::NoExpiry);
    }

    #[test]
    fn pttl_not_found() {
        let mut ks = Keyspace::new();
        assert_eq!(ks.pttl("missing"), TtlResult::NotFound);
    }

    #[test]
    fn pexpire_sets_ttl_in_millis() {
        let mut ks = Keyspace::new();
        ks.set("key".into(), Bytes::from("val"), None);
        assert!(ks.pexpire("key", 5000));
        match ks.pttl("key") {
            TtlResult::Milliseconds(ms) => assert!(ms > 4000 && ms <= 5000),
            other => panic!("expected Milliseconds, got {other:?}"),
        }
        assert_eq!(ks.stats().keys_with_expiry, 1);
    }

    #[test]
    fn pexpire_missing_key_returns_false() {
        let mut ks = Keyspace::new();
        assert!(!ks.pexpire("missing", 5000));
    }

    #[test]
    fn pexpire_overwrites_existing_ttl() {
        let mut ks = Keyspace::new();
        ks.set(
            "key".into(),
            Bytes::from("val"),
            Some(Duration::from_secs(60)),
        );
        assert!(ks.pexpire("key", 500));
        match ks.pttl("key") {
            TtlResult::Milliseconds(ms) => assert!(ms <= 500),
            other => panic!("expected Milliseconds, got {other:?}"),
        }
        // expiry count shouldn't double-count
        assert_eq!(ks.stats().keys_with_expiry, 1);
    }

    // -- memory limit enforcement for list/zset --

    #[test]
    fn lpush_rejects_when_memory_full() {
        let config = ShardConfig {
            max_memory: Some(150),
            eviction_policy: EvictionPolicy::NoEviction,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        // first key eats up most of the budget
        assert_eq!(ks.set("a".into(), Bytes::from("val"), None), SetResult::Ok);

        // lpush should be rejected — not enough room
        let result = ks.lpush("list", &[Bytes::from("big-value-here")]);
        assert_eq!(result, Err(WriteError::OutOfMemory));

        // original key should be untouched
        assert!(ks.exists("a"));
    }

    #[test]
    fn rpush_rejects_when_memory_full() {
        let config = ShardConfig {
            max_memory: Some(150),
            eviction_policy: EvictionPolicy::NoEviction,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        assert_eq!(ks.set("a".into(), Bytes::from("val"), None), SetResult::Ok);

        let result = ks.rpush("list", &[Bytes::from("big-value-here")]);
        assert_eq!(result, Err(WriteError::OutOfMemory));
    }

    #[test]
    fn zadd_rejects_when_memory_full() {
        let config = ShardConfig {
            max_memory: Some(150),
            eviction_policy: EvictionPolicy::NoEviction,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        assert_eq!(ks.set("a".into(), Bytes::from("val"), None), SetResult::Ok);

        let result = ks.zadd("z", &[(1.0, "member".into())], &ZAddFlags::default());
        assert!(matches!(result, Err(WriteError::OutOfMemory)));

        // original key should be untouched
        assert!(ks.exists("a"));
    }

    #[test]
    fn lpush_evicts_under_lru_policy() {
        let config = ShardConfig {
            max_memory: Some(200),
            eviction_policy: EvictionPolicy::AllKeysLru,
            ..ShardConfig::default()
        };
        let mut ks = Keyspace::with_config(config);

        assert_eq!(ks.set("a".into(), Bytes::from("val"), None), SetResult::Ok);

        // should evict "a" to make room for the list
        assert!(ks.lpush("list", &[Bytes::from("item")]).is_ok());
        assert!(!ks.exists("a"));
    }

    #[test]
    fn clear_removes_all_keys() {
        let mut ks = Keyspace::new();
        ks.set("a".into(), Bytes::from("1"), None);
        ks.set("b".into(), Bytes::from("2"), Some(Duration::from_secs(60)));
        ks.lpush("list", &[Bytes::from("x")]).unwrap();

        assert_eq!(ks.len(), 3);
        assert!(ks.stats().used_bytes > 0);
        assert_eq!(ks.stats().keys_with_expiry, 1);

        ks.clear();

        assert_eq!(ks.len(), 0);
        assert!(ks.is_empty());
        assert_eq!(ks.stats().used_bytes, 0);
        assert_eq!(ks.stats().keys_with_expiry, 0);
    }

    // --- scan ---

    #[test]
    fn scan_returns_keys() {
        let mut ks = Keyspace::new();
        ks.set("key1".into(), Bytes::from("a"), None);
        ks.set("key2".into(), Bytes::from("b"), None);
        ks.set("key3".into(), Bytes::from("c"), None);

        let (cursor, keys) = ks.scan_keys(0, 10, None);
        assert_eq!(cursor, 0); // complete in one pass
        assert_eq!(keys.len(), 3);
    }

    #[test]
    fn scan_empty_keyspace() {
        let ks = Keyspace::new();
        let (cursor, keys) = ks.scan_keys(0, 10, None);
        assert_eq!(cursor, 0);
        assert!(keys.is_empty());
    }

    #[test]
    fn scan_with_pattern() {
        let mut ks = Keyspace::new();
        ks.set("user:1".into(), Bytes::from("a"), None);
        ks.set("user:2".into(), Bytes::from("b"), None);
        ks.set("item:1".into(), Bytes::from("c"), None);

        let (cursor, keys) = ks.scan_keys(0, 10, Some("user:*"));
        assert_eq!(cursor, 0);
        assert_eq!(keys.len(), 2);
        for k in &keys {
            assert!(k.starts_with("user:"));
        }
    }

    #[test]
    fn scan_with_count_limit() {
        let mut ks = Keyspace::new();
        for i in 0..10 {
            ks.set(format!("k{i}"), Bytes::from("v"), None);
        }

        // first batch
        let (cursor, keys) = ks.scan_keys(0, 3, None);
        assert!(!keys.is_empty());
        assert!(keys.len() <= 3);

        // if there are more keys, cursor should be non-zero
        if cursor != 0 {
            let (cursor2, keys2) = ks.scan_keys(cursor, 3, None);
            assert!(!keys2.is_empty());
            // continue until complete
            let _ = (cursor2, keys2);
        }
    }

    #[test]
    fn scan_skips_expired_keys() {
        let mut ks = Keyspace::new();
        ks.set("live".into(), Bytes::from("a"), None);
        ks.set(
            "expired".into(),
            Bytes::from("b"),
            Some(Duration::from_millis(1)),
        );

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

        let (_, keys) = ks.scan_keys(0, 10, None);
        assert_eq!(keys.len(), 1);
        assert_eq!(keys[0], "live");
    }

    #[test]
    fn glob_match_star() {
        assert!(super::glob_match("user:*", "user:123"));
        assert!(super::glob_match("user:*", "user:"));
        assert!(super::glob_match("*:data", "foo:data"));
        assert!(!super::glob_match("user:*", "item:123"));
    }

    #[test]
    fn glob_match_question() {
        assert!(super::glob_match("key?", "key1"));
        assert!(super::glob_match("key?", "keya"));
        assert!(!super::glob_match("key?", "key"));
        assert!(!super::glob_match("key?", "key12"));
    }

    #[test]
    fn glob_match_brackets() {
        assert!(super::glob_match("key[abc]", "keya"));
        assert!(super::glob_match("key[abc]", "keyb"));
        assert!(!super::glob_match("key[abc]", "keyd"));
    }

    #[test]
    fn glob_match_literal() {
        assert!(super::glob_match("exact", "exact"));
        assert!(!super::glob_match("exact", "exactnot"));
        assert!(!super::glob_match("exact", "notexact"));
    }
}