wedb_embed 0.1.0

pub mod meta;

pub use meta::SetMeta;

pub type SetScanResult = (u64, Vec<Vec<u8>>);

use rapidhash::RapidHashSet as HashSet;
use std::str;

use crate::db::WeDb;
use crate::error::Result;
use crate::key_composer::{KeyComposer, matches_glob_bytes};

/// 获取当前时间戳毫秒数（基于 ts_::sec()）
#[inline]
fn current_now_ms() -> u64 {
    ts_::sec() * 1000
}

impl WeDb {
    /// 获取集合元数据（自动校验过期状态，若过期或不存在返回 None）
    #[inline]
    fn get_set_meta(&self, meta_k: &str, now_ms: u64) -> Result<Option<SetMeta>> {
        match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                if let Some(meta) = SetMeta::decode(&m_bytes)
                    && !meta.is_expired(now_ms)
                {
                    return Ok(Some(meta));
                }
                Ok(None)
            }
            None => Ok(None),
        }
    }

    /// 流式遍历集合成员字节切片（零拷贝借用底层迭代器切片，可随时提前终止）
    #[inline]
    pub fn siter<K: AsRef<[u8]>, F>(&self, key: K, mut f: F) -> Result<()>
    where
        F: FnMut(&[u8]) -> bool,
    {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        match self.get_set_meta(&meta_k, now_ms)? {
            Some(m) if !m.is_empty() => m,
            _ => return Ok(()),
        };

        let prefix = kc.set_prefix(k_str);
        for g in self.data_ks.prefix(&prefix) {
            let (k, _) = g.into_inner()?;
            if !k.starts_with(&prefix) {
                break;
            }
            let member = &k[prefix.len()..];
            if !f(member) {
                break;
            }
        }
        Ok(())
    }

    /// SADD key member [member ...] (对标 Apache Kvrocks Set::Add)
    pub fn sadd<K: AsRef<[u8]>, M: AsRef<[u8]>>(&self, key: K, members: &[M]) -> Result<usize> {
        if members.is_empty() {
            return Ok(0);
        }

        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        let raw_meta_opt = self
            .meta_ks
            .get(meta_k.as_bytes())?
            .and_then(|m_bytes| SetMeta::decode(&m_bytes));

        let mut batch = self.db.batch();

        let (mut meta, is_fresh) = match raw_meta_opt {
            Some(m) if !m.is_expired(now_ms) => (m, false),
            Some(_) => {
                // 已过期：清理旧残留子键，防止脏数据影响新集合
                let prefix = kc.set_prefix(k_str);
                for g in self.data_ks.prefix(&prefix) {
                    let (k, _) = g.into_inner()?;
                    if k.starts_with(&prefix) {
                        batch.remove(&self.data_ks, &*k);
                    } else {
                        break;
                    }
                }
                (SetMeta::new_with_version(0, 0), true)
            }
            None => (SetMeta::new_with_version(0, 0), true),
        };

        let mut added = 0usize;
        let mut seen = HashSet::with_capacity_and_hasher(members.len(), Default::default());
        for m in members {
            let m_bytes = m.as_ref();
            if !seen.insert(m_bytes) {
                continue;
            }
            let item_k = kc.set_key_bytes(k_str, m_bytes);
            if is_fresh || !self.data_ks.contains_key(&item_k)? {
                added += 1;
                meta.base.size = meta.base.size.saturating_add(1);
                batch.insert(&self.data_ks, item_k, b"");
            }
        }

        if added > 0 {
            batch.insert(&self.meta_ks, meta_k.as_bytes(), meta.encode());
            batch.commit()?;
        }
        Ok(added)
    }

    /// SREM key member [member ...] (对标 Apache Kvrocks Set::Remove)
    pub fn srem<K: AsRef<[u8]>, M: AsRef<[u8]>>(&self, key: K, members: &[M]) -> Result<usize> {
        if members.is_empty() {
            return Ok(0);
        }

        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        let mut meta = match self.get_set_meta(&meta_k, now_ms)? {
            Some(m) if !m.is_empty() => m,
            _ => return Ok(0),
        };

        let mut deleted = 0usize;
        let mut batch = self.db.batch();
        let mut seen = HashSet::with_capacity_and_hasher(members.len(), Default::default());

        for m in members {
            let m_bytes = m.as_ref();
            if !seen.insert(m_bytes) {
                continue;
            }
            let item_k = kc.set_key_bytes(k_str, m_bytes);
            if self.data_ks.contains_key(&item_k)? {
                deleted += 1;
                meta.base.size = meta.base.size.saturating_sub(1);
                batch.remove(&self.data_ks, item_k);
            }
        }

        if deleted > 0 {
            if meta.base.size == 0 {
                batch.remove(&self.meta_ks, meta_k.as_bytes());
            } else {
                batch.insert(&self.meta_ks, meta_k.as_bytes(), meta.encode());
            }
            batch.commit()?;
        }
        Ok(deleted)
    }

    /// SMEMBERS key (对标 Apache Kvrocks Set::Members)
    pub fn smembers<K: AsRef<[u8]>>(&self, key: K) -> Result<Vec<Vec<u8>>> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        let meta = match self.get_set_meta(&meta_k, now_ms)? {
            Some(m) if !m.is_empty() => m,
            _ => return Ok(Vec::new()),
        };

        let mut members = Vec::with_capacity(meta.size() as usize);
        self.siter(key, |m| {
            members.push(m.to_vec());
            true
        })?;
        Ok(members)
    }

    /// SISMEMBER key member (对标 Apache Kvrocks Set::IsMember)
    pub fn sismember<K: AsRef<[u8]>, M: AsRef<[u8]>>(&self, key: K, member: M) -> Result<bool> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        if match self.get_set_meta(&meta_k, now_ms)? {
            Some(m) => m.is_empty(),
            None => true,
        } {
            return Ok(false);
        }

        let item_k = kc.set_key_bytes(k_str, member.as_ref());
        Ok(self.data_ks.contains_key(item_k)?)
    }

    /// SMISMEMBER key member [member ...] (对标 Apache Kvrocks Set::MIsMember)
    pub fn smismember<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        members: &[M],
    ) -> Result<Vec<bool>> {
        if members.is_empty() {
            return Ok(Vec::new());
        }

        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        if match self.get_set_meta(&meta_k, now_ms)? {
            Some(m) => m.is_empty(),
            None => true,
        } {
            return Ok(vec![false; members.len()]);
        }

        let mut results = Vec::with_capacity(members.len());
        for m in members {
            let item_k = kc.set_key_bytes(k_str, m.as_ref());
            results.push(self.data_ks.contains_key(item_k)?);
        }
        Ok(results)
    }

    /// SCARD key (对标 Apache Kvrocks Set::Card)
    pub fn scard<K: AsRef<[u8]>>(&self, key: K) -> Result<u64> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        match self.get_set_meta(&meta_k, now_ms)? {
            Some(meta) => Ok(meta.base.size),
            None => Ok(0),
        }
    }

    /// SMOVE source destination member (原子单批次写入，对标 Apache Kvrocks Set::Move)
    pub fn smove<S: AsRef<[u8]>, D: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        src: S,
        dst: D,
        member: M,
    ) -> Result<bool> {
        let src_bytes = src.as_ref();
        let dst_bytes = dst.as_ref();
        let m_bytes = member.as_ref();

        if src_bytes == dst_bytes {
            return self.sismember(src_bytes, m_bytes);
        }

        let kc = KeyComposer::new("default");
        let src_k_str = str::from_utf8(src_bytes).unwrap_or("");
        let dst_k_str = str::from_utf8(dst_bytes).unwrap_or("");
        let now_ms = current_now_ms();

        let src_meta_k = kc.set_meta(src_k_str);
        let mut src_meta = match self.get_set_meta(&src_meta_k, now_ms)? {
            Some(m) if !m.is_empty() => m,
            _ => return Ok(false),
        };

        let src_item_k = kc.set_key_bytes(src_k_str, m_bytes);
        if !self.data_ks.contains_key(&src_item_k)? {
            return Ok(false);
        }

        let mut batch = self.db.batch();

        // 1. 从源集合移除
        src_meta.base.size = src_meta.base.size.saturating_sub(1);
        batch.remove(&self.data_ks, src_item_k);
        if src_meta.base.size == 0 {
            batch.remove(&self.meta_ks, src_meta_k.as_bytes());
        } else {
            batch.insert(&self.meta_ks, src_meta_k.as_bytes(), src_meta.encode());
        }

        // 2. 写入目标集合
        let dst_meta_k = kc.set_meta(dst_k_str);
        let dst_raw_meta = self
            .meta_ks
            .get(dst_meta_k.as_bytes())?
            .and_then(|b| SetMeta::decode(&b));

        let (mut dst_meta, dst_is_fresh) = match dst_raw_meta {
            Some(m) if !m.is_expired(now_ms) => (m, false),
            Some(_) => {
                let prefix = kc.set_prefix(dst_k_str);
                for g in self.data_ks.prefix(&prefix) {
                    let (k, _) = g.into_inner()?;
                    if k.starts_with(&prefix) {
                        batch.remove(&self.data_ks, &*k);
                    } else {
                        break;
                    }
                }
                (SetMeta::new_with_version(0, 0), true)
            }
            None => (SetMeta::new_with_version(0, 0), true),
        };

        let dst_item_k = kc.set_key_bytes(dst_k_str, m_bytes);
        let dst_needs_insert = dst_is_fresh || !self.data_ks.contains_key(&dst_item_k)?;

        if dst_needs_insert {
            dst_meta.base.size = dst_meta.base.size.saturating_add(1);
            batch.insert(&self.data_ks, dst_item_k, b"");
            batch.insert(&self.meta_ks, dst_meta_k.as_bytes(), dst_meta.encode());
        }

        batch.commit()?;
        Ok(true)
    }

    /// SPOP key [count] (对标 Apache Kvrocks Set::Take with pop=true)
    pub fn spop<K: AsRef<[u8]>>(&self, key: K, count: usize) -> Result<Vec<Vec<u8>>> {
        if count == 0 {
            return Ok(Vec::new());
        }

        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        let mut meta = match self.get_set_meta(&meta_k, now_ms)? {
            Some(m) if !m.is_empty() => m,
            _ => return Ok(Vec::new()),
        };

        let mut all = self.smembers(&key)?;
        if all.is_empty() {
            return Ok(Vec::new());
        }

        let n = all.len();
        let num_pop = count.min(n);

        if num_pop == n {
            let mut batch = self.db.batch();
            for m in &all {
                let item_k = kc.set_key_bytes(k_str, m);
                batch.remove(&self.data_ks, item_k);
            }
            batch.remove(&self.meta_ks, meta_k.as_bytes());
            batch.commit()?;
            return Ok(all);
        }

        // Fisher-Yates partial shuffle 随机打乱前 num_pop 个元素
        for i in 0..num_pop {
            let j = fastrand::usize(i..n);
            all.swap(i, j);
        }

        let popped: Vec<Vec<u8>> = all.drain(..num_pop).collect();

        let mut batch = self.db.batch();
        for m in &popped {
            let item_k = kc.set_key_bytes(k_str, m);
            batch.remove(&self.data_ks, item_k);
        }

        meta.base.size = meta.base.size.saturating_sub(popped.len() as u64);
        batch.insert(&self.meta_ks, meta_k.as_bytes(), meta.encode());
        batch.commit()?;

        Ok(popped)
    }

    /// SRANDMEMBER key [count] (对标 Apache Kvrocks Set::Take with pop=false)
    pub fn srandmember<K: AsRef<[u8]>>(&self, key: K, count: i64) -> Result<Vec<Vec<u8>>> {
        if count == 0 {
            return Ok(Vec::new());
        }
        let all = self.smembers(&key)?;
        if all.is_empty() {
            return Ok(Vec::new());
        }

        let n = all.len();

        if count > 0 {
            let k = (count as usize).min(n);
            if k == n {
                return Ok(all);
            }
            let mut samples = all;
            for i in 0..k {
                let j = fastrand::usize(i..n);
                samples.swap(i, j);
            }
            samples.truncate(k);
            Ok(samples)
        } else {
            let num = count.unsigned_abs() as usize;
            let mut results = Vec::with_capacity(num);
            for _ in 0..num {
                let idx = fastrand::usize(..n);
                results.push(all[idx].clone());
            }
            Ok(results)
        }
    }

    /// 覆盖写入集合数据（对标 Apache Kvrocks Set::Overwrite）
    pub fn overwrite_set<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        members: &[M],
    ) -> Result<usize> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let prefix = kc.set_prefix(k_str);

        let mut batch = self.db.batch();
        for g in self.data_ks.prefix(&prefix) {
            let (k, _) = g.into_inner()?;
            if k.starts_with(&prefix) {
                batch.remove(&self.data_ks, &*k);
            } else {
                break;
            }
        }
        let meta_k = kc.set_meta(k_str);
        batch.remove(&self.meta_ks, meta_k.as_bytes());

        let mut seen = HashSet::with_capacity_and_hasher(members.len(), Default::default());
        let mut count = 0u64;
        for m in members {
            let m_bytes = m.as_ref();
            if !seen.insert(m_bytes) {
                continue;
            }
            let item_k = kc.set_key_bytes(k_str, m_bytes);
            batch.insert(&self.data_ks, item_k, b"");
            count += 1;
        }

        if count > 0 {
            let meta = SetMeta::new_with_version(0, count);
            batch.insert(&self.meta_ks, meta_k.as_bytes(), meta.encode());
        }

        batch.commit()?;
        Ok(count as usize)
    }

    /// SDIFF key [key ...] (对标 Apache Kvrocks Set::Diff，自适应点查优化)
    pub fn sdiff<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<Vec<Vec<u8>>> {
        if keys.is_empty() {
            return Ok(Vec::new());
        }
        let source = self.smembers(&keys[0])?;
        if source.is_empty() || keys.len() == 1 {
            return Ok(source);
        }

        // 短路与单向哈希剔除：将源集合直接移入哈希集合，随后自适应剔除，并在集合为空时即时短路
        let mut source_set: HashSet<Vec<u8>> =
            HashSet::with_capacity_and_hasher(source.len(), Default::default());
        for m in source {
            source_set.insert(m);
        }

        for k in &keys[1..] {
            if source_set.is_empty() {
                break;
            }
            let card = self.scard(k)?;
            if card == 0 {
                continue;
            }

            // 自适应策略：当源候选集很小而排除集合庞大时，采用点查以避免扫描整集
            if (source_set.len() as u64) * 4 < card {
                source_set.retain(|m| !self.sismember(k, m).unwrap_or(false));
            } else {
                self.siter(k, |m| {
                    source_set.remove(m);
                    !source_set.is_empty()
                })?;
            }
        }

        Ok(source_set.into_iter().collect())
    }

    /// SDIFFSTORE destination key [key ...] (对标 Apache Kvrocks Set::DiffStore)
    pub fn sdiffstore<D: AsRef<[u8]>, K: AsRef<[u8]>>(&self, dst: D, keys: &[K]) -> Result<usize> {
        let diff = self.sdiff(keys)?;
        self.overwrite_set(dst, &diff)
    }

    /// SDIFFCARD numkeys key [key ...] [LIMIT limit] (对标 Redis 7.0 SDIFFCARD)
    pub fn sdiffcard<K: AsRef<[u8]>>(&self, keys: &[K], limit: usize) -> Result<usize> {
        if keys.is_empty() {
            return Ok(0);
        }
        let first_card = self.scard(&keys[0])? as usize;
        if first_card == 0 {
            return Ok(0);
        }
        if keys.len() == 1 {
            return Ok(if limit == 0 {
                first_card
            } else {
                first_card.min(limit)
            });
        }

        let diff = self.sdiff(keys)?;
        Ok(if limit == 0 {
            diff.len()
        } else {
            diff.len().min(limit)
        })
    }

    /// SUNION key [key ...] (对标 Apache Kvrocks Set::Union)
    pub fn sunion<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<Vec<Vec<u8>>> {
        if keys.is_empty() {
            return Ok(Vec::new());
        }
        if keys.len() == 1 {
            return self.smembers(&keys[0]);
        }

        let first_card = self.scard(&keys[0])? as usize;
        let mut union_set = HashSet::with_capacity_and_hasher(first_card, Default::default());
        for k in keys {
            self.siter(k, |m| {
                union_set.insert(m.to_vec());
                true
            })?;
        }
        Ok(union_set.into_iter().collect())
    }

    /// SUNIONSTORE destination key [key ...] (对标 Apache Kvrocks Set::UnionStore)
    pub fn sunionstore<D: AsRef<[u8]>, K: AsRef<[u8]>>(&self, dst: D, keys: &[K]) -> Result<usize> {
        let union_res = self.sunion(keys)?;
        self.overwrite_set(dst, &union_res)
    }

    /// SUNIONCARD numkeys key [key ...] [LIMIT limit] (对标 Redis 7.0+ SUNIONCARD)
    pub fn sunioncard<K: AsRef<[u8]>>(&self, keys: &[K], limit: usize) -> Result<usize> {
        if keys.is_empty() {
            return Ok(0);
        }

        let mut union_set = HashSet::default();
        if limit > 0 {
            for k in keys {
                self.siter(k, |m| {
                    union_set.insert(m.to_vec());
                    union_set.len() < limit
                })?;
                if union_set.len() >= limit {
                    return Ok(limit);
                }
            }
            Ok(union_set.len().min(limit))
        } else {
            for k in keys {
                self.siter(k, |m| {
                    union_set.insert(m.to_vec());
                    true
                })?;
            }
            Ok(union_set.len())
        }
    }

    /// SINTER key [key ...] (对标 Apache Kvrocks Set::Inter，基数排序与自适应点查优化)
    pub fn sinter<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<Vec<Vec<u8>>> {
        if keys.is_empty() {
            return Ok(Vec::new());
        }
        if keys.len() == 1 {
            return self.smembers(&keys[0]);
        }

        // 短路检查：如果任一集合为空，交集必然为空
        let mut key_cards: Vec<(&K, u64)> = Vec::with_capacity(keys.len());
        for k in keys {
            let card = self.scard(k)?;
            if card == 0 {
                return Ok(Vec::new());
            }
            key_cards.push((k, card));
        }

        // 依据基数由小到大排序，从最小的集合开始构建 Hash 表以优化性能
        key_cards.sort_unstable_by_key(|&(_, card)| card);

        let smallest_key = key_cards[0].0;
        let mut current: HashSet<Vec<u8>> =
            HashSet::with_capacity_and_hasher(key_cards[0].1 as usize, Default::default());
        self.siter(smallest_key, |m| {
            current.insert(m.to_vec());
            true
        })?;

        if current.is_empty() {
            return Ok(Vec::new());
        }

        for &(k, next_card) in &key_cards[1..] {
            if current.is_empty() {
                return Ok(Vec::new());
            }

            // 自适应分支：当候选集合大小远小于下一个集合的大小时，采用点查原地 retain 避免扫描与重复分配
            if (current.len() as u64) * 4 < next_card {
                current.retain(|m| self.sismember(k, m).unwrap_or(false));
            } else {
                let mut next_set =
                    HashSet::with_capacity_and_hasher(current.len(), Default::default());
                self.siter(k, |m| {
                    if current.contains(m) {
                        next_set.insert(m.to_vec());
                    }
                    true
                })?;
                current = next_set;
            }
        }

        Ok(current.into_iter().collect())
    }

    /// SINTERSTORE destination key [key ...] (对标 Apache Kvrocks Set::InterStore)
    pub fn sinterstore<D: AsRef<[u8]>, K: AsRef<[u8]>>(&self, dst: D, keys: &[K]) -> Result<usize> {
        let inter_res = self.sinter(keys)?;
        self.overwrite_set(dst, &inter_res)
    }

    /// SINTERCARD numkeys key [key ...] [LIMIT limit] (对标 Apache Kvrocks Set::InterCard)
    pub fn sintercard<K: AsRef<[u8]>>(&self, keys: &[K], limit: usize) -> Result<usize> {
        if keys.is_empty() {
            return Ok(0);
        }
        if keys.len() == 1 {
            let card = self.scard(&keys[0])? as usize;
            return Ok(if limit == 0 { card } else { card.min(limit) });
        }

        let mut key_cards: Vec<(&K, u64)> = Vec::with_capacity(keys.len());
        for k in keys {
            let card = self.scard(k)?;
            if card == 0 {
                return Ok(0);
            }
            key_cards.push((k, card));
        }

        key_cards.sort_unstable_by_key(|&(_, card)| card);

        let smallest_key = key_cards[0].0;
        let mut current: HashSet<Vec<u8>> =
            HashSet::with_capacity_and_hasher(key_cards[0].1 as usize, Default::default());
        self.siter(smallest_key, |m| {
            current.insert(m.to_vec());
            true
        })?;

        if current.is_empty() {
            return Ok(0);
        }

        for &(k, next_card) in &key_cards[1..] {
            if current.is_empty() {
                return Ok(0);
            }

            if (current.len() as u64) * 4 < next_card {
                current.retain(|m| self.sismember(k, m).unwrap_or(false));
            } else {
                let mut next_set =
                    HashSet::with_capacity_and_hasher(current.len(), Default::default());
                self.siter(k, |m| {
                    if current.contains(m) {
                        next_set.insert(m.to_vec());
                    }
                    true
                })?;
                current = next_set;
            }
        }

        let total = current.len();
        Ok(if limit == 0 { total } else { total.min(limit) })
    }

    /// SSCAN key cursor [MATCH pattern] [COUNT count] (流式分页扫描，对标 Apache Kvrocks Set::Scan)
    pub fn sscan<K: AsRef<[u8]>>(
        &self,
        key: K,
        cursor: u64,
        pattern: Option<&[u8]>,
        count: Option<usize>,
    ) -> Result<SetScanResult> {
        let total = self.scard(&key)? as usize;
        let start = cursor as usize;
        if start >= total || total == 0 {
            return Ok((0, Vec::new()));
        }

        let step = count.unwrap_or(10);
        let end = (start + step).min(total);
        let next_cursor = if end >= total { 0 } else { end as u64 };

        let mut results = Vec::new();
        let mut current_idx = 0usize;

        self.siter(key, |member| {
            if current_idx >= start && current_idx < end {
                let matches = match pattern {
                    Some(pat) => matches_glob_bytes(pat, member),
                    None => true,
                };
                if matches {
                    results.push(member.to_vec());
                }
            }
            current_idx += 1;
            current_idx < end
        })?;

        Ok((next_cursor, results))
    }

    /// SEXPIREAT key expire_at_ms (设置 Set 绝对过期毫秒时间戳)
    pub fn sexpireat<K: AsRef<[u8]>>(&self, key: K, expire_at_ms: u64) -> Result<bool> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        let mut meta = match self.get_set_meta(&meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(false),
        };

        meta.base.expire_at = expire_at_ms;
        let mut batch = self.db.batch();
        batch.insert(&self.meta_ks, meta_k.as_bytes(), meta.encode());
        batch.commit()?;
        Ok(true)
    }

    /// STTL key (获取 Set 剩余存活毫秒时间)
    pub fn sttl<K: AsRef<[u8]>>(&self, key: K) -> Result<i64> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        match self.get_set_meta(&meta_k, now_ms)? {
            Some(meta) => Ok(meta.ttl(now_ms)),
            None => Ok(-2),
        }
    }

    /// SPERSIST key (移除 Set 的过期时间)
    pub fn spersist<K: AsRef<[u8]>>(&self, key: K) -> Result<bool> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.set_meta(k_str);
        let now_ms = current_now_ms();

        let mut meta = match self.get_set_meta(&meta_k, now_ms)? {
            Some(m) if m.base.expire_at > 0 => m,
            _ => return Ok(false),
        };

        meta.base.expire_at = 0;
        let mut batch = self.db.batch();
        batch.insert(&self.meta_ks, meta_k.as_bytes(), meta.encode());
        batch.commit()?;
        Ok(true)
    }
}

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

    #[test]
    fn test_set_comprehensive_operations() -> Result<()> {
        let temp_dir = tempfile::tempdir().unwrap();
        let db = WeDb::open(temp_dir.path())?;

        // SADD & SCARD
        assert_eq!(db.sadd("s1", &[b"a", b"b", b"c"])?, 3);
        assert_eq!(db.sadd("s1", &[b"a", b"d"])?, 1);
        assert_eq!(db.sadd("s1", &[] as &[&[u8]])?, 0);
        assert_eq!(db.scard("s1")?, 4);

        // SISMEMBER & SMISMEMBER
        assert!(db.sismember("s1", b"a")?);
        assert!(!db.sismember("s1", b"z")?);
        assert_eq!(
            db.smismember("s1", &[b"a", b"z", b"b"])?,
            vec![true, false, true]
        );
        assert_eq!(
            db.smismember("non_existent", &[b"a", b"b"])?,
            vec![false, false]
        );

        // SMEMBERS & SITER
        let members = db.smembers("s1")?;
        assert_eq!(members.len(), 4);
        let mut iter_members = Vec::new();
        db.siter("s1", |m| {
            iter_members.push(m.to_vec());
            true
        })?;
        assert_eq!(iter_members.len(), 4);

        // SREM
        assert_eq!(db.srem("s1", &[b"d"])?, 1);
        assert_eq!(db.srem("s1", &[b"d"])?, 0);
        assert_eq!(db.scard("s1")?, 3);

        // SMOVE
        assert!(db.smove("s1", "s2", b"c")?);
        assert_eq!(db.scard("s1")?, 2);
        assert_eq!(db.scard("s2")?, 1);
        assert!(db.smove("s2", "s2", b"c")?); // 同 key 移动
        assert!(!db.smove("s2", "s2", b"nonexistent")?);

        // SDIFF, SUNION, SINTER
        db.sadd("setA", &[b"1", b"2", b"3", b"4"])?;
        db.sadd("setB", &[b"3", b"4", b"5", b"6"])?;

        let diff = db.sdiff(&["setA", "setB"])?;
        assert_eq!(diff.len(), 2);
        assert_eq!(db.sdiffcard(&["setA", "setB"], 0)?, 2);
        assert_eq!(db.sdiffcard(&["setA", "setB"], 1)?, 1);

        let union_res = db.sunion(&["setA", "setB"])?;
        assert_eq!(union_res.len(), 6);
        assert_eq!(db.sunioncard(&["setA", "setB"], 0)?, 6);
        assert_eq!(db.sunioncard(&["setA", "setB"], 3)?, 3);

        let inter_res = db.sinter(&["setA", "setB"])?;
        assert_eq!(inter_res.len(), 2);

        assert_eq!(db.sintercard(&["setA", "setB"], 1)?, 1);
        assert_eq!(db.sintercard(&["setA", "setB"], 0)?, 2);
        assert_eq!(db.sintercard(&["setA", "empty_key"], 0)?, 0);

        // SDIFFSTORE, SUNIONSTORE, SINTERSTORE
        assert_eq!(db.sdiffstore("out_diff", &["setA", "setB"])?, 2);
        assert_eq!(db.sunionstore("out_union", &["setA", "setB"])?, 6);
        assert_eq!(db.sinterstore("out_inter", &["setA", "setB"])?, 2);

        // SPOP & SRANDMEMBER
        let popped = db.spop("out_diff", 1)?;
        assert_eq!(popped.len(), 1);
        assert_eq!(db.scard("out_diff")?, 1);

        let rand_m = db.srandmember("out_union", 3)?;
        assert_eq!(rand_m.len(), 3);
        let rand_m_neg = db.srandmember("out_union", -5)?;
        assert_eq!(rand_m_neg.len(), 5);

        // SSCAN
        let (next_cur, scan_res) = db.sscan("out_union", 0, None, Some(4))?;
        assert_eq!(scan_res.len(), 4);
        assert_eq!(next_cur, 4);

        // SEXPIREAT & STTL & SPERSIST
        let future_ts = (ts_::sec() + 100) * 1000;
        assert!(db.sexpireat("out_union", future_ts)?);
        assert!(db.sttl("out_union")? > 0);
        assert!(db.spersist("out_union")?);
        assert_eq!(db.sttl("out_union")?, -1);
        assert_eq!(db.sttl("non_existent_key")?, -2);

        Ok(())
    }
}