wedb_embed 0.1.0

pub mod conf;
pub mod meta;
pub mod sortable;

pub use conf::{Aggregate, RangeLexSpec, RangeRankSpec, RangeScoreSpec, ZAdd, ZRange, ZRangeSpec};
pub use meta::ZSetMeta;
pub use sortable::{decode_sortable_f64, encode_sortable_f64};

pub type ZSetMemberScore = (Vec<u8>, f64);
pub type ZSetKeyMemberScore = (Vec<u8>, Vec<u8>, f64);
pub type ZScanResult = (u64, Vec<ZSetMemberScore>);

use rapidhash::{RapidHashMap as HashMap, RapidHashSet as HashSet};
use std::str;

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

const HEX_SCORE_LEN: usize = 16;

/// 编译期静态十六进制解码表（0xFF 表示非法字符，消除运行时多重分支与跳转）
const HEX_DECODE_LUT: [u8; 256] = {
    let mut table = [0xFF; 256];
    let mut i = b'0';
    while i <= b'9' {
        table[i as usize] = i - b'0';
        i += 1;
    }
    let mut i = b'a';
    while i <= b'f' {
        table[i as usize] = i - b'a' + 10;
        i += 1;
    }
    let mut i = b'A';
    while i <= b'F' {
        table[i as usize] = i - b'A' + 10;
        i += 1;
    }
    table
};

/// 16 字节十六进制极速解码为 8 字节（保序浮点数原始二进制）
#[inline(always)]
fn decode_hex_8b(s: &[u8]) -> Option<[u8; 8]> {
    if s.len() != HEX_SCORE_LEN {
        return None;
    }
    let mut out = [0u8; 8];
    let mut i = 0;
    while i < 8 {
        let hi = HEX_DECODE_LUT[s[i * 2] as usize];
        let lo = HEX_DECODE_LUT[s[i * 2 + 1] as usize];
        if (hi | lo) == 0xFF {
            return None;
        }
        out[i] = (hi << 4) | lo;
        i += 1;
    }
    Some(out)
}

/// 零拷贝解析 Score 索引子键中的 (score, member) 切片
#[inline(always)]
fn parse_score_sub(sub: &[u8]) -> Option<(f64, &[u8])> {
    if sub.len() > HEX_SCORE_LEN && sub[HEX_SCORE_LEN] == b':' {
        let hex_score = &sub[..HEX_SCORE_LEN];
        let member = &sub[HEX_SCORE_LEN + 1..];
        let b8 = decode_hex_8b(hex_score)?;
        Some((decode_sortable_f64(b8), member))
    } else {
        None
    }
}

/// 标准 Redis 索引范围规整化（支持负索引，下界 0 对齐，超范围终止）
#[inline]
fn normalize_range(card: usize, start: i64, stop: i64) -> Option<(usize, usize)> {
    if card == 0 {
        return None;
    }
    let card_i = card as i64;
    let mut s = if start < 0 { start + card_i } else { start };
    let mut e = if stop < 0 { stop + card_i } else { stop };
    if s < 0 {
        s = 0;
    }
    if e < 0 || s > e || s >= card_i {
        return None;
    }
    if e >= card_i {
        e = card_i - 1;
    }
    Some((s as usize, e as usize))
}

impl WeDb {
    /// 流式遍历有序集合成员与分数（按分数升序，同分数按 member 字典序升序）
    /// 零额外堆分配，直接借用数据库前缀迭代器中的字节
    #[inline]
    pub fn ziter<K: AsRef<[u8]>, F>(&self, key: K, mut f: F) -> Result<()>
    where
        F: FnMut(&[u8], f64) -> bool,
    {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                if let Some(meta) = ZSetMeta::decode(&m_bytes)
                    && (meta.is_expired(now_ms) || meta.is_empty())
                {
                    return Ok(());
                }
            }
            None => return Ok(()),
        }

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

    /// 获取有序集合全部成员与分数（按分数由小到大排序，同分数按 member 字典序升序）
    pub fn zget_all<K: AsRef<[u8]>>(&self, key: K) -> Result<Vec<ZSetMemberScore>> {
        let card = self.zcard(&key)? as usize;
        let mut items = Vec::with_capacity(card);
        self.ziter(key, |member, score| {
            items.push((member.to_vec(), score));
            true
        })?;
        Ok(items)
    }

    /// ZADD key [NX | XX] [GT | LT] [CH] [INCR] score member [score member ...]
    pub fn zadd<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        score_members: &[(f64, M)],
        conf_li: impl AsRef<[ZAdd]>,
    ) -> Result<usize> {
        if score_members.is_empty() {
            return Ok(0);
        }
        for (score, _) in score_members {
            if score.is_nan() {
                return Err(Error::invalid_data("ERR score is not a valid float"));
            }
        }

        let options = conf_li.as_ref();
        let nx = options.contains(&ZAdd::Nx);
        let xx = options.contains(&ZAdd::Xx);
        let gt = options.contains(&ZAdd::Gt);
        let lt = options.contains(&ZAdd::Lt);
        let ch = options.contains(&ZAdd::Ch);
        let incr = options.contains(&ZAdd::Incr);

        if nx && xx {
            return Err(Error::invalid_data(
                "ERR XX and NX options at the same time are not compatible",
            ));
        }
        if (gt && lt) || (nx && gt) || (nx && lt) {
            return Err(Error::invalid_data(
                "ERR GT, LT, and/or NX options at the same time are not compatible",
            ));
        }
        if incr && score_members.len() > 1 {
            return Err(Error::invalid_data(
                "ERR INCR option supports a single increment-element pair",
            ));
        }

        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        let (mut meta, metadata_existed) = match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                if let Some(m) = ZSetMeta::decode(&m_bytes) {
                    if m.is_expired(now_ms) || m.size() == 0 {
                        (ZSetMeta::new_with_version(0, 0), false)
                    } else {
                        (m, true)
                    }
                } else {
                    (ZSetMeta::new_with_version(0, 0), false)
                }
            }
            None => (ZSetMeta::new_with_version(0, 0), false),
        };

        let mut added = 0;
        let mut changed = 0;
        let mut batch = self.db.batch();
        let mut seen = HashSet::with_capacity_and_hasher(score_members.len(), Default::default());

        let prefix = kc.zset_prefix(k_str);
        let score_prefix = kc.zset_score_prefix(k_str);
        let mut m_key = Vec::with_capacity(prefix.len() + 32);
        let mut s_key = Vec::with_capacity(score_prefix.len() + HEX_SCORE_LEN + 1 + 32);

        // 逆序遍历去重：确保同一次指令中若同一 member 出现多次，以后出现的值为准（对标 Kvrocks / Redis）
        for (input_score, member) in score_members.iter().rev() {
            let m_bytes = member.as_ref();
            if !seen.insert(m_bytes) {
                continue;
            }

            m_key.clear();
            m_key.extend_from_slice(&prefix);
            m_key.extend_from_slice(m_bytes);

            let old_score_bytes = if metadata_existed {
                self.data_ks.get(&m_key)?
            } else {
                None
            };

            if let Some(old_sb) = old_score_bytes {
                if nx {
                    continue;
                }
                let mut sb = [0u8; 8];
                if old_sb.len() >= 8 {
                    sb.copy_from_slice(&old_sb[..8]);
                }
                let old_score = decode_sortable_f64(sb);

                let final_score = if incr {
                    if (lt && *input_score >= 0.0) || (gt && *input_score <= 0.0) {
                        continue;
                    }
                    old_score + *input_score
                } else {
                    *input_score
                };

                if final_score.is_nan() {
                    return Err(Error::invalid_data(
                        "ERR resulting score is not a number (NaN)",
                    ));
                }

                if (gt && final_score <= old_score) || (lt && final_score >= old_score) {
                    continue;
                }

                if final_score != old_score {
                    changed += 1;
                    let old_enc = encode_sortable_f64(old_score);
                    let old_hex = bytes_to_hex_16(old_enc);
                    s_key.clear();
                    s_key.extend_from_slice(&score_prefix);
                    s_key.extend_from_slice(&old_hex);
                    s_key.push(b':');
                    s_key.extend_from_slice(m_bytes);
                    batch.remove(&self.data_ks, &s_key);

                    let new_enc = encode_sortable_f64(final_score);
                    let new_hex = bytes_to_hex_16(new_enc);
                    s_key.clear();
                    s_key.extend_from_slice(&score_prefix);
                    s_key.extend_from_slice(&new_hex);
                    s_key.push(b':');
                    s_key.extend_from_slice(m_bytes);
                    batch.insert(&self.data_ks, &s_key, b"");
                    batch.insert(&self.data_ks, &m_key, new_enc);
                }
            } else {
                if xx {
                    continue;
                }
                let final_score = *input_score;
                added += 1;
                meta.base.size = meta.base.size.saturating_add(1);

                let new_enc = encode_sortable_f64(final_score);
                let new_hex = bytes_to_hex_16(new_enc);
                s_key.clear();
                s_key.extend_from_slice(&score_prefix);
                s_key.extend_from_slice(&new_hex);
                s_key.push(b':');
                s_key.extend_from_slice(m_bytes);

                batch.insert(&self.data_ks, &s_key, b"");
                batch.insert(&self.data_ks, &m_key, new_enc);
            }
        }

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

        Ok(if ch { added + changed } else { added })
    }

    /// ZREM key member [member ...]
    pub fn zrem<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.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        let mut meta = match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                let m = ZSetMeta::decode(&m_bytes).unwrap_or_else(|| ZSetMeta::new(0, 0, 0));
                if m.is_expired(now_ms) || m.size() == 0 {
                    return Ok(0);
                }
                m
            }
            None => return Ok(0),
        };

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

        let prefix = kc.zset_prefix(k_str);
        let score_prefix = kc.zset_score_prefix(k_str);
        let mut m_key = Vec::with_capacity(prefix.len() + 32);
        let mut s_key = Vec::with_capacity(score_prefix.len() + HEX_SCORE_LEN + 1 + 32);

        for member in members {
            let m_bytes = member.as_ref();
            if !seen.insert(m_bytes) {
                continue;
            }
            m_key.clear();
            m_key.extend_from_slice(&prefix);
            m_key.extend_from_slice(m_bytes);

            if let Some(sb) = self.data_ks.get(&m_key)? {
                deleted += 1;
                meta.base.size = meta.base.size.saturating_sub(1);
                let mut b = [0u8; 8];
                if sb.len() >= 8 {
                    b.copy_from_slice(&sb[..8]);
                }
                let score = decode_sortable_f64(b);
                let enc = encode_sortable_f64(score);
                let hex_bytes = bytes_to_hex_16(enc);

                s_key.clear();
                s_key.extend_from_slice(&score_prefix);
                s_key.extend_from_slice(&hex_bytes);
                s_key.push(b':');
                s_key.extend_from_slice(m_bytes);

                batch.remove(&self.data_ks, &s_key);
                batch.remove(&self.data_ks, &m_key);
            }
        }

        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)
    }

    /// ZSCORE key member
    pub fn zscore<K: AsRef<[u8]>, M: AsRef<[u8]>>(&self, key: K, member: M) -> Result<Option<f64>> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                if let Some(meta) = ZSetMeta::decode(&m_bytes)
                    && (meta.is_expired(now_ms) || meta.is_empty())
                {
                    return Ok(None);
                }
            }
            None => return Ok(None),
        }

        let m_key = kc.zset_key_bytes(k_str, member.as_ref());
        match self.data_ks.get(&m_key)? {
            Some(sb) if sb.len() >= 8 => {
                let mut b = [0u8; 8];
                b.copy_from_slice(&sb[..8]);
                Ok(Some(decode_sortable_f64(b)))
            }
            _ => Ok(None),
        }
    }

    /// ZMSCORE key member [member ...] (单次元数据检查与缓冲池点查，极致性能)
    pub fn zmscore<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        members: &[M],
    ) -> Result<Vec<Option<f64>>> {
        let mut scores = Vec::with_capacity(members.len());
        if members.is_empty() {
            return Ok(scores);
        }

        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                if let Some(meta) = ZSetMeta::decode(&m_bytes)
                    && (meta.is_expired(now_ms) || meta.is_empty())
                {
                    scores.resize(members.len(), None);
                    return Ok(scores);
                }
            }
            None => {
                scores.resize(members.len(), None);
                return Ok(scores);
            }
        }

        let prefix = kc.zset_prefix(k_str);
        let mut m_key = Vec::with_capacity(prefix.len() + 32);

        for m in members {
            m_key.clear();
            m_key.extend_from_slice(&prefix);
            m_key.extend_from_slice(m.as_ref());
            let score = match self.data_ks.get(&m_key)? {
                Some(sb) if sb.len() >= 8 => {
                    let mut b = [0u8; 8];
                    b.copy_from_slice(&sb[..8]);
                    Some(decode_sortable_f64(b))
                }
                _ => None,
            };
            scores.push(score);
        }
        Ok(scores)
    }

    /// ZMGET key member [member ...] (对标 Apache Kvrocks ZSet::MGet)
    pub fn zmget<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        members: &[M],
    ) -> Result<HashMap<Vec<u8>, f64>> {
        let mut mscores = HashMap::with_capacity_and_hasher(members.len(), Default::default());
        if members.is_empty() {
            return Ok(mscores);
        }

        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let meta_k = kc.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                if let Some(meta) = ZSetMeta::decode(&m_bytes)
                    && (meta.is_expired(now_ms) || meta.is_empty())
                {
                    return Ok(mscores);
                }
            }
            None => return Ok(mscores),
        }

        let prefix = kc.zset_prefix(k_str);
        let mut m_key = Vec::with_capacity(prefix.len() + 32);

        for m in members {
            let m_bytes = m.as_ref();
            m_key.clear();
            m_key.extend_from_slice(&prefix);
            m_key.extend_from_slice(m_bytes);
            if let Some(sb) = self.data_ks.get(&m_key)?
                && sb.len() >= 8
            {
                let mut b = [0u8; 8];
                b.copy_from_slice(&sb[..8]);
                mscores.insert(m_bytes.to_vec(), decode_sortable_f64(b));
            }
        }
        Ok(mscores)
    }

    /// ZCARD key
    pub fn zcard<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.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                let meta = ZSetMeta::decode(&m_bytes).unwrap_or_else(|| ZSetMeta::new(0, 0, 0));
                if meta.is_expired(now_ms) {
                    Ok(0)
                } else {
                    Ok(meta.base.size)
                }
            }
            None => Ok(0),
        }
    }

    /// ZCOUNT key min max（流式单遍扫描，遇到超上限提前中断，零额外内存分配）
    pub fn zcount<K: AsRef<[u8]>>(&self, key: K, spec: &RangeScoreSpec) -> Result<u64> {
        if spec.min > spec.max {
            return Ok(0);
        }
        let mut count = 0u64;
        self.ziter(key, |_, score| {
            if (spec.maxex && score >= spec.max) || score > spec.max {
                return false;
            }
            if spec.check(score) {
                count += 1;
            }
            true
        })?;
        Ok(count)
    }

    /// ZLEXCOUNT key min max（流式单遍扫描，零额外内存分配）
    pub fn zlexcount<K: AsRef<[u8]>>(&self, key: K, spec: &RangeLexSpec) -> Result<u64> {
        let mut count = 0u64;
        self.ziter(key, |member, _| {
            if spec.check(member) {
                count += 1;
            }
            true
        })?;
        Ok(count)
    }

    /// ZINCRBY key increment member
    pub fn zincrby<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        increment: f64,
        member: M,
    ) -> Result<f64> {
        if increment.is_nan() {
            return Err(Error::invalid_data("ERR increment is not a valid float"));
        }
        let old_score = self.zscore(&key, &member)?.unwrap_or(0.0);
        let new_score = old_score + increment;
        if new_score.is_nan() {
            return Err(Error::invalid_data(
                "ERR resulting score is not a number (NaN)",
            ));
        }
        self.zadd(&key, &[(new_score, member)], [])?;
        Ok(new_score)
    }

    /// ZRANK key member (0-based rank from smallest score to largest，单遍迭代，命中立即提前终止)
    pub fn zrank<K: AsRef<[u8]>, M: AsRef<[u8]>>(&self, key: K, member: M) -> Result<Option<u64>> {
        let m_ref = member.as_ref();
        let mut rank = 0u64;
        let mut found = false;
        self.ziter(key, |m, _| {
            if m == m_ref {
                found = true;
                false
            } else {
                rank += 1;
                true
            }
        })?;
        Ok(if found { Some(rank) } else { None })
    }

    /// ZRANK key member [WITHSCORE] (对标 Redis 7.2 / Kvrocks ZSet::Rank，单遍迭代获取排名与分数)
    pub fn zrank_with_score<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        member: M,
    ) -> Result<Option<(u64, f64)>> {
        let m_ref = member.as_ref();
        let mut rank = 0u64;
        let mut found_score = None;
        self.ziter(key, |m, score| {
            if m == m_ref {
                found_score = Some(score);
                false
            } else {
                rank += 1;
                true
            }
        })?;
        Ok(found_score.map(|s| (rank, s)))
    }

    /// ZREVRANK key member (0-based rank from largest score to smallest)
    pub fn zrevrank<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        member: M,
    ) -> Result<Option<u64>> {
        let card = self.zcard(&key)?;
        if card == 0 {
            return Ok(None);
        }
        let m_ref = member.as_ref();
        let mut rank = 0u64;
        let mut found = false;
        self.ziter(key, |m, _| {
            if m == m_ref {
                found = true;
                false
            } else {
                rank += 1;
                true
            }
        })?;
        Ok(if found {
            Some((card - 1).saturating_sub(rank))
        } else {
            None
        })
    }

    /// ZREVRANK key member [WITHSCORE]
    pub fn zrevrank_with_score<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        member: M,
    ) -> Result<Option<(u64, f64)>> {
        let card = self.zcard(&key)?;
        if card == 0 {
            return Ok(None);
        }
        let m_ref = member.as_ref();
        let mut rank = 0u64;
        let mut found_score = None;
        self.ziter(key, |m, score| {
            if m == m_ref {
                found_score = Some(score);
                false
            } else {
                rank += 1;
                true
            }
        })?;
        Ok(found_score.map(|score| ((card - 1).saturating_sub(rank), score)))
    }

    /// ZRANGE key start stop (流式截取指定排名区间，支持负索引，达到上限即终止)
    pub fn zrange<K: AsRef<[u8]>>(
        &self,
        key: K,
        start: i64,
        stop: i64,
    ) -> Result<Vec<ZSetMemberScore>> {
        let card = self.zcard(&key)? as usize;
        let (s, e) = match normalize_range(card, start, stop) {
            Some(range) => range,
            None => return Ok(Vec::new()),
        };

        let count = e - s + 1;
        let mut items = Vec::with_capacity(count);
        let mut current_idx = 0usize;

        self.ziter(key, |member, score| {
            if current_idx >= s {
                items.push((member.to_vec(), score));
                if items.len() >= count {
                    return false;
                }
            }
            current_idx += 1;
            true
        })?;

        Ok(items)
    }

    /// ZREVRANGE key start stop (按降序流式截取，达到上限即终止)
    pub fn zrevrange<K: AsRef<[u8]>>(
        &self,
        key: K,
        start: i64,
        stop: i64,
    ) -> Result<Vec<ZSetMemberScore>> {
        let card = self.zcard(&key)? as usize;
        let (s, e) = match normalize_range(card, start, stop) {
            Some(range) => range,
            None => return Ok(Vec::new()),
        };

        let asc_start = card - 1 - e;
        let asc_stop = card - 1 - s;
        let count = asc_stop - asc_start + 1;

        let mut items = Vec::with_capacity(count);
        let mut current_idx = 0usize;

        self.ziter(key, |member, score| {
            if current_idx >= asc_start {
                items.push((member.to_vec(), score));
                if items.len() >= count {
                    return false;
                }
            }
            current_idx += 1;
            true
        })?;

        items.reverse();
        Ok(items)
    }

    /// ZRANGEBYSCORE key min max [LIMIT offset count]（流式过滤，超出 max 即提前中断）
    pub fn zrangebyscore<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &RangeScoreSpec,
    ) -> Result<Vec<ZSetMemberScore>> {
        if spec.count == Some(0) || spec.min > spec.max {
            return Ok(Vec::new());
        }
        let mut items = Vec::new();
        let mut skipped = 0usize;

        self.ziter(key, |member, score| {
            if (spec.maxex && score >= spec.max) || score > spec.max {
                return false;
            }
            if spec.check(score) {
                if skipped < spec.offset {
                    skipped += 1;
                    return true;
                }
                items.push((member.to_vec(), score));
                if let Some(limit) = spec.count
                    && items.len() >= limit
                {
                    return false;
                }
            }
            true
        })?;
        Ok(items)
    }

    /// ZREVRANGEBYSCORE key max min [LIMIT offset count]
    pub fn zrevrangebyscore<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &RangeScoreSpec,
    ) -> Result<Vec<ZSetMemberScore>> {
        if spec.count == Some(0) || spec.min > spec.max {
            return Ok(Vec::new());
        }
        let mut matching = Vec::new();
        self.ziter(key, |member, score| {
            if (spec.maxex && score >= spec.max) || score > spec.max {
                return false;
            }
            if spec.check(score) {
                matching.push((member.to_vec(), score));
            }
            true
        })?;

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

        matching.reverse();
        let offset = spec.offset;
        let limit = spec.count.unwrap_or(matching.len());
        if offset >= matching.len() {
            return Ok(Vec::new());
        }
        let end = (offset + limit).min(matching.len());
        Ok(matching[offset..end].to_vec())
    }

    /// ZRANGEBYLEX key min max [LIMIT offset count]（流式过滤，达到限制立即提前终止）
    pub fn zrangebylex<K: AsRef<[u8]>>(&self, key: K, spec: &RangeLexSpec) -> Result<Vec<Vec<u8>>> {
        if spec.count == Some(0) {
            return Ok(Vec::new());
        }
        let mut items = Vec::new();
        let mut skipped = 0usize;

        self.ziter(key, |member, _| {
            if spec.check(member) {
                if skipped < spec.offset {
                    skipped += 1;
                    return true;
                }
                items.push(member.to_vec());
                if let Some(limit) = spec.count
                    && items.len() >= limit
                {
                    return false;
                }
            }
            true
        })?;
        Ok(items)
    }

    /// ZREVRANGEBYLEX key max min [LIMIT offset count]
    pub fn zrevrangebylex<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &RangeLexSpec,
    ) -> Result<Vec<Vec<u8>>> {
        if spec.count == Some(0) {
            return Ok(Vec::new());
        }
        let mut matching = Vec::new();
        self.ziter(key, |member, _| {
            if spec.check(member) {
                matching.push(member.to_vec());
            }
            true
        })?;

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

        matching.reverse();
        let offset = spec.offset;
        let limit = spec.count.unwrap_or(matching.len());
        if offset >= matching.len() {
            return Ok(Vec::new());
        }
        let end = (offset + limit).min(matching.len());
        Ok(matching[offset..end].to_vec())
    }

    /// 统一 ZRANGE 范围查询（支持 BYSCORE / BYLEX / REV / LIMIT，对标 Redis 6.2+）
    pub fn zrange_spec<K: AsRef<[u8]>>(
        &self,
        key: K,
        start_or_min: &[u8],
        stop_or_max: &[u8],
        spec: &ZRangeSpec,
    ) -> Result<Vec<ZSetMemberScore>> {
        if spec.by_score {
            let s_min = str::from_utf8(start_or_min).unwrap_or("-inf");
            let s_max = str::from_utf8(stop_or_max).unwrap_or("+inf");
            let (min, minex) = RangeScoreSpec::parse_bound(s_min)?;
            let (max, maxex) = RangeScoreSpec::parse_bound(s_max)?;
            let range_spec = RangeScoreSpec {
                min,
                max,
                minex,
                maxex,
                offset: spec.offset,
                count: spec.count,
            };
            if spec.rev {
                self.zrevrangebyscore(key, &range_spec)
            } else {
                self.zrangebyscore(key, &range_spec)
            }
        } else if spec.by_lex {
            let (min, minex, min_inf) = RangeLexSpec::parse_bound(start_or_min)?;
            let (max, maxex, max_inf) = RangeLexSpec::parse_bound(stop_or_max)?;
            let range_spec = RangeLexSpec {
                min,
                max,
                minex,
                maxex,
                min_infinite: min_inf,
                max_infinite: max_inf,
                offset: spec.offset,
                count: spec.count,
            };
            let members = if spec.rev {
                self.zrevrangebylex(&key, &range_spec)?
            } else {
                self.zrangebylex(&key, &range_spec)?
            };
            let mut results = Vec::with_capacity(members.len());
            for m in members {
                let score = if spec.with_scores {
                    self.zscore(&key, &m)?.unwrap_or(0.0)
                } else {
                    0.0
                };
                results.push((m, score));
            }
            Ok(results)
        } else {
            let s_start = str::from_utf8(start_or_min)
                .unwrap_or("0")
                .parse::<i64>()
                .unwrap_or(0);
            let s_stop = str::from_utf8(stop_or_max)
                .unwrap_or("-1")
                .parse::<i64>()
                .unwrap_or(-1);
            if spec.rev {
                self.zrevrange(key, s_start, s_stop)
            } else {
                self.zrange(key, s_start, s_stop)
            }
        }
    }

    /// ZPOPMIN key [count]（单批次读取并删除极小值，原子高效）
    pub fn zpopmin<K: AsRef<[u8]>>(&self, key: K, count: usize) -> Result<Vec<ZSetMemberScore>> {
        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.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        let mut meta = match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                let m = ZSetMeta::decode(&m_bytes).unwrap_or_else(|| ZSetMeta::new(0, 0, 0));
                if m.is_expired(now_ms) || m.size() == 0 {
                    return Ok(Vec::new());
                }
                m
            }
            None => return Ok(Vec::new()),
        };

        let prefix = kc.zset_score_prefix(k_str);
        let mut popped = Vec::with_capacity(count.min(meta.size() as usize));
        let mut batch = self.db.batch();

        for g in self.data_ks.prefix(&prefix) {
            let (k, _) = g.into_inner()?;
            if !k.starts_with(&prefix) {
                break;
            }
            if let Some((score, member)) = parse_score_sub(&k[prefix.len()..]) {
                let m_key = kc.zset_key_bytes(k_str, member);
                batch.remove(&self.data_ks, &*k);
                batch.remove(&self.data_ks, m_key);
                popped.push((member.to_vec(), score));
                if popped.len() >= count {
                    break;
                }
            }
        }

        if !popped.is_empty() {
            meta.base.size = meta.base.size.saturating_sub(popped.len() as u64);
            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(popped)
    }

    /// ZPOPMAX key [count]（基于逆序范围精准截取，避免全量堆分配，原子高效）
    pub fn zpopmax<K: AsRef<[u8]>>(&self, key: K, count: usize) -> Result<Vec<ZSetMemberScore>> {
        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.zset_meta(k_str);
        let now_ms = ts_::sec() * 1000;

        let mut meta = match self.meta_ks.get(meta_k.as_bytes())? {
            Some(m_bytes) => {
                let m = ZSetMeta::decode(&m_bytes).unwrap_or_else(|| ZSetMeta::new(0, 0, 0));
                if m.is_expired(now_ms) || m.size() == 0 {
                    return Ok(Vec::new());
                }
                m
            }
            None => return Ok(Vec::new()),
        };

        let num_pop = count.min(meta.size() as usize);
        if num_pop == 0 {
            return Ok(Vec::new());
        }

        let to_pop = self.zrevrange(&key, 0, (num_pop as i64) - 1)?;
        if to_pop.is_empty() {
            return Ok(Vec::new());
        }

        let mut batch = self.db.batch();
        for (member, score) in &to_pop {
            let m_key = kc.zset_key_bytes(k_str, member);
            let s_key = kc.zset_score_key_bytes(k_str, *score, member);
            batch.remove(&self.data_ks, s_key);
            batch.remove(&self.data_ks, m_key);
        }

        meta.base.size = meta.base.size.saturating_sub(to_pop.len() as u64);
        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(to_pop)
    }

    /// BZPOPMIN key [key ...] (检查多键并弹出第一个非空的最小值)
    pub fn bzpopmin<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<Option<ZSetKeyMemberScore>> {
        for k in keys {
            let popped = self.zpopmin(k, 1)?;
            if let Some((member, score)) = popped.into_iter().next() {
                return Ok(Some((k.as_ref().to_vec(), member, score)));
            }
        }
        Ok(None)
    }

    /// BZPOPMAX key [key ...] (检查多键并弹出第一个非空的最大值)
    pub fn bzpopmax<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<Option<ZSetKeyMemberScore>> {
        for k in keys {
            let popped = self.zpopmax(k, 1)?;
            if let Some((member, score)) = popped.into_iter().next() {
                return Ok(Some((k.as_ref().to_vec(), member, score)));
            }
        }
        Ok(None)
    }

    /// ZRANDMEMBER key [count] (对标 Apache Kvrocks ExtractRandMemberFromSet)
    pub fn zrandmember<K: AsRef<[u8]>>(&self, key: K, count: i64) -> Result<Vec<ZSetMemberScore>> {
        if count == 0 {
            return Ok(Vec::new());
        }
        let all = self.zget_all(key)?;
        let total = all.len();
        if total == 0 {
            return Ok(Vec::new());
        }

        if count > 0 {
            let num = (count as usize).min(total);
            if num == total {
                return Ok(all);
            }
            // Fisher-Yates 局部置乱无放回抽样
            let mut indices: Vec<usize> = (0..total).collect();
            for i in 0..num {
                let j = fastrand::usize(i..total);
                indices.swap(i, j);
            }
            let mut results = Vec::with_capacity(num);
            for &idx in &indices[..num] {
                results.push(all[idx].clone());
            }
            Ok(results)
        } else {
            // 有放回独立随机抽样
            let num = count.unsigned_abs() as usize;
            let mut results = Vec::with_capacity(num);
            for _ in 0..num {
                let idx = fastrand::usize(0..total);
                results.push(all[idx].clone());
            }
            Ok(results)
        }
    }

    /// ZREMRANGEBYRANK key start stop
    pub fn zremrangebyrank<K: AsRef<[u8]>>(&self, key: K, start: i64, stop: i64) -> Result<usize> {
        let range = self.zrange(&key, start, stop)?;
        if range.is_empty() {
            return Ok(0);
        }
        let members: Vec<Vec<u8>> = range.into_iter().map(|(m, _)| m).collect();
        self.zrem(&key, &members)
    }

    /// ZREMRANGEBYSCORE key min max
    pub fn zremrangebyscore<K: AsRef<[u8]>>(&self, key: K, spec: &RangeScoreSpec) -> Result<usize> {
        let range = self.zrangebyscore(&key, spec)?;
        if range.is_empty() {
            return Ok(0);
        }
        let members: Vec<Vec<u8>> = range.into_iter().map(|(m, _)| m).collect();
        self.zrem(&key, &members)
    }

    /// ZREMRANGEBYLEX key min max
    pub fn zremrangebylex<K: AsRef<[u8]>>(&self, key: K, spec: &RangeLexSpec) -> Result<usize> {
        let members = self.zrangebylex(&key, spec)?;
        if members.is_empty() {
            return Ok(0);
        }
        self.zrem(&key, &members)
    }

    /// 覆盖写入有序集合数据（对标 Apache Kvrocks ZSet::Overwrite）
    pub fn overwrite_zset<K: AsRef<[u8]>, M: AsRef<[u8]>>(
        &self,
        key: K,
        score_members: &[(M, f64)],
    ) -> Result<usize> {
        let kc = KeyComposer::new("default");
        let k_str = str::from_utf8(key.as_ref()).unwrap_or("");
        let z_prefix = kc.zset_prefix(k_str);
        let zs_prefix = kc.zset_score_prefix(k_str);

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

        let mut seen = HashSet::with_capacity_and_hasher(score_members.len(), Default::default());
        let mut count = 0u64;

        let mut m_key = Vec::with_capacity(z_prefix.len() + 32);
        let mut s_key = Vec::with_capacity(zs_prefix.len() + HEX_SCORE_LEN + 1 + 32);

        for (member, score) in score_members {
            let m_bytes = member.as_ref();
            if !seen.insert(m_bytes) {
                continue;
            }
            let enc = encode_sortable_f64(*score);
            let hex_bytes = bytes_to_hex_16(enc);

            s_key.clear();
            s_key.extend_from_slice(&zs_prefix);
            s_key.extend_from_slice(&hex_bytes);
            s_key.push(b':');
            s_key.extend_from_slice(m_bytes);

            m_key.clear();
            m_key.extend_from_slice(&z_prefix);
            m_key.extend_from_slice(m_bytes);

            batch.insert(&self.data_ks, &s_key, b"");
            batch.insert(&self.data_ks, &m_key, enc);
            count += 1;
        }

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

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

    /// ZDIFF numkeys key [key ...] [WITHSCORES]
    pub fn zdiff<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<Vec<ZSetMemberScore>> {
        if keys.is_empty() {
            return Ok(Vec::new());
        }
        let first_card = self.zcard(&keys[0])?;
        if first_card == 0 {
            return Ok(Vec::new());
        }
        if keys.len() == 1 {
            return self.zget_all(&keys[0]);
        }

        let mut exclude: HashSet<Vec<u8>> = HashSet::default();
        for k in &keys[1..] {
            self.ziter(k, |m, _| {
                exclude.insert(m.to_vec());
                true
            })?;
        }

        let mut diff = Vec::new();
        self.ziter(&keys[0], |m, score| {
            if !exclude.contains(m) {
                diff.push((m.to_vec(), score));
            }
            true
        })?;
        Ok(diff)
    }

    /// ZDIFFSTORE destination numkeys key [key ...]
    pub fn zdiffstore<D: AsRef<[u8]>, K: AsRef<[u8]>>(&self, dst: D, keys: &[K]) -> Result<usize> {
        let diff = self.zdiff(keys)?;
        self.overwrite_zset(dst, &diff)
    }

    /// ZUNION numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE <SUM | MIN | MAX>]
    pub fn zunion<K: AsRef<[u8]>>(
        &self,
        keys_weights: &[(K, f64)],
        aggregate: Aggregate,
    ) -> Result<Vec<ZSetMemberScore>> {
        if keys_weights.is_empty() {
            return Ok(Vec::new());
        }

        let mut map: HashMap<Vec<u8>, f64> = HashMap::default();

        for (k, weight) in keys_weights {
            self.ziter(k, |member, score| {
                let mut weighted_score = score * weight;
                if weighted_score.is_nan() {
                    weighted_score = 0.0;
                }
                match map.get_mut(member) {
                    Some(cur_score) => {
                        *cur_score = aggregate.apply(*cur_score, weighted_score);
                    }
                    None => {
                        map.insert(member.to_vec(), weighted_score);
                    }
                }
                true
            })?;
        }

        let mut results: Vec<ZSetMemberScore> = map.into_iter().collect();
        results.sort_by(|a, b| a.1.total_cmp(&b.1).then_with(|| a.0.cmp(&b.0)));
        Ok(results)
    }

    /// ZUNIONSTORE destination numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE <SUM | MIN | MAX>]
    pub fn zunionstore<D: AsRef<[u8]>, K: AsRef<[u8]>>(
        &self,
        dst: D,
        keys_weights: &[(K, f64)],
        aggregate: Aggregate,
    ) -> Result<usize> {
        let union_res = self.zunion(keys_weights, aggregate)?;
        self.overwrite_zset(dst, &union_res)
    }

    /// ZINTER numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE <SUM | MIN | MAX>]
    pub fn zinter<K: AsRef<[u8]>>(
        &self,
        keys_weights: &[(K, f64)],
        aggregate: Aggregate,
    ) -> Result<Vec<ZSetMemberScore>> {
        if keys_weights.is_empty() {
            return Ok(Vec::new());
        }

        // 基数检测：选取基数最小的 key 作为基准集，若任一 key 为空则交集立即可知为空
        let mut min_idx = 0;
        let mut min_card = u64::MAX;
        for (i, (k, _)) in keys_weights.iter().enumerate() {
            let card = self.zcard(k)?;
            if card == 0 {
                return Ok(Vec::new());
            }
            if card < min_card {
                min_card = card;
                min_idx = i;
            }
        }

        let (base_k, base_w) = &keys_weights[min_idx];
        let base_items = self.zget_all(base_k)?;
        if base_items.is_empty() {
            return Ok(Vec::new());
        }

        let mut current_map: HashMap<Vec<u8>, f64> =
            HashMap::with_capacity_and_hasher(base_items.len(), Default::default());

        for (m, s) in base_items {
            let mut score = s * base_w;
            if score.is_nan() {
                score = 0.0;
            }
            current_map.insert(m, score);
        }

        // 对其余 keys 依次求交集
        for (i, (k, weight)) in keys_weights.iter().enumerate() {
            if i == min_idx {
                continue;
            }
            current_map.retain(|member, cur_score| match self.zscore(k, member) {
                Ok(Some(score)) => {
                    let mut weighted = score * weight;
                    if weighted.is_nan() {
                        weighted = 0.0;
                    }
                    *cur_score = aggregate.apply(*cur_score, weighted);
                    true
                }
                _ => false,
            });
            if current_map.is_empty() {
                return Ok(Vec::new());
            }
        }

        let mut results: Vec<ZSetMemberScore> = current_map.into_iter().collect();
        results.sort_by(|a, b| a.1.total_cmp(&b.1).then_with(|| a.0.cmp(&b.0)));
        Ok(results)
    }

    /// ZINTERSTORE destination numkeys key [key ...] [WEIGHTS weight [weight ...]] [AGGREGATE <SUM | MIN | MAX>]
    pub fn zinterstore<D: AsRef<[u8]>, K: AsRef<[u8]>>(
        &self,
        dst: D,
        keys_weights: &[(K, f64)],
        aggregate: Aggregate,
    ) -> Result<usize> {
        let inter_res = self.zinter(keys_weights, aggregate)?;
        self.overwrite_zset(dst, &inter_res)
    }

    /// ZINTERCARD numkeys key [key ...] [LIMIT limit]（基于基数升序优先扫描与 O(1) 存在性探针，提前中断）
    pub fn zintercard<K: AsRef<[u8]>>(&self, keys: &[K], limit: usize) -> Result<usize> {
        if keys.is_empty() {
            return Ok(0);
        }
        let mut key_cards: Vec<(&K, u64)> = Vec::with_capacity(keys.len());
        for k in keys {
            let card = self.zcard(k)?;
            if card == 0 {
                return Ok(0);
            }
            key_cards.push((k, card));
        }

        key_cards.sort_by_key(|(_, card)| *card);

        let (smallest_key, _) = key_cards[0];
        let kc = KeyComposer::new("default");

        let other_prefixes: Vec<Vec<u8>> = key_cards[1..]
            .iter()
            .map(|(k, _)| {
                let k_str = str::from_utf8(k.as_ref()).unwrap_or("");
                kc.zset_prefix(k_str)
            })
            .collect();

        let mut cardinality = 0;
        let mut probe_buf = Vec::new();

        self.ziter(smallest_key, |member, _| {
            let in_all = other_prefixes.iter().all(|prefix| {
                probe_buf.clear();
                probe_buf.extend_from_slice(prefix);
                probe_buf.extend_from_slice(member);
                self.data_ks.contains_key(&probe_buf).unwrap_or(false)
            });

            if in_all {
                cardinality += 1;
                if limit > 0 && cardinality >= limit {
                    return false;
                }
            }
            true
        })?;

        Ok(cardinality)
    }

    /// ZSCAN key cursor [MATCH pattern] [COUNT count] (流式分页扫描，低内存占用)
    pub fn zscan<K: AsRef<[u8]>>(
        &self,
        key: K,
        cursor: u64,
        pattern: Option<&[u8]>,
        count: Option<usize>,
    ) -> Result<ZScanResult> {
        let total = self.zcard(&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.ziter(key, |item, score| {
            if current_idx >= start && current_idx < end {
                if let Some(pat) = pattern {
                    if matches_glob_bytes(pat, item) {
                        results.push((item.to_vec(), score));
                    }
                } else {
                    results.push((item.to_vec(), score));
                }
            }
            current_idx += 1;
            current_idx < end
        })?;

        Ok((next_cursor, results))
    }
}

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

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

        // 1. ZADD & ZCARD & ZSCORE & ZMSCORE & ZMGET
        assert_eq!(
            db.zadd(
                "z1",
                &[(10.0, "alice"), (20.0, "bob"), (30.0, "charlie")],
                []
            )?,
            3
        );
        assert_eq!(db.zcard("z1")?, 3);
        assert_eq!(db.zscore("z1", "bob")?, Some(20.0));
        assert_eq!(
            db.zmscore("z1", &["alice", "david", "charlie"])?,
            vec![Some(10.0), None, Some(30.0)]
        );
        let mget = db.zmget("z1", &["alice", "david", "bob"])?;
        assert_eq!(mget.len(), 2);
        assert_eq!(mget.get(b"alice".as_slice()), Some(&10.0));
        assert_eq!(mget.get(b"bob".as_slice()), Some(&20.0));

        // 2. ZADD options (NX, XX, GT, LT, CH, INCR)
        // NX on existing -> no op
        assert_eq!(db.zadd("z1", &[(15.0, "bob")], [ZAdd::Nx])?, 0);
        assert_eq!(db.zscore("z1", "bob")?, Some(20.0));

        // XX on existing -> update
        assert_eq!(db.zadd("z1", &[(25.0, "bob")], [ZAdd::Xx, ZAdd::Ch])?, 1);
        assert_eq!(db.zscore("z1", "bob")?, Some(25.0));

        // GT on lower score -> ignored
        assert_eq!(db.zadd("z1", &[(20.0, "bob")], [ZAdd::Gt])?, 0);
        assert_eq!(db.zscore("z1", "bob")?, Some(25.0));

        // GT on higher score -> updated
        assert_eq!(db.zadd("z1", &[(35.0, "bob")], [ZAdd::Gt, ZAdd::Ch])?, 1);
        assert_eq!(db.zscore("z1", "bob")?, Some(35.0));

        // LT on higher score -> ignored
        assert_eq!(db.zadd("z1", &[(40.0, "bob")], [ZAdd::Lt])?, 0);
        assert_eq!(db.zscore("z1", "bob")?, Some(35.0));

        // LT on lower score -> updated
        assert_eq!(db.zadd("z1", &[(22.0, "bob")], [ZAdd::Lt, ZAdd::Ch])?, 1);
        assert_eq!(db.zscore("z1", "bob")?, Some(22.0));

        // INCR option
        assert_eq!(db.zadd("z1", &[(5.0, "alice")], [ZAdd::Incr, ZAdd::Ch])?, 1);
        assert_eq!(db.zscore("z1", "alice")?, Some(15.0));

        // 3. ZRANK & ZREVRANK & WithScore
        // Current scores: alice(15), bob(22), charlie(30)
        assert_eq!(db.zrank("z1", "alice")?, Some(0));
        assert_eq!(db.zrank("z1", "bob")?, Some(1));
        assert_eq!(db.zrank("z1", "charlie")?, Some(2));
        assert_eq!(db.zrank("z1", "nonexistent")?, None);

        assert_eq!(db.zrank_with_score("z1", "bob")?, Some((1, 22.0)));
        assert_eq!(db.zrevrank_with_score("z1", "bob")?, Some((1, 22.0)));

        assert_eq!(db.zrevrank("z1", "charlie")?, Some(0));
        assert_eq!(db.zrevrank("z1", "bob")?, Some(1));
        assert_eq!(db.zrevrank("z1", "alice")?, Some(2));

        // 4. ZRANGE & ZREVRANGE & Negative Indices
        let range = db.zrange("z1", 0, -1)?;
        assert_eq!(range.len(), 3);
        assert_eq!(range[0].0, b"alice");
        assert_eq!(range[2].0, b"charlie");

        let revrange = db.zrevrange("z1", 0, 1)?;
        assert_eq!(revrange.len(), 2);
        assert_eq!(revrange[0].0, b"charlie");
        assert_eq!(revrange[1].0, b"bob");

        // Edge negative index
        assert_eq!(db.zrange("z1", 0, -5)?, Vec::new());

        // 5. ZRANGEBYSCORE & ZCOUNT
        let spec = RangeScoreSpec {
            min: 20.0,
            max: 40.0,
            ..Default::default()
        };
        assert_eq!(db.zcount("z1", &spec)?, 2);
        let byscore = db.zrangebyscore("z1", &spec)?;
        assert_eq!(byscore.len(), 2);

        // 6. ZPOPMIN & ZPOPMAX & BZPOPMIN & BZPOPMAX
        let bzmin = db.bzpopmin(&["empty_key", "z1"])?;
        assert_eq!(bzmin, Some((b"z1".to_vec(), b"alice".to_vec(), 15.0)));
        assert_eq!(db.zcard("z1")?, 2);

        let popmax = db.zpopmax("z1", 1)?;
        assert_eq!(popmax, vec![(b"charlie".to_vec(), 30.0)]);
        assert_eq!(db.zcard("z1")?, 1);

        // 7. ZUNION & ZINTER & ZDIFF
        db.zadd("zA", &[(1.0, "x"), (2.0, "y")], [])?;
        db.zadd("zB", &[(3.0, "y"), (4.0, "z")], [])?;

        let diff = db.zdiff(&["zA", "zB"])?;
        assert_eq!(diff, vec![(b"x".to_vec(), 1.0)]);

        let union_res = db.zunion(&[("zA", 1.0), ("zB", 2.0)], Aggregate::Sum)?;
        // x: 1.0, y: 2.0 + 3.0*2.0 = 8.0, z: 4.0*2.0 = 8.0
        assert_eq!(union_res.len(), 3);

        let inter_res = db.zinter(&[("zA", 1.0), ("zB", 1.0)], Aggregate::Max)?;
        assert_eq!(inter_res, vec![(b"y".to_vec(), 3.0)]);
        assert_eq!(db.zintercard(&["zA", "zB"], 0)?, 1);

        // 8. ZRANDMEMBER
        let r_pos = db.zrandmember("zA", 2)?;
        assert_eq!(r_pos.len(), 2);
        let r_neg = db.zrandmember("zA", -5)?;
        assert_eq!(r_neg.len(), 5);

        // 9. ZRANGE_SPEC (Unified ZRANGE)
        let spec_score = ZRangeSpec {
            by_score: true,
            ..Default::default()
        };
        let r_spec = db.zrange_spec("zA", b"0.5", b"1.5", &spec_score)?;
        assert_eq!(r_spec, vec![(b"x".to_vec(), 1.0)]);

        Ok(())
    }
}