feat: EXPIRETIME, PEXPIRETIME, SMOVE, and SINTERCARD

crates/ember-core/src/keyspace/mod.rs

@@ -966,6 +966,38 @@ impl Keyspace {
         }
     }
 
+    /// Returns the absolute Unix timestamp (seconds) when the key expires.
+    ///
+    /// Returns `-2` if the key doesn't exist, `-1` if it has no expiry.
+    pub fn expiretime(&mut self, key: &str) -> i64 {
+        if self.remove_if_expired(key) {
+            return -2;
+        }
+        match self.entries.get(key) {
+            None => -2,
+            Some(entry) => match time::monotonic_to_unix_ms(entry.expires_at_ms) {
+                None => -1,
+                Some(unix_ms) => (unix_ms / 1000) as i64,
+            },
+        }
+    }
+
+    /// Returns the absolute Unix timestamp (milliseconds) when the key expires.
+    ///
+    /// Returns `-2` if the key doesn't exist, `-1` if it has no expiry.
+    pub fn pexpiretime(&mut self, key: &str) -> i64 {
+        if self.remove_if_expired(key) {
+            return -2;
+        }
+        match self.entries.get(key) {
+            None => -2,
+            Some(entry) => match time::monotonic_to_unix_ms(entry.expires_at_ms) {
+                None => -1,
+                Some(unix_ms) => unix_ms as i64,
+            },
+        }
+    }
+
     /// Returns all keys matching a glob pattern.
     ///
     /// Warning: O(n) scan of the entire keyspace. Use SCAN for production

crates/ember-core/src/keyspace/set.rs

@@ -457,6 +457,92 @@ impl Keyspace {
         }
     }
 
+    /// Atomically moves `member` from `source` set to `destination` set.
+    ///
+    /// Returns `true` if the member was moved, `false` if it wasn't in the source
+    /// set. Returns an error if either key holds a non-set value.
+    ///
+    /// Both keys must hash to the same shard. The caller is responsible for
+    /// enforcing that constraint before routing to this method.
+    pub fn smove(
+        &mut self,
+        source: &str,
+        destination: &str,
+        member: &str,
+    ) -> Result<bool, WriteError> {
+        self.remove_if_expired(source);
+        self.remove_if_expired(destination);
+
+        // type-check source
+        match self.entries.get(source) {
+            None => return Ok(false),
+            Some(entry) => {
+                if !matches!(&entry.value, Value::Set(_)) {
+                    return Err(WriteError::WrongType);
+                }
+            }
+        }
+
+        // type-check destination if it exists
+        if let Some(entry) = self.entries.get(destination) {
+            if !matches!(&entry.value, Value::Set(_)) {
+                return Err(WriteError::WrongType);
+            }
+        }
+
+        let old_src_size = self
+            .entries
+            .get(source)
+            .map(|e| e.entry_size(source))
+            .unwrap_or(0);
+
+        // try removing the member from source
+        let removed = if let Some(entry) = self.entries.get_mut(source) {
+            if let Value::Set(ref mut set) = entry.value {
+                set.remove(member)
+            } else {
+                false
+            }
+        } else {
+            false
+        };
+
+        if !removed {
+            return Ok(false);
+        }
+
+        let member_bytes = member.len() + memory::HASHSET_MEMBER_OVERHEAD;
+
+        let src_empty = self
+            .entries
+            .get(source)
+            .map(|e| matches!(&e.value, Value::Set(s) if s.is_empty()))
+            .unwrap_or(false);
+
+        self.cleanup_after_remove(source, old_src_size, src_empty, member_bytes);
+        self.bump_version(source);
+
+        // add to destination (creates the set if needed)
+        self.sadd(destination, &[member.to_string()])?;
+
+        Ok(true)
+    }
+
+    /// Returns the cardinality of the intersection of all given sets.
+    ///
+    /// If `limit` is nonzero, the count is capped at that value.
+    /// Returns 0 if any key is missing. Returns an error if any key
+    /// holds a non-set type.
+    pub fn sintercard(&mut self, keys: &[String], limit: usize) -> Result<usize, WrongType> {
+        let members = self.sinter(keys)?;
+        let count = if limit > 0 {
+            members.len().min(limit)
+        } else {
+            members.len()
+        };
+        Ok(count)
+    }
+
     /// Returns the cardinality (number of elements) of a set.
     pub fn scard(&mut self, key: &str) -> Result<usize, WrongType> {
         if self.remove_if_expired(key) {
@@ -931,4 +1017,108 @@ mod tests {
         assert_eq!(ks.len(), 0);
         assert!(!ks.exists("s"));
     }
+
+    // --- smove ---
+
+    #[test]
+    fn smove_moves_member() {
+        let mut ks = Keyspace::new();
+        ks.sadd("src", &["a".into(), "b".into()]).unwrap();
+
+        let moved = ks.smove("src", "dst", "a").unwrap();
+        assert!(moved);
+        assert!(!ks.sismember("src", "a").unwrap());
+        assert!(ks.sismember("dst", "a").unwrap());
+        assert_eq!(ks.scard("src").unwrap(), 1);
+        assert_eq!(ks.scard("dst").unwrap(), 1);
+    }
+
+    #[test]
+    fn smove_missing_member_returns_false() {
+        let mut ks = Keyspace::new();
+        ks.sadd("src", &["x".into()]).unwrap();
+
+        let moved = ks.smove("src", "dst", "missing").unwrap();
+        assert!(!moved);
+        assert_eq!(ks.scard("src").unwrap(), 1);
+        assert!(!ks.exists("dst"));
+    }
+
+    #[test]
+    fn smove_missing_source_returns_false() {
+        let mut ks = Keyspace::new();
+        let moved = ks.smove("nosrc", "dst", "m").unwrap();
+        assert!(!moved);
+    }
+
+    #[test]
+    fn smove_removes_empty_source() {
+        let mut ks = Keyspace::new();
+        ks.sadd("src", &["only".into()]).unwrap();
+
+        ks.smove("src", "dst", "only").unwrap();
+        // source set is auto-deleted when it becomes empty
+        assert!(!ks.exists("src"));
+        assert_eq!(ks.scard("dst").unwrap(), 1);
+    }
+
+    #[test]
+    fn smove_wrong_type_source_returns_error() {
+        let mut ks = Keyspace::new();
+        ks.set("src".into(), Bytes::from("string"), None, false, false);
+        assert!(ks.smove("src", "dst", "m").is_err());
+    }
+
+    #[test]
+    fn smove_wrong_type_destination_returns_error() {
+        let mut ks = Keyspace::new();
+        ks.sadd("src", &["m".into()]).unwrap();
+        ks.set("dst".into(), Bytes::from("string"), None, false, false);
+        assert!(ks.smove("src", "dst", "m").is_err());
+    }
+
+    // --- sintercard ---
+
+    #[test]
+    fn sintercard_basic() {
+        let mut ks = Keyspace::new();
+        ks.sadd("s1", &["a".into(), "b".into(), "c".into()])
+            .unwrap();
+        ks.sadd("s2", &["b".into(), "c".into(), "d".into()])
+            .unwrap();
+
+        assert_eq!(ks.sintercard(&["s1".into(), "s2".into()], 0).unwrap(), 2);
+    }
+
+    #[test]
+    fn sintercard_with_limit() {
+        let mut ks = Keyspace::new();
+        ks.sadd("s1", &["a".into(), "b".into(), "c".into()])
+            .unwrap();
+        ks.sadd("s2", &["a".into(), "b".into(), "c".into()])
+            .unwrap();
+
+        // limit caps the result
+        assert_eq!(ks.sintercard(&["s1".into(), "s2".into()], 2).unwrap(), 2);
+        // limit 0 means no cap
+        assert_eq!(ks.sintercard(&["s1".into(), "s2".into()], 0).unwrap(), 3);
+    }
+
+    #[test]
+    fn sintercard_missing_key_returns_zero() {
+        let mut ks = Keyspace::new();
+        ks.sadd("s1", &["a".into()]).unwrap();
+
+        assert_eq!(
+            ks.sintercard(&["s1".into(), "missing".into()], 0).unwrap(),
+            0
+        );
+    }
+
+    #[test]
+    fn sintercard_wrong_type_returns_error() {
+        let mut ks = Keyspace::new();
+        ks.set("str".into(), Bytes::from("val"), None, false, false);
+        assert!(ks.sintercard(&["str".into()], 0).is_err());
+    }
 }

crates/ember-core/src/shard/aof.rs

@@ -197,6 +197,24 @@ pub(super) fn to_aof_records(
                 members: members.clone(),
             }]
         }
+        // SMOVE: persist as SREM from source + SADD to destination
+        (
+            ShardRequest::SMove {
+                source,
+                destination,
+                member,
+            },
+            ShardResponse::Bool(true),
+        ) => smallvec![
+            AofRecord::SRem {
+                key: source,
+                members: vec![member.clone()],
+            },
+            AofRecord::SAdd {
+                key: destination,
+                members: vec![member],
+            },
+        ],
         // STORE commands: persist as DEL + SADD with the resulting members
         (
             ShardRequest::SUnionStore { dest, .. }

crates/ember-core/src/shard/mod.rs

@@ -445,6 +445,25 @@ pub enum ShardRequest {
         key: String,
         members: Vec<String>,
     },
+    /// SMOVE — atomically moves a member between two sets on the same shard.
+    SMove {
+        source: String,
+        destination: String,
+        member: String,
+    },
+    /// SINTERCARD — returns cardinality of set intersection, capped at limit (0 = no limit).
+    SInterCard {
+        keys: Vec<String>,
+        limit: usize,
+    },
+    /// EXPIRETIME — returns the absolute expiry timestamp in seconds (-1 or -2 for missing/no-expiry).
+    Expiretime {
+        key: String,
+    },
+    /// PEXPIRETIME — returns the absolute expiry timestamp in milliseconds (-1 or -2 for missing/no-expiry).
+    Pexpiretime {
+        key: String,
+    },
     /// LMOVE: atomically pops from source and pushes to destination.
     LMove {
         source: String,
@@ -699,6 +718,7 @@ impl ShardRequest {
             | ShardRequest::SUnionStore { .. }
             | ShardRequest::SInterStore { .. }
             | ShardRequest::SDiffStore { .. }
+            | ShardRequest::SMove { .. }
             | ShardRequest::LMove { .. }
             | ShardRequest::GetDel { .. }
             | ShardRequest::GetEx { .. }
@@ -1899,6 +1919,21 @@ fn dispatch(
             Ok(results) => ShardResponse::BoolArray(results),
             Err(_) => ShardResponse::WrongType,
         },
+        ShardRequest::SMove {
+            source,
+            destination,
+            member,
+        } => match ks.smove(source, destination, member) {
+            Ok(moved) => ShardResponse::Bool(moved),
+            Err(WriteError::WrongType) => ShardResponse::WrongType,
+            Err(WriteError::OutOfMemory) => ShardResponse::OutOfMemory,
+        },
+        ShardRequest::SInterCard { keys, limit } => match ks.sintercard(keys, *limit) {
+            Ok(n) => ShardResponse::Integer(n as i64),
+            Err(_) => ShardResponse::WrongType,
+        },
+        ShardRequest::Expiretime { key } => ShardResponse::Integer(ks.expiretime(key)),
+        ShardRequest::Pexpiretime { key } => ShardResponse::Integer(ks.pexpiretime(key)),
         ShardRequest::LMove {
             source,
             destination,

crates/ember-core/src/time.rs

@@ -2,9 +2,13 @@
 //!
 //! Uses a process-local monotonic clock for timestamps that are smaller
 //! than std::time::Instant (8 bytes vs 16 bytes for Option<Instant>).
+//!
+//! All internal expiry values are stored as monotonic milliseconds since
+//! process start. Use `monotonic_to_unix_ms` to convert to wall-clock Unix
+//! timestamps for commands like EXPIRETIME and PEXPIRETIME.
 
 use std::sync::OnceLock;
-use std::time::Instant;
+use std::time::{Instant, SystemTime, UNIX_EPOCH};
 
 /// Returns current monotonic time in milliseconds since process start.
 #[inline]
@@ -26,6 +30,48 @@ pub fn now_secs() -> u32 {
     start.elapsed().as_secs() as u32
 }
 
+/// Converts a monotonic expiry timestamp (ms since process start) to a Unix
+/// epoch timestamp in milliseconds.
+///
+/// The conversion anchors the monotonic clock to wall time on first call using
+/// a single `SystemTime::now()` sample. Subsequent calls use only the fast
+/// monotonic clock and arithmetic — no system call.
+///
+/// Returns `None` if the system clock predates the Unix epoch (shouldn't
+/// happen on any real machine) or if `expires_at_ms` is `NO_EXPIRY`.
+#[inline]
+pub fn monotonic_to_unix_ms(expires_at_ms: u64) -> Option<u64> {
+    if expires_at_ms == NO_EXPIRY {
+        return None;
+    }
+
+    // Capture the relationship between monotonic and wall-clock time once.
+    struct Anchor {
+        /// Unix epoch ms at the moment we captured the anchor.
+        unix_ms_at_capture: u64,
+        /// Monotonic ms at the moment we captured the anchor.
+        mono_ms_at_capture: u64,
+    }
+
+    static ANCHOR: OnceLock<Anchor> = OnceLock::new();
+    let anchor = ANCHOR.get_or_init(|| {
+        let unix_ms_at_capture = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis()
+            .min(u64::MAX as u128) as u64;
+        let mono_ms_at_capture = now_ms();
+        Anchor {
+            unix_ms_at_capture,
+            mono_ms_at_capture,
+        }
+    });
+
+    // unix_ms = unix_at_capture + (mono_expiry - mono_at_capture)
+    let offset = expires_at_ms.saturating_sub(anchor.mono_ms_at_capture);
+    Some(anchor.unix_ms_at_capture.saturating_add(offset))
+}
+
 /// Sentinel value meaning "no expiry".
 pub const NO_EXPIRY: u64 = 0;