wedb_embed 0.1.0

use crate::zset::encode_sortable_f64;
use memchr::{memchr, memrchr};
use std::borrow::Cow;
use std::str;

#[inline]
pub fn bytes_to_hex_16(bytes: [u8; 8]) -> [u8; 16] {
    const HEX_CHARS: &[u8; 16] = b"0123456789abcdef";
    let mut out = [0u8; 16];
    for (i, &b) in bytes.iter().enumerate() {
        out[i * 2] = HEX_CHARS[(b >> 4) as usize];
        out[i * 2 + 1] = HEX_CHARS[(b & 0x0f) as usize];
    }
    out
}

pub const DEFAULT_NAMESPACE: &str = "default";
pub const KVROCKS_DEFAULT_NAMESPACE: &str = "__namespace";
pub const KVROCKS_DB_NAMESPACE_PREFIX: &str = "db";
pub const KVROCKS_LEGACY_DB_NAMESPACE_PREFIX: &str = "__db_";

pub const HASH_SLOTS_MASK: u16 = 0x3fff;
pub const HASH_SLOTS_SIZE: u16 = HASH_SLOTS_MASK + 1; // 16384

#[rustfmt::skip]
const CRC16_TAB: [u16; 256] = [
    0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7, 0x8108, 0x9129, 0xa14a, 0xb16b, 0xc18c, 0xd1ad,
    0xe1ce, 0xf1ef, 0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6, 0x9339, 0x8318, 0xb37b, 0xa35a,
    0xd3bd, 0xc39c, 0xf3ff, 0xe3de, 0x2462, 0x3443, 0x0420, 0x1401, 0x64e6, 0x74c7, 0x44a4, 0x5485, 0xa56a, 0xb54b,
    0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d, 0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4,
    0xb75b, 0xa77a, 0x9719, 0x8738, 0xf7df, 0xe7fe, 0xd79d, 0xc7bc, 0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861,
    0x2802, 0x3823, 0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b, 0x5af5, 0x4ad4, 0x7ab7, 0x6a96,
    0x1a71, 0x0a50, 0x3a33, 0x2a12, 0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a, 0x6ca6, 0x7c87,
    0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41, 0xedae, 0xfd8f, 0xcdec, 0xddcd, 0xad2a, 0xbd0b, 0x8d68, 0x9d49,
    0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70, 0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a,
    0x9f59, 0x8f78, 0x9188, 0x81a9, 0xb1ca, 0xa1eb, 0xd10c, 0xc12d, 0xf14e, 0xe16f, 0x1080, 0x00a1, 0x30c2, 0x20e3,
    0x5004, 0x4025, 0x7046, 0x6067, 0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e, 0x02b1, 0x1290,
    0x22f3, 0x32d2, 0x4235, 0x5214, 0x6277, 0x7256, 0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d,
    0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405, 0xa7db, 0xb7fa, 0x8799, 0x97b8, 0xe75f, 0xf77e,
    0xc71d, 0xd73c, 0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634, 0xd94c, 0xc96d, 0xf90e, 0xe92f,
    0x99c8, 0x89e9, 0xb98a, 0xa9ab, 0x5844, 0x4865, 0x7806, 0x6827, 0x18c0, 0x08e1, 0x3882, 0x28a3, 0xcb7d, 0xdb5c,
    0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a, 0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92,
    0xfd2e, 0xed0f, 0xdd6c, 0xcd4d, 0xbdaa, 0xad8b, 0x9de8, 0x8dc9, 0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83,
    0x1ce0, 0x0cc1, 0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8, 0x6e17, 0x7e36, 0x4e55, 0x5e74,
    0x2e93, 0x3eb2, 0x0ed1, 0x1ef0,
];

/// 计算 Redis Cluster Slot CRC16 校验和（对标 Kvrocks Crc16）
#[inline]
pub fn crc16(buf: &[u8]) -> u16 {
    let mut crc = 0u16;
    for &b in buf {
        crc = (crc << 8) ^ CRC16_TAB[((crc >> 8) as u8 ^ b) as usize];
    }
    crc
}

/// 提取 Redis Hash Tag（对标 Kvrocks GetTagFromKey）
#[inline]
pub fn get_tag_from_key(key: &[u8]) -> &[u8] {
    if let Some(left_pos) = memchr(b'{', key)
        && let Some(right_pos) = memchr(b'}', &key[left_pos + 1..])
        && right_pos > 0
    {
        return &key[left_pos + 1..left_pos + 1 + right_pos];
    }
    b""
}

/// 计算 Key 的 Slot ID（对标 Kvrocks GetSlotIdFromKey）
#[inline]
pub fn get_slot_id_from_key(key: &[u8]) -> u16 {
    let tag = get_tag_from_key(key);
    let target = if tag.is_empty() { key } else { tag };
    crc16(target) & HASH_SLOTS_MASK
}

/// 构造 Slot Key 前缀（对标 Kvrocks ComposeSlotKeyPrefix）
#[inline]
pub fn compose_slot_key_prefix(ns: &[u8], slot_id: u16) -> Vec<u8> {
    let mut output = Vec::with_capacity(1 + ns.len() + 2);
    output.push(ns.len() as u8);
    output.extend_from_slice(ns);
    output.extend_from_slice(&slot_id.to_be_bytes());
    output
}

/// 构造 Slot Key 上界（对标 Kvrocks ComposeSlotKeyUpperBound）
#[inline]
pub fn compose_slot_key_upper_bound(ns: &[u8], slot_id: u16) -> Vec<u8> {
    compose_slot_key_prefix(ns, slot_id.saturating_add(1))
}

/// SELECT db 与命名空间转换（对标 Kvrocks SELECT db 映射机制，db 0 -> "default", db N -> "dbN"）
#[inline]
pub fn db_to_namespace(db: u32) -> Cow<'static, str> {
    match db {
        0 => Cow::Borrowed("default"),
        1 => Cow::Borrowed("db1"),
        2 => Cow::Borrowed("db2"),
        3 => Cow::Borrowed("db3"),
        4 => Cow::Borrowed("db4"),
        5 => Cow::Borrowed("db5"),
        6 => Cow::Borrowed("db6"),
        7 => Cow::Borrowed("db7"),
        8 => Cow::Borrowed("db8"),
        9 => Cow::Borrowed("db9"),
        10 => Cow::Borrowed("db10"),
        11 => Cow::Borrowed("db11"),
        12 => Cow::Borrowed("db12"),
        13 => Cow::Borrowed("db13"),
        14 => Cow::Borrowed("db14"),
        15 => Cow::Borrowed("db15"),
        _ => Cow::Owned(format!("db{db}")),
    }
}

/// SELECT db 与 Kvrocks 命名空间转换（对标 Kvrocks kDatabaseNamespacePrefix "db"）
#[inline]
pub fn db_to_namespace_kvrocks(db: u32) -> Cow<'static, str> {
    match db {
        0 => Cow::Borrowed(KVROCKS_DEFAULT_NAMESPACE),
        1 => Cow::Borrowed("db1"),
        2 => Cow::Borrowed("db2"),
        3 => Cow::Borrowed("db3"),
        4 => Cow::Borrowed("db4"),
        5 => Cow::Borrowed("db5"),
        6 => Cow::Borrowed("db6"),
        7 => Cow::Borrowed("db7"),
        8 => Cow::Borrowed("db8"),
        9 => Cow::Borrowed("db9"),
        10 => Cow::Borrowed("db10"),
        11 => Cow::Borrowed("db11"),
        12 => Cow::Borrowed("db12"),
        13 => Cow::Borrowed("db13"),
        14 => Cow::Borrowed("db14"),
        15 => Cow::Borrowed("db15"),
        _ => Cow::Owned(format!("{KVROCKS_DB_NAMESPACE_PREFIX}{db}")),
    }
}

/// 判断命名空间是否为默认命名空间（支持 "default", "__namespace", "0", "db0", "__db_0", "" 等）
#[inline]
pub fn is_default_namespace(ns: &str) -> bool {
    ns == DEFAULT_NAMESPACE
        || ns == KVROCKS_DEFAULT_NAMESPACE
        || ns == "0"
        || ns == "db0"
        || ns == "__db_0"
        || ns.is_empty()
}

/// 字节切片版本判断
#[inline]
pub fn is_default_namespace_bytes(ns: &[u8]) -> bool {
    ns == DEFAULT_NAMESPACE.as_bytes()
        || ns == KVROCKS_DEFAULT_NAMESPACE.as_bytes()
        || ns == b"0"
        || ns == b"db0"
        || ns == b"__db_0"
        || ns.is_empty()
}

/// 命名空间解析为 DB 索引
#[inline]
pub fn namespace_to_db(ns: &str) -> u32 {
    if is_default_namespace(ns) {
        0
    } else if let Some(stripped) = ns.strip_prefix("__db_").or_else(|| ns.strip_prefix("db")) {
        stripped.parse::<u32>().unwrap_or(0)
    } else {
        ns.parse::<u32>().unwrap_or_default()
    }
}

/// 校验命名空间合法性（字节切片版本，对标 Apache Kvrocks IsNamespaceLegal）
#[inline]
pub fn is_namespace_legal_bytes(ns: &[u8]) -> crate::error::Result<()> {
    if ns.is_empty() {
        return Err(crate::error::Error::invalid_data(
            "namespace cannot be empty",
        ));
    }
    if ns.len() > 255 {
        return Err(crate::error::Error::invalid_data(
            "size exceed limit 255".to_string(),
        ));
    }
    let last_byte = ns[ns.len() - 1];
    if last_byte == 0x7f || last_byte == 0xff {
        return Err(crate::error::Error::invalid_data(
            "namespace contain illegal letter",
        ));
    }
    Ok(())
}

/// 校验命名空间合法性（字符串版本，对标 Apache Kvrocks IsNamespaceLegal）
#[inline]
pub fn is_namespace_legal(ns: &str) -> crate::error::Result<()> {
    is_namespace_legal_bytes(ns.as_bytes())
}

/// Kvrocks 1:1 内部子键结构（对标 Apache Kvrocks InternalKey）
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InternalKey<'a> {
    pub namespace: &'a [u8],
    pub slot_id: Option<u16>,
    pub key: &'a [u8],
    pub version: u64,
    pub sub_key: &'a [u8],
}

impl<'a> InternalKey<'a> {
    #[inline]
    pub fn new(
        namespace: &'a [u8],
        key: &'a [u8],
        sub_key: &'a [u8],
        version: u64,
        slot_id: Option<u16>,
    ) -> Self {
        Self {
            namespace,
            slot_id,
            key,
            version,
            sub_key,
        }
    }

    /// 从已由 ComposeNamespaceKey 编码的前缀构造 InternalKey（对标 Kvrocks InternalKey 构造函数）
    #[inline]
    pub fn from_ns_key(
        mut ns_key: &'a [u8],
        sub_key: &'a [u8],
        version: u64,
        slot_id_encoded: bool,
    ) -> Option<Self> {
        if ns_key.is_empty() {
            return None;
        }
        let ns_len = ns_key[0] as usize;
        ns_key = &ns_key[1..];
        if ns_key.len() < ns_len {
            return None;
        }
        let namespace = &ns_key[..ns_len];
        ns_key = &ns_key[ns_len..];

        let slot_id = if slot_id_encoded {
            if ns_key.len() < 2 {
                return None;
            }
            let mut s_buf = [0u8; 2];
            s_buf.copy_from_slice(&ns_key[..2]);
            ns_key = &ns_key[2..];
            Some(u16::from_be_bytes(s_buf))
        } else {
            None
        };

        let key = ns_key;
        Some(Self {
            namespace,
            slot_id,
            key,
            version,
            sub_key,
        })
    }

    /// 编码为 Kvrocks 1:1 二进制存储布局：
    /// [1B ns_len][ns][(optional 2B slot_id)][4B key_len(BE)][key][8B version(BE)][sub_key]
    #[inline]
    pub fn encode(&self) -> Vec<u8> {
        let slot_len = if self.slot_id.is_some() { 2 } else { 0 };
        let total =
            1 + self.namespace.len() + slot_len + 4 + self.key.len() + 8 + self.sub_key.len();
        let mut out = Vec::with_capacity(total);
        self.encode_to(&mut out);
        out
    }

    /// 编码到已有的缓冲区中（复用堆分配）
    #[inline]
    pub fn encode_to(&self, out: &mut Vec<u8>) {
        let slot_len = if self.slot_id.is_some() { 2 } else { 0 };
        let total =
            1 + self.namespace.len() + slot_len + 4 + self.key.len() + 8 + self.sub_key.len();
        out.reserve(total);
        out.push(self.namespace.len() as u8);
        out.extend_from_slice(self.namespace);
        if let Some(slot) = self.slot_id {
            out.extend_from_slice(&slot.to_be_bytes());
        }
        out.extend_from_slice(&(self.key.len() as u32).to_be_bytes());
        out.extend_from_slice(self.key);
        out.extend_from_slice(&self.version.to_be_bytes());
        out.extend_from_slice(self.sub_key);
    }

    /// 解码 Kvrocks 二进制内部键（零拷贝）
    #[inline]
    pub fn decode(bytes: &'a [u8], slot_id_encoded: bool) -> Option<Self> {
        if bytes.is_empty() {
            return None;
        }
        let ns_len = bytes[0] as usize;
        let mut offset = 1;
        if bytes.len() < offset + ns_len {
            return None;
        }
        let namespace = &bytes[offset..offset + ns_len];
        offset += ns_len;

        let slot_id = if slot_id_encoded {
            if bytes.len() < offset + 2 {
                return None;
            }
            let mut s_buf = [0u8; 2];
            s_buf.copy_from_slice(&bytes[offset..offset + 2]);
            offset += 2;
            Some(u16::from_be_bytes(s_buf))
        } else {
            None
        };

        if bytes.len() < offset + 4 {
            return None;
        }
        let mut klen_buf = [0u8; 4];
        klen_buf.copy_from_slice(&bytes[offset..offset + 4]);
        let key_len = u32::from_be_bytes(klen_buf) as usize;
        offset += 4;

        if bytes.len() < offset + key_len + 8 {
            return None;
        }
        let key = &bytes[offset..offset + key_len];
        offset += key_len;

        let mut ver_buf = [0u8; 8];
        ver_buf.copy_from_slice(&bytes[offset..offset + 8]);
        let version = u64::from_be_bytes(ver_buf);
        offset += 8;

        let sub_key = &bytes[offset..];

        Some(Self {
            namespace,
            slot_id,
            key,
            version,
            sub_key,
        })
    }
}

/// 构造命名空间键（对标 Kvrocks ComposeNamespaceKey）
#[inline]
pub fn compose_namespace_key(ns: &[u8], key: &[u8], slot_id: Option<u16>) -> Vec<u8> {
    let slot_len = if slot_id.is_some() { 2 } else { 0 };
    let mut out = Vec::with_capacity(1 + ns.len() + slot_len + key.len());
    out.push(ns.len() as u8);
    out.extend_from_slice(ns);
    if let Some(slot) = slot_id {
        out.extend_from_slice(&slot.to_be_bytes());
    }
    out.extend_from_slice(key);
    out
}

/// 构造命名空间键并自动根据 slot_id_encoded 计算 Slot ID（对标 Apache Kvrocks ComposeNamespaceKey）
#[inline]
pub fn compose_namespace_key_with_slot(ns: &[u8], key: &[u8], slot_id_encoded: bool) -> Vec<u8> {
    let slot_id = if slot_id_encoded {
        Some(get_slot_id_from_key(key))
    } else {
        None
    };
    compose_namespace_key(ns, key, slot_id)
}

/// 从命名空间键中零拷贝提取命名空间、槽位与用户键（对标 Kvrocks ExtractNamespaceKey）
#[inline]
pub fn extract_namespace_key(
    ns_key: &[u8],
    slot_id_encoded: bool,
) -> Option<(&[u8], Option<u16>, &[u8])> {
    if ns_key.is_empty() {
        return None;
    }
    let ns_len = ns_key[0] as usize;
    let mut offset = 1;
    if ns_key.len() < offset + ns_len {
        return None;
    }
    let ns = &ns_key[offset..offset + ns_len];
    offset += ns_len;

    let slot_id = if slot_id_encoded {
        if ns_key.len() < offset + 2 {
            return None;
        }
        let mut s_buf = [0u8; 2];
        s_buf.copy_from_slice(&ns_key[offset..offset + 2]);
        offset += 2;
        Some(u16::from_be_bytes(s_buf))
    } else {
        None
    };

    let key = &ns_key[offset..];
    Some((ns, slot_id, key))
}

/// 提取 Slot ID（对标 Kvrocks ExtractSlotId）
#[inline]
pub fn extract_slot_id(ns_key: &[u8]) -> Option<u16> {
    if ns_key.is_empty() {
        return None;
    }
    let ns_len = ns_key[0] as usize;
    let offset = 1 + ns_len;
    if ns_key.len() < offset + 2 {
        return None;
    }
    let mut s_buf = [0u8; 2];
    s_buf.copy_from_slice(&ns_key[offset..offset + 2]);
    Some(u16::from_be_bytes(s_buf))
}

/// 命名空间与多租户键构造器（对标 Apache Kvrocks 与 Redis 规范）
#[derive(Debug, Clone)]
pub struct KeyComposer<'a> {
    ns: &'a str,
}

impl<'a> KeyComposer<'a> {
    #[inline]
    pub fn new(ns: &'a str) -> Self {
        Self { ns }
    }

    #[inline]
    pub fn is_default(&self) -> bool {
        is_default_namespace(self.ns)
    }

    #[inline]
    pub fn ns(&self) -> &'a str {
        self.ns
    }

    /// 获取当前用户 Key 在所有数据类型下的元数据键列表
    #[inline]
    pub fn all_meta_keys(&self, key: &str) -> [String; 13] {
        [
            self.hash_meta(key),
            self.list_meta(key),
            self.set_meta(key),
            self.zset_meta(key),
            self.bf_meta(key),
            self.cf_meta(key),
            self.bm_meta(key),
            self.hll_meta(key),
            self.json_meta(key),
            self.si_meta(key),
            self.stream_meta(key),
            self.tdigest_meta(key),
            self.ts_meta(key),
        ]
    }

    /// 获取当前命名空间在所有复杂数据类型下的元数据前缀列表
    #[inline]
    pub fn all_meta_prefixes(&self) -> Vec<Vec<u8>> {
        if self.is_default() {
            vec![
                b"\x00hm:".to_vec(),
                b"\x00lm:".to_vec(),
                b"\x00sm:".to_vec(),
                b"\x00zm:".to_vec(),
                b"\x00bfm:".to_vec(),
                b"\x00cfm:".to_vec(),
                b"\x00bmm:".to_vec(),
                b"\x00hllm:".to_vec(),
                b"\x00jm:".to_vec(),
                b"\x00sim:".to_vec(),
                b"\x00strm:".to_vec(),
                b"\x00tdm:".to_vec(),
                b"\x00tsm:".to_vec(),
            ]
        } else {
            let ns = self.ns;
            vec![
                format!("\x00ns:{ns}:hm:").into_bytes(),
                format!("\x00ns:{ns}:lm:").into_bytes(),
                format!("\x00ns:{ns}:sm:").into_bytes(),
                format!("\x00ns:{ns}:zm:").into_bytes(),
                format!("\x00ns:{ns}:bfm:").into_bytes(),
                format!("\x00ns:{ns}:cfm:").into_bytes(),
                format!("\x00ns:{ns}:bmm:").into_bytes(),
                format!("\x00ns:{ns}:hllm:").into_bytes(),
                format!("\x00ns:{ns}:jm:").into_bytes(),
                format!("\x00ns:{ns}:sim:").into_bytes(),
                format!("\x00ns:{ns}:strm:").into_bytes(),
                format!("\x00ns:{ns}:tdm:").into_bytes(),
                format!("\x00ns:{ns}:tsm:").into_bytes(),
            ]
        }
    }

    /// 获取当前命名空间下该 key 在所有子键集合中的数据前缀列表
    #[inline]
    pub fn all_data_prefixes(&self, key: &str) -> Vec<Vec<u8>> {
        vec![
            self.hash_prefix(key),
            self.list_prefix(key),
            self.set_prefix(key),
            self.zset_prefix(key),
            self.zset_score_prefix(key),
            self.bm_prefix(key),
            self.bf_prefix(key),
            self.cf_prefix(key),
            self.si_prefix(key),
            self.stream_prefix(key),
            self.stream_group_prefix(key),
            self.stream_consumer_prefix_all(key),
            self.stream_pel_prefix_all(key),
            self.ts_prefix(key),
        ]
    }

    #[inline]
    pub fn raw_key(&self, key: &str) -> String {
        if self.is_default() {
            key.to_string()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:{key}")
        }
    }

    #[inline]
    pub fn raw_key_bytes<'b>(&self, key: &'b [u8]) -> Cow<'b, [u8]> {
        if self.is_default() {
            Cow::Borrowed(key)
        } else {
            let ns_bytes = self.ns.as_bytes();
            let mut out = Vec::with_capacity(4 + ns_bytes.len() + 1 + key.len());
            out.extend_from_slice(b"\x00ns:");
            out.extend_from_slice(ns_bytes);
            out.push(b':');
            out.extend_from_slice(key);
            Cow::Owned(out)
        }
    }

    #[inline]
    pub fn hash_key(&self, key: &str, field: &str) -> String {
        if self.is_default() {
            format!("\x00h:{key}:{field}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:h:{key}:{field}")
        }
    }

    #[inline]
    pub fn hash_key_bytes(&self, key: &str, field: &[u8]) -> Vec<u8> {
        let mut prefix = self.hash_prefix(key);
        prefix.extend_from_slice(field);
        prefix
    }

    #[inline]
    pub fn hash_item(&self, key: &str, field: &str) -> String {
        self.hash_key(key, field)
    }

    #[inline]
    pub fn hash_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00hm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:hm:{key}")
        }
    }

    #[inline]
    pub fn hash_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00h:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:h:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn list_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00lm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:lm:{key}")
        }
    }

    #[inline]
    pub fn list_item(&self, key: &str, idx: u64) -> String {
        if self.is_default() {
            format!("\x00l:{key}:{idx:016x}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:l:{key}:{idx:016x}")
        }
    }

    #[inline]
    pub fn list_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00l:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:l:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn set_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00sm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:sm:{key}")
        }
    }

    #[inline]
    pub fn set_key(&self, key: &str, member: &str) -> String {
        if self.is_default() {
            format!("\x00s:{key}:{member}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:s:{key}:{member}")
        }
    }

    #[inline]
    pub fn set_item(&self, key: &str, member: &str) -> String {
        self.set_key(key, member)
    }

    #[inline]
    pub fn set_item_bytes(&self, key: &str, member: &[u8]) -> String {
        if self.is_default() {
            format!("\x00s:{key}:{}", String::from_utf8_lossy(member))
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:s:{key}:{}", String::from_utf8_lossy(member))
        }
    }

    #[inline]
    pub fn set_key_bytes(&self, key: &str, member: &[u8]) -> Vec<u8> {
        let mut prefix = self.set_prefix(key);
        prefix.extend_from_slice(member);
        prefix
    }

    #[inline]
    pub fn set_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00s:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:s:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn zset_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00zm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:zm:{key}")
        }
    }

    #[inline]
    pub fn zset_key(&self, key: &str, member: &str) -> String {
        if self.is_default() {
            format!("\x00z:{key}:{member}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:z:{key}:{member}")
        }
    }

    #[inline]
    pub fn zset_key_bytes(&self, key: &str, member: &[u8]) -> Vec<u8> {
        let mut out = self.zset_prefix(key);
        out.extend_from_slice(member);
        out
    }

    #[inline]
    pub fn zset_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00z:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:z:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn zset_score_key(&self, key: &str, score: f64, member: &str) -> String {
        let enc = encode_sortable_f64(score);
        let hex_bytes = bytes_to_hex_16(enc);
        let hex = str::from_utf8(&hex_bytes).unwrap_or("");
        if self.is_default() {
            format!("\x00zs:{key}:{hex}:{member}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:zs:{key}:{hex}:{member}")
        }
    }

    #[inline]
    pub fn zset_score_key_bytes(&self, key: &str, score: f64, member: &[u8]) -> Vec<u8> {
        let enc = encode_sortable_f64(score);
        let hex_bytes = bytes_to_hex_16(enc);
        let mut out = self.zset_score_prefix(key);
        out.extend_from_slice(&hex_bytes);
        out.push(b':');
        out.extend_from_slice(member);
        out
    }

    #[inline]
    pub fn zset_score_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00zs:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:zs:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn hll_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00hllm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:hllm:{key}")
        }
    }

    #[inline]
    pub fn hll_meta_prefix(&self) -> Vec<u8> {
        if self.is_default() {
            b"\x00hllm:".to_vec()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:hllm:").into_bytes()
        }
    }

    #[inline]
    pub fn hll_key(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00hll:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:hll:{key}")
        }
    }

    #[inline]
    pub fn bf_key(&self, key: &str) -> String {
        self.bf_meta(key)
    }

    #[inline]
    pub fn bf_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00bfm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:bfm:{key}")
        }
    }

    #[inline]
    pub fn bf_item(&self, key: &str, filter_idx: u16) -> String {
        if self.is_default() {
            format!("\x00bf:{key}:{filter_idx:04x}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:bf:{key}:{filter_idx:04x}")
        }
    }

    #[inline]
    pub fn bf_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00bf:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:bf:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn cf_key(&self, key: &str) -> String {
        self.cf_meta(key)
    }

    #[inline]
    pub fn cf_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00cfm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:cfm:{key}")
        }
    }

    #[inline]
    pub fn cf_page(&self, key: &str, filter_idx: u16, page_idx: u32) -> String {
        if self.is_default() {
            format!("\x00cf:{key}:{filter_idx:04x}:{page_idx:08x}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:cf:{key}:{filter_idx:04x}:{page_idx:08x}")
        }
    }

    #[inline]
    pub fn cf_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00cf:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:cf:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn si_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00sim:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:sim:{key}")
        }
    }

    #[inline]
    pub fn si_key(&self, key: &str, id: u64) -> String {
        if self.is_default() {
            format!("\x00si:{key}:{id:016x}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:si:{key}:{id:016x}")
        }
    }

    #[inline]
    pub fn si_item(&self, key: &str, id: u64) -> String {
        self.si_key(key, id)
    }

    #[inline]
    pub fn si_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00si:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:si:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn bm_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00bmm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:bmm:{key}")
        }
    }

    #[inline]
    pub fn bm_segment(&self, key: &str, seg_idx: u32) -> String {
        if self.is_default() {
            format!("\x00bm:{key}:{seg_idx:08x}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:bm:{key}:{seg_idx:08x}")
        }
    }

    #[inline]
    pub fn bm_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00bm:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:bm:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn stream_key(&self, key: &str) -> String {
        self.stream_meta(key)
    }

    #[inline]
    pub fn stream_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00strm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:strm:{key}")
        }
    }

    #[inline]
    pub fn stream_item(&self, key: &str, ms: u64, seq: u64) -> String {
        if self.is_default() {
            format!("\x00str:{key}:{ms:016x}:{seq:016x}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str:{key}:{ms:016x}:{seq:016x}")
        }
    }

    #[inline]
    pub fn stream_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00str:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn stream_group_meta(&self, key: &str, group: &str) -> String {
        if self.is_default() {
            format!("\x00str_g:{key}:{group}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str_g:{key}:{group}")
        }
    }

    #[inline]
    pub fn stream_group_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00str_g:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str_g:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn stream_consumer_meta(&self, key: &str, group: &str, consumer: &str) -> String {
        if self.is_default() {
            format!("\x00str_c:{key}:{group}:{consumer}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str_c:{key}:{group}:{consumer}")
        }
    }

    #[inline]
    pub fn stream_consumer_prefix(&self, key: &str, group: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00str_c:{key}:{group}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str_c:{key}:{group}:").into_bytes()
        }
    }

    #[inline]
    pub fn stream_consumer_prefix_all(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00str_c:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str_c:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn stream_pel_item(&self, key: &str, group: &str, ms: u64, seq: u64) -> String {
        if self.is_default() {
            format!("\x00str_p:{key}:{group}:{ms:016x}:{seq:016x}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str_p:{key}:{group}:{ms:016x}:{seq:016x}")
        }
    }

    #[inline]
    pub fn stream_pel_prefix(&self, key: &str, group: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00str_p:{key}:{group}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str_p:{key}:{group}:").into_bytes()
        }
    }

    #[inline]
    pub fn stream_pel_prefix_all(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00str_p:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:str_p:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn json_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00jm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:jm:{key}")
        }
    }

    #[inline]
    pub fn json_meta_prefix(&self) -> Vec<u8> {
        if self.is_default() {
            b"\x00jm:".to_vec()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:jm:").into_bytes()
        }
    }

    #[inline]
    pub fn json_key(&self, key: &str) -> String {
        self.json_meta(key)
    }

    #[inline]
    pub fn ft_schema(&self, index: &str) -> String {
        if self.is_default() {
            format!("\x00fts:{index}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:fts:{index}")
        }
    }

    #[inline]
    pub fn ft_schema_prefix(&self) -> Vec<u8> {
        if self.is_default() {
            b"\x00fts:".to_vec()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:fts:").into_bytes()
        }
    }

    #[inline]
    pub fn ft_alias(&self, alias: &str) -> String {
        if self.is_default() {
            format!("\x00fta:{alias}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:fta:{alias}")
        }
    }

    #[inline]
    pub fn ft_alias_prefix(&self) -> Vec<u8> {
        if self.is_default() {
            b"\x00fta:".to_vec()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:fta:").into_bytes()
        }
    }

    #[inline]
    pub fn ft_index_key(&self, index: &str, field: &str, term: &str, doc_id: &str) -> String {
        if self.is_default() {
            format!("\x00ft_idx:{index}:{field}:{term}:{doc_id}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:ft_idx:{index}:{field}:{term}:{doc_id}")
        }
    }

    #[inline]
    pub fn ft_index_prefix(&self, index: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00ft_idx:{index}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:ft_idx:{index}:").into_bytes()
        }
    }

    #[inline]
    pub fn ft_index_field_prefix(&self, index: &str, field: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00ft_idx:{index}:{field}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:ft_idx:{index}:{field}:").into_bytes()
        }
    }

    #[inline]
    pub fn ft_index_term_prefix(&self, index: &str, field: &str, term: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00ft_idx:{index}:{field}:{term}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:ft_idx:{index}:{field}:{term}:").into_bytes()
        }
    }

    #[inline]
    pub fn ft_index_term_scan_prefix(
        &self,
        index: &str,
        field: &str,
        term_prefix: &str,
    ) -> Vec<u8> {
        if self.is_default() {
            format!("\x00ft_idx:{index}:{field}:{term_prefix}").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:ft_idx:{index}:{field}:{term_prefix}").into_bytes()
        }
    }

    #[inline]
    pub fn tdigest_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00tdm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:tdm:{key}")
        }
    }

    #[inline]
    pub fn tdigest_meta_prefix(&self) -> Vec<u8> {
        if self.is_default() {
            b"\x00tdm:".to_vec()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:tdm:").into_bytes()
        }
    }

    #[inline]
    pub fn tdigest_key(&self, key: &str) -> String {
        self.tdigest_meta(key)
    }

    #[inline]
    pub fn ts_meta(&self, key: &str) -> String {
        if self.is_default() {
            format!("\x00tsm:{key}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:tsm:{key}")
        }
    }

    #[inline]
    pub fn ts_meta_prefix(&self) -> Vec<u8> {
        if self.is_default() {
            b"\x00tsm:".to_vec()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:tsm:").into_bytes()
        }
    }

    #[inline]
    pub fn ts_item(&self, key: &str, timestamp: u64) -> String {
        if self.is_default() {
            format!("\x00ts:{key}:{timestamp:016x}")
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:ts:{key}:{timestamp:016x}")
        }
    }

    #[inline]
    pub fn ts_prefix(&self, key: &str) -> Vec<u8> {
        if self.is_default() {
            format!("\x00ts:{key}:").into_bytes()
        } else {
            let ns = self.ns;
            format!("\x00ns:{ns}:ts:{key}:").into_bytes()
        }
    }

    #[inline]
    pub fn ts_key(&self, key: &str) -> String {
        self.ts_meta(key)
    }

    #[inline]
    pub fn ft_key(&self, key: &str) -> String {
        self.ft_schema(key)
    }

    #[inline]
    pub fn is_key_in_ns(&self, full_key: &[u8]) -> bool {
        if full_key.starts_with(b"_meta:")
            || full_key.starts_with(b"_ttl:")
            || full_key.starts_with(b"_raft:")
        {
            return false;
        }
        if self.is_default() {
            !full_key.starts_with(b"\x00ns:")
        } else {
            let ns_bytes = self.ns.as_bytes();
            let prefix_len = 4 + ns_bytes.len() + 1;
            full_key.len() >= prefix_len
                && full_key.starts_with(b"\x00ns:")
                && &full_key[4..4 + ns_bytes.len()] == ns_bytes
                && full_key[4 + ns_bytes.len()] == b':'
        }
    }

    /// 高性能零分配提取用户 Key（基于 memchr / memrchr 快速前缀与边界扫描）
    #[inline]
    pub fn extract_user_key<'b>(&self, full_key: &'b [u8]) -> Option<&'b [u8]> {
        if full_key.starts_with(b"_meta:")
            || full_key.starts_with(b"_ttl:")
            || full_key.starts_with(b"_raft:")
        {
            return None;
        }

        if self.is_default() {
            if full_key.starts_with(b"\x00ns:") {
                return None;
            }
            if let Some(remain) = full_key.strip_prefix(b"\x00") {
                parse_tag_and_key(remain)
            } else {
                Some(full_key)
            }
        } else {
            let ns_bytes = self.ns.as_bytes();
            let prefix_len = 4 + ns_bytes.len() + 1;
            if full_key.len() >= prefix_len
                && full_key.starts_with(b"\x00ns:")
                && &full_key[4..4 + ns_bytes.len()] == ns_bytes
                && full_key[4 + ns_bytes.len()] == b':'
            {
                let remain = &full_key[prefix_len..];
                parse_tag_and_key(remain)
            } else {
                None
            }
        }
    }

    /// 将当前命名空间的底层键转换为目标命名空间的底层键（零拷贝/二进制安全）
    #[inline]
    pub fn transform_key_to_target_bytes(
        &self,
        full_key: &[u8],
        target_kc: &KeyComposer<'_>,
    ) -> Option<Vec<u8>> {
        if !self.is_key_in_ns(full_key) {
            return None;
        }
        if self.is_default() {
            if target_kc.is_default() {
                return Some(full_key.to_vec());
            }
            let target_ns = target_kc.ns().as_bytes();
            if let Some(remain) = full_key.strip_prefix(b"\x00") {
                let mut out = Vec::with_capacity(4 + target_ns.len() + 1 + remain.len());
                out.extend_from_slice(b"\x00ns:");
                out.extend_from_slice(target_ns);
                out.push(b':');
                out.extend_from_slice(remain);
                Some(out)
            } else {
                let mut out = Vec::with_capacity(4 + target_ns.len() + 1 + full_key.len());
                out.extend_from_slice(b"\x00ns:");
                out.extend_from_slice(target_ns);
                out.push(b':');
                out.extend_from_slice(full_key);
                Some(out)
            }
        } else {
            let ns_prefix_len = 4 + self.ns.len() + 1;
            let remain = &full_key[ns_prefix_len..];
            if target_kc.is_default() {
                if is_internal_data_tag(remain) {
                    let mut out = Vec::with_capacity(1 + remain.len());
                    out.push(0);
                    out.extend_from_slice(remain);
                    Some(out)
                } else {
                    Some(remain.to_vec())
                }
            } else {
                let target_ns = target_kc.ns().as_bytes();
                let mut out = Vec::with_capacity(4 + target_ns.len() + 1 + remain.len());
                out.extend_from_slice(b"\x00ns:");
                out.extend_from_slice(target_ns);
                out.push(b':');
                out.extend_from_slice(remain);
                Some(out)
            }
        }
    }

    /// 将当前命名空间的底层键转换为目标命名空间的底层键（字符串包装）
    #[inline]
    pub fn transform_key_to_target(
        &self,
        full_key: &[u8],
        target_kc: &KeyComposer<'_>,
    ) -> Option<String> {
        self.transform_key_to_target_bytes(full_key, target_kc)
            .map(|bytes| {
                String::from_utf8(bytes)
                    .unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).into_owned())
            })
    }
}

#[inline]
fn is_internal_data_tag(remain: &[u8]) -> bool {
    if remain.is_empty() {
        return false;
    }
    match remain[0] {
        b'l' => remain.starts_with(b"lm:") || remain.starts_with(b"l:"),
        b'h' => {
            remain.starts_with(b"hm:")
                || remain.starts_with(b"hllm:")
                || remain.starts_with(b"hll:")
                || remain.starts_with(b"h:")
        }
        b's' => {
            remain.starts_with(b"sm:")
                || remain.starts_with(b"sim:")
                || remain.starts_with(b"si:")
                || remain.starts_with(b"strm:")
                || remain.starts_with(b"str_g:")
                || remain.starts_with(b"str_c:")
                || remain.starts_with(b"str_p:")
                || remain.starts_with(b"str:")
                || remain.starts_with(b"s:")
        }
        b'z' => {
            remain.starts_with(b"zm:") || remain.starts_with(b"zs:") || remain.starts_with(b"z:")
        }
        b'b' => {
            remain.starts_with(b"bmm:")
                || remain.starts_with(b"bm:")
                || remain.starts_with(b"bfm:")
                || remain.starts_with(b"bf:")
        }
        b'c' => remain.starts_with(b"cfm:") || remain.starts_with(b"cf:"),
        b'j' => remain.starts_with(b"jm:") || remain.starts_with(b"j:"),
        b't' => {
            remain.starts_with(b"tsm:")
                || remain.starts_with(b"tdm:")
                || remain.starts_with(b"ts:")
                || remain.starts_with(b"td:")
        }
        b'f' => {
            remain.starts_with(b"ft:")
                || remain.starts_with(b"fts:")
                || remain.starts_with(b"fta:")
                || remain.starts_with(b"ft_idx:")
        }
        _ => false,
    }
}

#[inline]
fn parse_tag_and_key(remain: &[u8]) -> Option<&[u8]> {
    if remain.is_empty() {
        return Some(remain);
    }
    match remain[0] {
        b'l' => {
            if let Some(rest) = remain.strip_prefix(b"lm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"l:") {
                if let Some(last_colon) = memrchr(b':', rest) {
                    Some(&rest[..last_colon])
                } else {
                    Some(rest)
                }
            } else {
                Some(remain)
            }
        }
        b'h' => {
            if let Some(rest) = remain.strip_prefix(b"hm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"hllm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"hll:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"h:") {
                if let Some(last_colon) = memrchr(b':', rest) {
                    Some(&rest[..last_colon])
                } else {
                    Some(rest)
                }
            } else {
                Some(remain)
            }
        }
        b's' => {
            if let Some(rest) = remain.strip_prefix(b"sm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"sim:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"strm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"str_g:") {
                if let Some(last_colon) = memrchr(b':', rest) {
                    Some(&rest[..last_colon])
                } else {
                    Some(rest)
                }
            } else if let Some(rest) = remain.strip_prefix(b"str_c:") {
                if let Some(last_colon) = memrchr(b':', rest) {
                    let before = &rest[..last_colon];
                    if let Some(second_last) = memrchr(b':', before) {
                        Some(&before[..second_last])
                    } else {
                        Some(before)
                    }
                } else {
                    Some(rest)
                }
            } else if let Some(rest) = remain.strip_prefix(b"str_p:") {
                if rest.len() > 34 && rest[rest.len() - 34] == b':' && rest[rest.len() - 17] == b':'
                {
                    let before = &rest[..rest.len() - 34];
                    if let Some(last_colon) = memrchr(b':', before) {
                        Some(&before[..last_colon])
                    } else {
                        Some(before)
                    }
                } else {
                    Some(rest)
                }
            } else if let Some(rest) = remain.strip_prefix(b"str:") {
                if rest.len() > 34 && rest[rest.len() - 34] == b':' && rest[rest.len() - 17] == b':'
                {
                    Some(&rest[..rest.len() - 34])
                } else {
                    Some(rest)
                }
            } else if let Some(rest) = remain.strip_prefix(b"si:") {
                if let Some(last_colon) = memrchr(b':', rest) {
                    Some(&rest[..last_colon])
                } else {
                    Some(rest)
                }
            } else if let Some(rest) = remain.strip_prefix(b"s:") {
                if let Some(last_colon) = memrchr(b':', rest) {
                    Some(&rest[..last_colon])
                } else {
                    Some(rest)
                }
            } else {
                Some(remain)
            }
        }
        b'z' => {
            if let Some(rest) = remain.strip_prefix(b"zm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"zs:") {
                let mut idx = 0;
                while let Some(colon_pos) = memchr(b':', &rest[idx..]) {
                    let abs_pos = idx + colon_pos;
                    if abs_pos + 17 < rest.len() && rest[abs_pos + 17] == b':' {
                        let hex_slice = &rest[abs_pos + 1..abs_pos + 17];
                        if hex_slice.iter().all(|b| b.is_ascii_hexdigit()) {
                            return Some(&rest[..abs_pos]);
                        }
                    }
                    idx = abs_pos + 1;
                }
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"z:") {
                if let Some(last_colon) = memrchr(b':', rest) {
                    Some(&rest[..last_colon])
                } else {
                    Some(rest)
                }
            } else {
                Some(remain)
            }
        }
        b'b' => {
            if let Some(rest) = remain.strip_prefix(b"bmm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"bfm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"bf:") {
                if rest.len() > 5 && rest[rest.len() - 5] == b':' {
                    Some(&rest[..rest.len() - 5])
                } else {
                    Some(rest)
                }
            } else if let Some(rest) = remain.strip_prefix(b"bm:") {
                if let Some(last_colon) = memrchr(b':', rest) {
                    Some(&rest[..last_colon])
                } else {
                    Some(rest)
                }
            } else {
                Some(remain)
            }
        }
        b'c' => {
            if let Some(rest) = remain.strip_prefix(b"cfm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"cf:") {
                if rest.len() > 14 && rest[rest.len() - 14] == b':' && rest[rest.len() - 9] == b':'
                {
                    Some(&rest[..rest.len() - 14])
                } else {
                    Some(rest)
                }
            } else {
                Some(remain)
            }
        }
        b'j' => {
            if let Some(rest) = remain.strip_prefix(b"jm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"j:") {
                Some(rest)
            } else {
                Some(remain)
            }
        }
        b't' => {
            if let Some(rest) = remain.strip_prefix(b"tsm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"tdm:") {
                Some(rest)
            } else if let Some(rest) = remain.strip_prefix(b"ts:") {
                if rest.len() > 17 && rest[rest.len() - 17] == b':' {
                    Some(&rest[..rest.len() - 17])
                } else {
                    Some(rest)
                }
            } else if let Some(rest) = remain.strip_prefix(b"td:") {
                Some(rest)
            } else {
                Some(remain)
            }
        }
        b'f' => {
            if remain.starts_with(b"fts:")
                || remain.starts_with(b"ft:")
                || remain.starts_with(b"fta:")
                || remain.starts_with(b"ft_idx:")
            {
                None
            } else {
                Some(remain)
            }
        }
        _ => Some(remain),
    }
}

/// 匹配通配符 Pattern (支持 `*`, `?`, `[a-z]`, `[^a-z]`, `\x` 转义，字节切片零拷贝)
#[inline]
pub fn matches_glob_bytes(mut pattern: &[u8], mut text: &[u8]) -> bool {
    while !pattern.is_empty() {
        match pattern[0] {
            b'*' => {
                while pattern.len() > 1 && pattern[1] == b'*' {
                    pattern = &pattern[1..];
                }
                if pattern.len() == 1 {
                    return true;
                }
                while !text.is_empty() {
                    if matches_glob_bytes(&pattern[1..], text) {
                        return true;
                    }
                    text = &text[1..];
                }
                return matches_glob_bytes(&pattern[1..], text);
            }
            b'?' => {
                if text.is_empty() {
                    return false;
                }
                pattern = &pattern[1..];
                text = &text[1..];
            }
            b'[' => {
                if text.is_empty() {
                    return false;
                }
                pattern = &pattern[1..];
                let not = if !pattern.is_empty() && (pattern[0] == b'^' || pattern[0] == b'!') {
                    pattern = &pattern[1..];
                    true
                } else {
                    false
                };
                let mut match_found = false;
                let c = text[0];
                loop {
                    if pattern.is_empty() {
                        return false;
                    }
                    if pattern[0] == b'\\' && pattern.len() > 1 {
                        pattern = &pattern[1..];
                        if pattern[0] == c {
                            match_found = true;
                        }
                    } else if pattern[0] == b']' {
                        pattern = &pattern[1..];
                        break;
                    } else if pattern.len() >= 3 && pattern[1] == b'-' && pattern[2] != b']' {
                        let start = pattern[0];
                        let end = pattern[2];
                        if (start <= c && c <= end) || (end <= c && c <= start) {
                            match_found = true;
                        }
                        pattern = &pattern[2..];
                    } else if pattern[0] == c {
                        match_found = true;
                    }
                    pattern = &pattern[1..];
                }
                if not {
                    match_found = !match_found;
                }
                if !match_found {
                    return false;
                }
                text = &text[1..];
            }
            b'\\' => {
                if pattern.len() > 1 {
                    pattern = &pattern[1..];
                }
                if text.is_empty() || pattern[0] != text[0] {
                    return false;
                }
                pattern = &pattern[1..];
                text = &text[1..];
            }
            c => {
                if text.is_empty() || text[0] != c {
                    return false;
                }
                pattern = &pattern[1..];
                text = &text[1..];
            }
        }
    }
    text.is_empty()
}

/// 匹配通配符 Pattern (字符串包装)
#[inline]
pub fn matches_glob(pattern: &str, text: &str) -> bool {
    matches_glob_bytes(pattern.as_bytes(), text.as_bytes())
}

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

    #[test]
    fn test_internal_key_roundtrip() {
        let key = InternalKey::new(
            b"test_ns",
            b"my_key",
            b"my_field",
            0x1234567890abcdef,
            Some(100),
        );
        let enc = key.encode();
        let decoded = InternalKey::decode(&enc, true).expect("decode failed");

        assert_eq!(decoded.namespace, b"test_ns");
        assert_eq!(decoded.slot_id, Some(100));
        assert_eq!(decoded.key, b"my_key");
        assert_eq!(decoded.version, 0x1234567890abcdef);
        assert_eq!(decoded.sub_key, b"my_field");
    }

    #[test]
    fn test_internal_key_no_slot() {
        let key = InternalKey::new(b"", b"k", b"f", 42, None);
        let enc = key.encode();
        let decoded = InternalKey::decode(&enc, false).expect("decode failed");

        assert_eq!(decoded.namespace, b"");
        assert_eq!(decoded.slot_id, None);
        assert_eq!(decoded.key, b"k");
        assert_eq!(decoded.version, 42);
        assert_eq!(decoded.sub_key, b"f");
    }

    #[test]
    fn test_compose_and_extract_namespace_key() {
        let ns_key = compose_namespace_key(b"test_ns", b"user_key_1", Some(512));
        let (ns, slot, key) = extract_namespace_key(&ns_key, true).expect("extract failed");
        assert_eq!(ns, b"test_ns");
        assert_eq!(slot, Some(512));
        assert_eq!(key, b"user_key_1");

        let slot_id = extract_slot_id(&ns_key);
        assert_eq!(slot_id, Some(512));
    }

    #[test]
    fn test_db_namespace_conversion() {
        assert_eq!(db_to_namespace(0).as_ref(), "default");
        assert_eq!(db_to_namespace(1).as_ref(), "db1");
        assert_eq!(db_to_namespace(15).as_ref(), "db15");
        assert_eq!(db_to_namespace(16).as_ref(), "db16");
        assert_eq!(db_to_namespace(100).as_ref(), "db100");

        assert_eq!(namespace_to_db("default"), 0);
        assert_eq!(namespace_to_db("__namespace"), 0);
        assert_eq!(namespace_to_db("0"), 0);
        assert_eq!(namespace_to_db("db0"), 0);
        assert_eq!(namespace_to_db("__db_0"), 0);
        assert_eq!(namespace_to_db("db1"), 1);
        assert_eq!(namespace_to_db("__db_2"), 2);
        assert_eq!(namespace_to_db("15"), 15);
        assert_eq!(namespace_to_db("db16"), 16);
        assert_eq!(db_to_namespace_kvrocks(0), "__namespace");
        assert_eq!(db_to_namespace_kvrocks(3), "db3");
        assert_eq!(db_to_namespace_kvrocks(15), "db15");
    }

    #[test]
    fn test_is_namespace_legal_checks() {
        assert!(is_namespace_legal("tenant_1").is_ok());
        assert!(is_namespace_legal("a".repeat(255).as_str()).is_ok());
        assert!(is_namespace_legal("").is_err());
        assert!(is_namespace_legal("a".repeat(256).as_str()).is_err());
        // 非法末尾字符 (0x7f / 0xff)
        let invalid_ns = "tenant\x7f".to_string();
        assert!(is_namespace_legal(&invalid_ns).is_err());
    }

    #[test]
    fn test_key_composer_extraction_and_meta() {
        let kc = KeyComposer::new("tenant_x");
        assert!(!kc.is_default());
        assert_eq!(kc.ns(), "tenant_x");

        let all_meta = kc.all_meta_keys("user:1");
        assert_eq!(all_meta.len(), 13);
        assert_eq!(all_meta[0], "\x00ns:tenant_x:hm:user:1");
        assert_eq!(all_meta[1], "\x00ns:tenant_x:lm:user:1");
        assert_eq!(all_meta[2], "\x00ns:tenant_x:sm:user:1");
        assert_eq!(all_meta[3], "\x00ns:tenant_x:zm:user:1");

        let raw_k = kc.raw_key("user:1");
        assert!(kc.is_key_in_ns(raw_k.as_bytes()));
        assert_eq!(
            kc.extract_user_key(raw_k.as_bytes()),
            Some(b"user:1".as_slice())
        );

        let hash_meta = kc.hash_meta("user:1");
        assert_eq!(
            kc.extract_user_key(hash_meta.as_bytes()),
            Some(b"user:1".as_slice())
        );

        let hash_item = kc.hash_item("user:1", "email");
        assert_eq!(
            kc.extract_user_key(hash_item.as_bytes()),
            Some(b"user:1".as_slice())
        );
    }

    #[test]
    fn test_internal_key_from_ns_key() {
        let ns_key = compose_namespace_key(b"test_ns", b"user_key_2", Some(1234));
        let ik = InternalKey::from_ns_key(&ns_key, b"sub_field", 9999, true)
            .expect("from_ns_key failed");
        assert_eq!(ik.namespace, b"test_ns");
        assert_eq!(ik.slot_id, Some(1234));
        assert_eq!(ik.key, b"user_key_2");
        assert_eq!(ik.version, 9999);
        assert_eq!(ik.sub_key, b"sub_field");
    }

    #[test]
    fn test_slot_hashing_and_tag() {
        assert_eq!(get_tag_from_key(b"user:{1000}:profile"), b"1000");
        assert_eq!(get_tag_from_key(b"plain_key"), b"");
        assert_eq!(get_tag_from_key(b"empty{}tag"), b"");

        let slot1 = get_slot_id_from_key(b"user:{1000}:profile");
        let slot2 = get_slot_id_from_key(b"user:{1000}:orders");
        assert_eq!(slot1, slot2);
        assert!(slot1 < HASH_SLOTS_SIZE);

        let prefix = compose_slot_key_prefix(b"ns1", 100);
        let upper = compose_slot_key_upper_bound(b"ns1", 100);
        assert_eq!(prefix, compose_slot_key_prefix(b"ns1", 100));
        assert_eq!(upper, compose_slot_key_prefix(b"ns1", 101));
    }
}