wedb_embed 0.1.1

pub mod conf;
pub mod meta;

pub use conf::{
    ERR_MAX_NOT_INT, ERR_MIN_GT_MAX, ERR_MIN_NOT_INT, ERR_WRONG_TYPE, HEX_CHARS, HEX_LEN,
    SortedintRangeSpec, decode_be_u64, decode_hex_u64, encode_be_u64, encode_hex_u64,
    parse_range_spec,
};
pub use meta::SortedintMeta;

use rapidhash::RapidHashSet as HashSet;

use crate::db::WeDb;
use crate::error::Result;
use crate::key_composer::{KeyComposer, KeyTag};
use crate::meta::{current_now_ms, normalize_range};

/// 默认命名空间常量
const DEFAULT_NS: &str = "default";

/// 构造有序整型元数据存储键（二进制安全）
#[inline]
fn compose_si_meta_key(kc: &KeyComposer, key: &[u8]) -> Vec<u8> {
    kc.compose_meta_key(KeyTag::SortedIntMeta.as_slice(), key)
}

/// 构造有序整型子键前缀（二进制安全，格式为 `\x00si:{key}:` 或 `\x00ns:{ns}:si:{key}:`）
#[inline]
fn compose_si_prefix(kc: &KeyComposer, key: &[u8]) -> Vec<u8> {
    kc.compose_prefix(KeyTag::SortedIntData.as_slice(), key)
}

/// 基于前缀字节切片构造元素存储键（预分配容量，零二次分配）
#[inline]
fn compose_si_item_key(prefix: &[u8], id: u64) -> Vec<u8> {
    let mut out = Vec::with_capacity(prefix.len() + HEX_LEN);
    out.extend_from_slice(prefix);
    let hex = encode_hex_u64(id);
    out.extend_from_slice(&hex);
    out
}

/// 从存储键字节切片中提取 64 位整数 ID（零堆分配，自适应 16 字节十六进制与 8 字节大端序原生二进制）
#[inline(always)]
fn extract_id(key_bytes: &[u8], prefix_len: usize) -> Option<u64> {
    let sub = key_bytes.get(prefix_len..)?;
    if let Some(hex_slice) = sub.get(..HEX_LEN)
        && let Some(id) = decode_hex_u64(hex_slice)
    {
        return Some(id);
    }
    if let Some(be_slice) = sub.get(..8) {
        decode_be_u64(be_slice)
    } else {
        None
    }
}

impl WeDb {
    #[inline]
    fn get_si_meta_checked(
        &self,
        kc: &KeyComposer<'_>,
        k: &[u8],
        mk: &[u8],
        now: u64,
    ) -> Result<Option<SortedintMeta>> {
        self.get_meta_checked(kc, k, mk, now)
    }

    /// 准备写入时的元数据（若旧集合已过期或被覆盖则自动在批处理中清理残留子键与 String 键）
    #[inline]
    fn prepare_si_meta_for_write(
        &self,
        kc: &KeyComposer,
        k_bytes: &[u8],
        prefix: &[u8],
        meta_k: &[u8],
        now_ms: u64,
        batch: &mut fjall::OwnedWriteBatch,
    ) -> Result<(SortedintMeta, bool)> {
        match self.meta.get(meta_k)? {
            Some(m_bytes) => match SortedintMeta::decode(&m_bytes) {
                Some(meta) => {
                    if !meta.is_expired(now_ms) {
                        return Ok((meta, false));
                    }
                    // 已过期：清理旧版本全部子键与同名 String 键
                    self.clear_prefix_in_batch(prefix, batch)?;
                    let raw_k = kc.raw_key_bytes(k_bytes);
                    batch.remove(&self.data, &*raw_k);
                    Ok((SortedintMeta::default(), true))
                }
                None => {
                    self.check_key_not_other_type(
                        kc,
                        k_bytes,
                        KeyTag::SortedIntMeta.as_slice(),
                        now_ms,
                    )?;
                    let raw_k = kc.raw_key_bytes(k_bytes);
                    batch.remove(&self.data, &*raw_k);
                    Ok((SortedintMeta::default(), true))
                }
            },
            None => {
                self.check_key_not_other_type(
                    kc,
                    k_bytes,
                    KeyTag::SortedIntMeta.as_slice(),
                    now_ms,
                )?;
                let raw_k = kc.raw_key_bytes(k_bytes);
                batch.remove(&self.data, &*raw_k);
                Ok((SortedintMeta::default(), true))
            }
        }
    }

    /// 流式升序遍历有序整型集合（零额外内存分配，借用底层迭代器切片，可随时提前终止）
    #[inline]
    pub fn si_iter<K: AsRef<[u8]>, F: FnMut(u64) -> bool>(&self, key: K, f: F) -> Result<()> {
        self.si_iter_ns(DEFAULT_NS, key, f)
    }

    /// 流式升序遍历指定命名空间下的有序整型集合
    #[inline]
    pub fn si_iter_ns<K: AsRef<[u8]>, F: FnMut(u64) -> bool>(
        &self,
        ns: &str,
        key: K,
        mut f: F,
    ) -> Result<()> {
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        if self
            .get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)?
            .is_none()
        {
            return Ok(());
        }

        let prefix = compose_si_prefix(&kc, k_bytes);
        let prefix_len = prefix.len();

        for g in self.data.prefix(&prefix) {
            let (k, _) = g.into_inner()?;
            if !k.starts_with(&prefix) {
                break;
            }
            if let Some(id) = extract_id(&k, prefix_len)
                && !f(id)
            {
                break;
            }
        }
        Ok(())
    }

    /// SIADD key id [id ...] (对标 Apache Kvrocks Sortedint::Add)
    pub fn si_add<K: AsRef<[u8]>>(&self, key: K, ids: &[u64]) -> Result<usize> {
        self.si_add_ns(DEFAULT_NS, key, ids)
    }

    /// SIADD key id [id ...] (支持自定义命名空间)
    pub fn si_add_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K, ids: &[u64]) -> Result<usize> {
        if ids.is_empty() {
            return Ok(0);
        }
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let prefix = compose_si_prefix(&kc, k_bytes);
        let now_ms = current_now_ms();

        let mut batch = self.db.batch();
        let (mut meta, is_fresh) =
            self.prepare_si_meta_for_write(&kc, k_bytes, &prefix, &meta_k, now_ms, &mut batch)?;

        let mut added = 0usize;
        let prefix_len = prefix.len();
        let mut item_buf = Vec::with_capacity(prefix_len + HEX_LEN);
        item_buf.extend_from_slice(&prefix);
        item_buf.resize(prefix_len + HEX_LEN, 0);

        if ids.len() == 1 {
            let id = ids[0];
            let hex_bytes = encode_hex_u64(id);
            item_buf[prefix_len..].copy_from_slice(&hex_bytes);
            if is_fresh || !self.data.contains_key(&item_buf)? {
                added = 1;
                meta.base.size = meta.base.size.saturating_add(1);
                batch.insert(&self.data, &item_buf, b"");
            }
        } else {
            let mut seen = HashSet::with_capacity_and_hasher(ids.len(), Default::default());
            for &id in ids {
                if !seen.insert(id) {
                    continue;
                }
                let hex_bytes = encode_hex_u64(id);
                item_buf[prefix_len..].copy_from_slice(&hex_bytes);
                if is_fresh || !self.data.contains_key(&item_buf)? {
                    added += 1;
                    meta.base.size = meta.base.size.saturating_add(1);
                    batch.insert(&self.data, &item_buf, b"");
                }
            }
        }

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

        Ok(added)
    }

    /// SIREM key id [id ...] (对标 Apache Kvrocks Sortedint::Remove)
    pub fn si_rem<K: AsRef<[u8]>>(&self, key: K, ids: &[u64]) -> Result<usize> {
        self.si_rem_ns(DEFAULT_NS, key, ids)
    }

    /// SIREM key id [id ...] (支持自定义命名空间)
    pub fn si_rem_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K, ids: &[u64]) -> Result<usize> {
        if ids.is_empty() {
            return Ok(0);
        }
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        let mut meta = match self.get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(0),
        };

        let prefix = compose_si_prefix(&kc, k_bytes);
        let prefix_len = prefix.len();
        let mut item_buf = Vec::with_capacity(prefix_len + HEX_LEN);
        item_buf.extend_from_slice(&prefix);
        item_buf.resize(prefix_len + HEX_LEN, 0);

        let mut deleted = 0usize;
        let mut batch = self.db.batch();

        if ids.len() == 1 {
            let id = ids[0];
            let hex_bytes = encode_hex_u64(id);
            item_buf[prefix_len..].copy_from_slice(&hex_bytes);
            if self.data.contains_key(&item_buf)? {
                deleted = 1;
                meta.base.size = meta.base.size.saturating_sub(1);
                batch.remove(&self.data, &item_buf);
            }
        } else {
            let mut seen = HashSet::with_capacity_and_hasher(ids.len(), Default::default());
            for &id in ids {
                if !seen.insert(id) {
                    continue;
                }
                let hex_bytes = encode_hex_u64(id);
                item_buf[prefix_len..].copy_from_slice(&hex_bytes);
                if self.data.contains_key(&item_buf)? {
                    deleted += 1;
                    meta.base.size = meta.base.size.saturating_sub(1);
                    batch.remove(&self.data, &item_buf);
                }
            }
        }

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

        Ok(deleted)
    }

    /// SICARD key (对标 Apache Kvrocks Sortedint::Card)
    pub fn si_card<K: AsRef<[u8]>>(&self, key: K) -> Result<u64> {
        self.si_card_ns(DEFAULT_NS, key)
    }

    /// SICARD key (支持自定义命名空间)
    pub fn si_card_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K) -> Result<u64> {
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();
        Ok(self
            .get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)?
            .map_or(0, |m| m.base.size))
    }

    /// SIEXISTS key id (判断指定元素是否存在)
    pub fn si_exists<K: AsRef<[u8]>>(&self, key: K, id: u64) -> Result<bool> {
        self.si_exists_ns(DEFAULT_NS, key, id)
    }

    /// SIEXISTS key id (支持自定义命名空间)
    pub fn si_exists_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K, id: u64) -> Result<bool> {
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        if self
            .get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)?
            .is_none()
        {
            return Ok(false);
        }

        let prefix = compose_si_prefix(&kc, k_bytes);
        let item_k = compose_si_item_key(&prefix, id);
        Ok(self.data.contains_key(&item_k)?)
    }

    /// SIISMEMBER key id (SIEXISTS 别名)
    #[inline]
    pub fn si_ismember<K: AsRef<[u8]>>(&self, key: K, id: u64) -> Result<bool> {
        self.si_exists(key, id)
    }

    /// SIISMEMBER key id (支持自定义命名空间)
    #[inline]
    pub fn si_ismember_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K, id: u64) -> Result<bool> {
        self.si_exists_ns(ns, key, id)
    }

    /// SIMEXIST / SIEXISTS key id [id ...] (对标 Apache Kvrocks Sortedint::MExist)
    pub fn si_mexist<K: AsRef<[u8]>>(&self, key: K, ids: &[u64]) -> Result<Vec<bool>> {
        self.si_mexist_ns(DEFAULT_NS, key, ids)
    }

    /// SIMEXIST / SIEXISTS key id [id ...] (支持自定义命名空间)
    pub fn si_mexist_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K, ids: &[u64]) -> Result<Vec<bool>> {
        let mut results = Vec::with_capacity(ids.len());
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        if self
            .get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)?
            .is_none()
        {
            results.resize(ids.len(), false);
            return Ok(results);
        }

        let prefix = compose_si_prefix(&kc, k_bytes);
        let prefix_len = prefix.len();
        let mut item_buf = Vec::with_capacity(prefix_len + HEX_LEN);
        item_buf.extend_from_slice(&prefix);
        item_buf.resize(prefix_len + HEX_LEN, 0);

        for &id in ids {
            let hex_bytes = encode_hex_u64(id);
            item_buf[prefix_len..].copy_from_slice(&hex_bytes);
            results.push(self.data.contains_key(&item_buf)?);
        }
        Ok(results)
    }

    /// SIMEMBERS key (获取有序集合中的所有整型元素，按数值升序排列)
    pub fn si_members<K: AsRef<[u8]>>(&self, key: K) -> Result<Vec<u64>> {
        self.si_members_ns(DEFAULT_NS, key)
    }

    /// SIMEMBERS key (支持自定义命名空间)
    pub fn si_members_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K) -> Result<Vec<u64>> {
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        let meta = match self.get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(Vec::new()),
        };

        let mut results = Vec::with_capacity((meta.base.size as usize).min(4096));
        self.si_iter_ns(ns, key, |id| {
            results.push(id);
            true
        })?;
        Ok(results)
    }

    /// SIRANGE key cursor offset limit reversed (对标 Apache Kvrocks Sortedint::Range)
    pub fn si_range<K: AsRef<[u8]>>(
        &self,
        key: K,
        cursor: u64,
        offset: usize,
        limit: usize,
        reversed: bool,
    ) -> Result<Vec<u64>> {
        self.si_range_ns(DEFAULT_NS, key, cursor, offset, limit, reversed)
    }

    /// SIRANGE key cursor offset limit reversed (支持自定义命名空间)
    pub fn si_range_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        cursor: u64,
        offset: usize,
        limit: usize,
        reversed: bool,
    ) -> Result<Vec<u64>> {
        if limit == 0 {
            return Ok(Vec::new());
        }
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        if self
            .get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)?
            .is_none()
        {
            return Ok(Vec::new());
        }

        let prefix = compose_si_prefix(&kc, k_bytes);
        let prefix_len = prefix.len();

        if !reversed {
            let start_k = compose_si_item_key(&prefix, cursor);
            let end_k = compose_si_item_key(&prefix, u64::MAX);
            let mut results = Vec::with_capacity(limit.min(1024));
            let mut pos = 0usize;

            for g in self.data.range(start_k..=end_k) {
                let (k, _) = g.into_inner()?;
                if let Some(id) = extract_id(&k, prefix_len) {
                    if cursor > 0 && id == cursor {
                        continue;
                    }
                    if pos < offset {
                        pos += 1;
                        continue;
                    }
                    results.push(id);
                    if results.len() >= limit {
                        break;
                    }
                }
            }
            Ok(results)
        } else {
            let start_k = compose_si_item_key(&prefix, 0);
            let end_k = compose_si_item_key(&prefix, if cursor == 0 { u64::MAX } else { cursor });
            let mut results = Vec::with_capacity(limit.min(1024));
            let mut pos = 0usize;

            for g in self.data.range(start_k..=end_k).rev() {
                let (k, _) = g.into_inner()?;
                if let Some(id) = extract_id(&k, prefix_len) {
                    if cursor > 0 && id == cursor {
                        continue;
                    }
                    if pos < offset {
                        pos += 1;
                        continue;
                    }
                    results.push(id);
                    if results.len() >= limit {
                        break;
                    }
                }
            }
            Ok(results)
        }
    }

    /// SIREVRANGE key cursor offset limit (逆序分页扫描)
    #[inline]
    pub fn si_rev_range<K: AsRef<[u8]>>(
        &self,
        key: K,
        cursor: u64,
        offset: usize,
        limit: usize,
    ) -> Result<Vec<u64>> {
        self.si_range_ns(DEFAULT_NS, key, cursor, offset, limit, true)
    }

    /// SIREVRANGE key cursor offset limit (支持自定义命名空间)
    #[inline]
    pub fn si_rev_range_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        cursor: u64,
        offset: usize,
        limit: usize,
    ) -> Result<Vec<u64>> {
        self.si_range_ns(ns, key, cursor, offset, limit, true)
    }

    /// SIRANGEBYVALUE key spec (对标 Apache Kvrocks Sortedint::RangeByValue)
    pub fn si_range_by_value<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<Vec<u64>> {
        self.si_range_by_value_ns(DEFAULT_NS, key, spec)
    }

    /// SIRANGEBYVALUE key spec (支持自定义命名空间)
    pub fn si_range_by_value_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<Vec<u64>> {
        if spec.is_empty_range() {
            return Ok(Vec::new());
        }
        if let Some(0) = spec.count {
            return Ok(Vec::new());
        }

        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        if self
            .get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)?
            .is_none()
        {
            return Ok(Vec::new());
        }

        let prefix = compose_si_prefix(&kc, k_bytes);
        let prefix_len = prefix.len();
        let start_k = compose_si_item_key(&prefix, spec.min);
        let end_k = compose_si_item_key(&prefix, spec.max);

        if !spec.reversed {
            let mut results = Vec::with_capacity(spec.count.unwrap_or(16).min(1024));
            let mut pos = 0usize;
            for g in self.data.range(start_k..=end_k) {
                let (k, _) = g.into_inner()?;
                if let Some(id) = extract_id(&k, prefix_len) {
                    if spec.minex && id == spec.min {
                        continue;
                    }
                    if spec.maxex && id == spec.max {
                        break;
                    }
                    if pos < spec.offset {
                        pos += 1;
                        continue;
                    }
                    results.push(id);
                    if let Some(cnt) = spec.count
                        && results.len() >= cnt
                    {
                        break;
                    }
                }
            }
            Ok(results)
        } else {
            let mut results = Vec::with_capacity(spec.count.unwrap_or(16).min(1024));
            let mut pos = 0usize;
            for g in self.data.range(start_k..=end_k).rev() {
                let (k, _) = g.into_inner()?;
                if let Some(id) = extract_id(&k, prefix_len) {
                    if spec.maxex && id == spec.max {
                        continue;
                    }
                    if spec.minex && id == spec.min {
                        break;
                    }
                    if pos < spec.offset {
                        pos += 1;
                        continue;
                    }
                    results.push(id);
                    if let Some(cnt) = spec.count
                        && results.len() >= cnt
                    {
                        break;
                    }
                }
            }
            Ok(results)
        }
    }

    /// SIREVRANGEBYVALUE key spec (按数值逆序范围查询)
    #[inline]
    pub fn si_rev_range_by_value<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<Vec<u64>> {
        self.si_rev_range_by_value_ns(DEFAULT_NS, key, spec)
    }

    /// SIREVRANGEBYVALUE key spec (支持自定义命名空间)
    #[inline]
    pub fn si_rev_range_by_value_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<Vec<u64>> {
        let mut rev_spec = spec.clone();
        rev_spec.reversed = true;
        self.si_range_by_value_ns(ns, key, &rev_spec)
    }

    /// SIRANGEBYSCORE key spec (按分数/数值正序范围查询别名)
    #[inline]
    pub fn si_range_by_score<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<Vec<u64>> {
        self.si_range_by_value(key, spec)
    }

    /// SIRANGEBYSCORE key spec (支持自定义命名空间)
    #[inline]
    pub fn si_range_by_score_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<Vec<u64>> {
        self.si_range_by_value_ns(ns, key, spec)
    }

    /// SIREVRANGEBYSCORE key spec (按分数/数值逆序范围查询别名)
    #[inline]
    pub fn si_rev_range_by_score<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<Vec<u64>> {
        self.si_rev_range_by_value(key, spec)
    }

    /// SIREVRANGEBYSCORE key spec (支持自定义命名空间)
    #[inline]
    pub fn si_rev_range_by_score_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<Vec<u64>> {
        self.si_rev_range_by_value_ns(ns, key, spec)
    }

    /// SIREMRANGEBYVALUE key spec (按数值范围单次扫描高效批量移除元素，采用 seek 区间跳跃优化)
    pub fn si_rem_range_by_value<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<usize> {
        self.si_rem_range_by_value_ns(DEFAULT_NS, key, spec)
    }

    /// SIREMRANGEBYVALUE key spec (支持自定义命名空间)
    pub fn si_rem_range_by_value_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<usize> {
        if spec.is_empty_range() {
            return Ok(0);
        }
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        let mut meta = match self.get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(0),
        };

        let prefix = compose_si_prefix(&kc, k_bytes);
        let prefix_len = prefix.len();
        let start_k = compose_si_item_key(&prefix, spec.min);
        let end_k = compose_si_item_key(&prefix, spec.max);

        let mut deleted = 0usize;
        let mut batch = self.db.batch();

        for g in self.data.range(start_k..=end_k) {
            let (k, _) = g.into_inner()?;
            if let Some(id) = extract_id(&k, prefix_len) {
                if spec.minex && id == spec.min {
                    continue;
                }
                if spec.maxex && id == spec.max {
                    break;
                }
                deleted += 1;
                batch.remove(&self.data, &*k);
            }
        }

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

        Ok(deleted)
    }

    /// SIREMRANGEBYSCORE key spec (按数值范围移除元素别名)
    #[inline]
    pub fn si_rem_range_by_score<K: AsRef<[u8]>>(
        &self,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<usize> {
        self.si_rem_range_by_value(key, spec)
    }

    /// SIREMRANGEBYSCORE key spec (支持自定义命名空间)
    #[inline]
    pub fn si_rem_range_by_score_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<usize> {
        self.si_rem_range_by_value_ns(ns, key, spec)
    }

    /// SIREMRANGEBYRANK key start stop (按排名范围单次扫描移除元素，支持负索引)
    pub fn si_rem_range_by_rank<K: AsRef<[u8]>>(
        &self,
        key: K,
        start: i64,
        stop: i64,
    ) -> Result<usize> {
        self.si_rem_range_by_rank_ns(DEFAULT_NS, key, start, stop)
    }

    /// SIREMRANGEBYRANK key start stop (支持自定义命名空间)
    pub fn si_rem_range_by_rank_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        start: i64,
        stop: i64,
    ) -> Result<usize> {
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        let mut meta = match self.get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(0),
        };

        let card = meta.base.size as i64;
        if card == 0 {
            return Ok(0);
        }
        let (s, e) = normalize_range(start, stop, card);
        if s > e {
            return Ok(0);
        }
        let offset = s as usize;
        let limit = (e - s + 1) as usize;

        let prefix = compose_si_prefix(&kc, k_bytes);
        let prefix_len = prefix.len();

        let mut skipped = 0usize;
        let mut deleted = 0usize;
        let mut batch = self.db.batch();

        for g in self.data.prefix(&prefix) {
            let (k, _) = g.into_inner()?;
            if !k.starts_with(&prefix) {
                break;
            }
            if extract_id(&k, prefix_len).is_some() {
                if skipped < offset {
                    skipped += 1;
                    continue;
                }
                deleted += 1;
                batch.remove(&self.data, &*k);
                if deleted >= limit {
                    break;
                }
            }
        }

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

        Ok(deleted)
    }

    /// SIRANK key id (获取元素正序排名，0-indexed)
    pub fn si_rank<K: AsRef<[u8]>>(&self, key: K, id: u64) -> Result<Option<usize>> {
        self.si_rank_ns(DEFAULT_NS, key, id)
    }

    /// SIRANK key id (支持自定义命名空间)
    pub fn si_rank_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K, id: u64) -> Result<Option<usize>> {
        let mut rank = 0usize;
        let mut found = false;
        self.si_iter_ns(ns, key, |cur_id| {
            if cur_id == id {
                found = true;
                return false;
            }
            if cur_id > id {
                return false;
            }
            rank += 1;
            true
        })?;
        if found { Ok(Some(rank)) } else { Ok(None) }
    }

    /// SIREVRANK key id (获取元素逆序排名，0-indexed)
    pub fn si_revrank<K: AsRef<[u8]>>(&self, key: K, id: u64) -> Result<Option<usize>> {
        self.si_revrank_ns(DEFAULT_NS, key, id)
    }

    /// SIREVRANK key id (支持自定义命名空间)
    pub fn si_revrank_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        id: u64,
    ) -> Result<Option<usize>> {
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        let meta = match self.get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)? {
            Some(m) => m,
            None => return Ok(None),
        };

        if let Some(rank) = self.si_rank_ns(ns, key, id)? {
            Ok(Some(
                (meta.base.size as usize)
                    .saturating_sub(1)
                    .saturating_sub(rank),
            ))
        } else {
            Ok(None)
        }
    }

    /// SICOUNT key spec (统计指定范围内的元素数量，seek 区间直接跳跃，零堆分配)
    pub fn si_count<K: AsRef<[u8]>>(&self, key: K, spec: &SortedintRangeSpec) -> Result<usize> {
        self.si_count_ns(DEFAULT_NS, key, spec)
    }

    /// SICOUNT key spec (支持自定义命名空间)
    pub fn si_count_ns<K: AsRef<[u8]>>(
        &self,
        ns: &str,
        key: K,
        spec: &SortedintRangeSpec,
    ) -> Result<usize> {
        if spec.is_empty_range() {
            return Ok(0);
        }
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let now_ms = current_now_ms();

        if self
            .get_si_meta_checked(&kc, k_bytes, &meta_k, now_ms)?
            .is_none()
        {
            return Ok(0);
        }

        let prefix = compose_si_prefix(&kc, k_bytes);
        let prefix_len = prefix.len();
        let start_k = compose_si_item_key(&prefix, spec.min);
        let end_k = compose_si_item_key(&prefix, spec.max);

        let mut count = 0usize;
        for g in self.data.range(start_k..=end_k) {
            let (k, _) = g.into_inner()?;
            if let Some(id) = extract_id(&k, prefix_len) {
                if spec.minex && id == spec.min {
                    continue;
                }
                if spec.maxex && id == spec.max {
                    break;
                }
                count += 1;
            }
        }
        Ok(count)
    }

    /// 清理指定 Sortedint 集合的全部数据项及元数据
    pub fn si_clear<K: AsRef<[u8]>>(&self, key: K) -> Result<usize> {
        self.si_clear_ns(DEFAULT_NS, key)
    }

    /// 清理指定 Sortedint 集合的全部数据项及元数据（支持自定义命名空间）
    pub fn si_clear_ns<K: AsRef<[u8]>>(&self, ns: &str, key: K) -> Result<usize> {
        let k_bytes = key.as_ref();
        let kc = KeyComposer::new(ns);
        let meta_k = compose_si_meta_key(&kc, k_bytes);
        let prefix = compose_si_prefix(&kc, k_bytes);

        let mut deleted = 0usize;
        let mut batch = self.db.batch();

        for g in self.data.prefix(&prefix) {
            let (k, _) = g.into_inner()?;
            if !k.starts_with(&prefix) {
                break;
            }
            deleted += 1;
            batch.remove(&self.data, &*k);
        }

        let has_meta = self.meta.contains_key(&meta_k)?;
        if has_meta {
            batch.remove(&self.meta, &meta_k);
        }

        if deleted > 0 || has_meta {
            batch.commit()?;
        }

        Ok(deleted)
    }
}