wedb_embed 0.1.1

use std::borrow::Cow;

use memchr::memrchr;

use crate::key_composer::ns::is_default_namespace;
use crate::key_composer::small_key::SmallKey;
use crate::key_composer::tag::KeyTag;
use crate::meta::{
    bytes_to_hex_16, encode_sortable_f64, u8_to_hex_2, u16_to_hex_4, u32_to_hex_8, u64_to_hex_16,
};

/// 通用子键复用构建器（零堆分配，支持高频循环迭代）
#[derive(Debug, Clone)]
pub struct SubkeyComposer {
    buf: Vec<u8>,
    prefix_len: usize,
}

impl SubkeyComposer {
    #[inline]
    pub fn new(prefix: Vec<u8>) -> Self {
        let prefix_len = prefix.len();
        Self {
            buf: prefix,
            prefix_len,
        }
    }

    #[inline]
    pub fn from_slice(prefix: &[u8]) -> Self {
        let prefix_len = prefix.len();
        let mut buf = Vec::with_capacity(prefix_len + 64);
        buf.extend_from_slice(prefix);
        Self { buf, prefix_len }
    }

    #[inline]
    pub fn compose_sub(&mut self, subkey: &[u8]) -> &[u8] {
        self.buf.truncate(self.prefix_len);
        self.buf.extend_from_slice(subkey);
        &self.buf
    }

    #[inline]
    pub fn compose_sub_u64_hex(&mut self, val: u64) -> &[u8] {
        self.buf.truncate(self.prefix_len);
        self.buf.extend_from_slice(&u64_to_hex_16(val));
        &self.buf
    }

    #[inline(always)]
    pub fn compose_sub_u8_hex(&mut self, val: u8) -> &[u8] {
        self.buf.truncate(self.prefix_len);
        self.buf.extend_from_slice(&u8_to_hex_2(val));
        &self.buf
    }

    #[inline(always)]
    pub fn key_for<'a>(&'a mut self, subkey: &[u8]) -> &'a [u8] {
        self.compose_sub(subkey)
    }

    #[inline(always)]
    pub fn key_for_hex_u64(&mut self, val: u64) -> &[u8] {
        self.compose_sub_u64_hex(val)
    }

    #[inline]
    pub fn compose_sub_u32_hex(&mut self, val: u32) -> &[u8] {
        self.buf.truncate(self.prefix_len);
        self.buf.extend_from_slice(&u32_to_hex_8(val));
        &self.buf
    }

    #[inline]
    pub fn compose_sub_u16_hex(&mut self, val: u16) -> &[u8] {
        self.buf.truncate(self.prefix_len);
        self.buf.extend_from_slice(&u16_to_hex_4(val));
        &self.buf
    }

    #[inline]
    pub fn prefix(&self) -> &[u8] {
        &self.buf[..self.prefix_len]
    }
}

/// 统一的物理键编排器（负责多租户隔离、复合数据结构 Key 与子键前缀构造）
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct KeyComposer<'a> {
    ns: &'a str,
}

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

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

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

    #[inline]
    pub fn compose_meta_key_into(&self, tag: &[u8], key_bytes: &[u8], buf: &mut Vec<u8>) {
        buf.clear();
        if self.is_default() {
            buf.reserve(1 + tag.len() + 1 + key_bytes.len());
            buf.push(0);
            buf.extend_from_slice(tag);
            buf.push(b':');
            buf.extend_from_slice(key_bytes);
        } else {
            let ns_bytes = self.ns.as_bytes();
            buf.reserve(4 + ns_bytes.len() + 1 + tag.len() + 1 + key_bytes.len());
            buf.extend_from_slice(b"\x00ns:");
            buf.extend_from_slice(ns_bytes);
            buf.push(b':');
            buf.extend_from_slice(tag);
            buf.push(b':');
            buf.extend_from_slice(key_bytes);
        }
    }

    #[inline]
    pub fn compose_meta_key(&self, tag: &[u8], key_bytes: &[u8]) -> Vec<u8> {
        let mut buf = Vec::new();
        self.compose_meta_key_into(tag, key_bytes, &mut buf);
        buf
    }

    #[inline]
    pub fn compose_meta_key_stack(&self, tag: &[u8], key_bytes: &[u8]) -> SmallKey {
        let mut sk = SmallKey::new();
        if self.is_default() {
            sk.push(0);
            sk.extend_from_slice(tag);
            sk.push(b':');
            sk.extend_from_slice(key_bytes);
        } else {
            let ns_bytes = self.ns.as_bytes();
            sk.extend_from_slice(b"\x00ns:");
            sk.extend_from_slice(ns_bytes);
            sk.push(b':');
            sk.extend_from_slice(tag);
            sk.push(b':');
            sk.extend_from_slice(key_bytes);
        }
        sk
    }

    #[inline]
    pub fn compose_prefix_into(&self, tag: &[u8], key_bytes: &[u8], buf: &mut Vec<u8>) {
        buf.clear();
        if self.is_default() {
            buf.reserve(1 + tag.len() + 1 + key_bytes.len() + 1);
            buf.push(0);
            buf.extend_from_slice(tag);
            buf.push(b':');
            buf.extend_from_slice(key_bytes);
            buf.push(b':');
        } else {
            let ns_bytes = self.ns.as_bytes();
            buf.reserve(4 + ns_bytes.len() + 1 + tag.len() + 1 + key_bytes.len() + 1);
            buf.extend_from_slice(b"\x00ns:");
            buf.extend_from_slice(ns_bytes);
            buf.push(b':');
            buf.extend_from_slice(tag);
            buf.push(b':');
            buf.extend_from_slice(key_bytes);
            buf.push(b':');
        }
    }

    #[inline]
    pub fn compose_prefix(&self, tag: &[u8], key_bytes: &[u8]) -> Vec<u8> {
        let mut buf = Vec::new();
        self.compose_prefix_into(tag, key_bytes, &mut buf);
        buf
    }

    #[inline]
    pub fn compose_prefix_stack(&self, tag: &[u8], key_bytes: &[u8]) -> SmallKey {
        let mut sk = SmallKey::new();
        if self.is_default() {
            sk.push(0);
            sk.extend_from_slice(tag);
            sk.push(b':');
            sk.extend_from_slice(key_bytes);
            sk.push(b':');
        } else {
            let ns_bytes = self.ns.as_bytes();
            sk.extend_from_slice(b"\x00ns:");
            sk.extend_from_slice(ns_bytes);
            sk.push(b':');
            sk.extend_from_slice(tag);
            sk.push(b':');
            sk.extend_from_slice(key_bytes);
            sk.push(b':');
        }
        sk
    }

    #[inline]
    pub fn compose_subkey_into(
        &self,
        tag: &[u8],
        key_bytes: &[u8],
        subkey: &[u8],
        buf: &mut Vec<u8>,
    ) {
        self.compose_prefix_into(tag, key_bytes, buf);
        buf.extend_from_slice(subkey);
    }

    #[inline]
    fn compose_meta_str(&self, tag: &[u8], key: &str) -> String {
        let mut v = Vec::new();
        self.compose_meta_key_into(tag, key.as_bytes(), &mut v);
        unsafe { String::from_utf8_unchecked(v) }
    }

    #[inline]
    fn compose_prefix_vec(&self, tag: &[u8], key: &str) -> Vec<u8> {
        let mut v = Vec::new();
        self.compose_prefix_into(tag, key.as_bytes(), &mut v);
        v
    }

    #[inline]
    fn compose_meta_prefix_vec(&self, tag: &[u8]) -> Vec<u8> {
        if self.is_default() {
            let mut v = Vec::with_capacity(1 + tag.len() + 1);
            v.push(0);
            v.extend_from_slice(tag);
            v.push(b':');
            v
        } else {
            let ns_bytes = self.ns.as_bytes();
            let mut v = Vec::with_capacity(4 + ns_bytes.len() + 1 + tag.len() + 1);
            v.extend_from_slice(b"\x00ns:");
            v.extend_from_slice(ns_bytes);
            v.push(b':');
            v.extend_from_slice(tag);
            v.push(b':');
            v
        }
    }

    #[inline]
    fn compose_sub_key_str(&self, tag: &[u8], key: &str, subkey: &str) -> String {
        let mut v = Vec::new();
        self.compose_subkey_into(tag, key.as_bytes(), subkey.as_bytes(), &mut v);
        unsafe { String::from_utf8_unchecked(v) }
    }

    #[inline]
    fn compose_sub_key_vec(&self, tag: &[u8], key: &str, subkey: &[u8]) -> Vec<u8> {
        let mut v = Vec::new();
        self.compose_subkey_into(tag, key.as_bytes(), subkey, &mut v);
        v
    }

    #[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>> {
        use crate::key_composer::tag::ALL_COMPOSITE_META_TAGS;
        ALL_COMPOSITE_META_TAGS
            .iter()
            .map(|&tag| self.compose_meta_prefix_vec(tag))
            .collect()
    }

    #[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.bf_prefix(key),
            self.cf_prefix(key),
            self.bm_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),
            self.ft_index_prefix(key),
        ]
    }

    #[inline]
    pub fn namespace_prefix_for(ns: &str) -> Vec<u8> {
        if is_default_namespace(ns) {
            Vec::new()
        } else {
            let mut prefix = Vec::with_capacity(4 + ns.len() + 1);
            prefix.extend_from_slice(b"\x00ns:");
            prefix.extend_from_slice(ns.as_bytes());
            prefix.push(b':');
            prefix
        }
    }

    #[inline]
    pub fn namespace_prefix(&self) -> Vec<u8> {
        Self::namespace_prefix_for(self.ns)
    }

    #[inline]
    pub fn raw_key(&self, key: &str) -> String {
        if self.is_default() {
            key.to_string()
        } else {
            format!("\x00ns:{}:{}", self.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 v = Vec::with_capacity(4 + ns_bytes.len() + 1 + key.len());
            v.extend_from_slice(b"\x00ns:");
            v.extend_from_slice(ns_bytes);
            v.push(b':');
            v.extend_from_slice(key);
            Cow::Owned(v)
        }
    }

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

    #[inline]
    pub fn hash_key_bytes(&self, key: &str, field: &[u8]) -> Vec<u8> {
        self.compose_sub_key_vec(KeyTag::HashData.as_slice(), key, field)
    }

    #[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 {
        self.compose_meta_str(KeyTag::HashMeta.as_slice(), key)
    }

    #[inline]
    pub fn hash_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::HashData.as_slice(), key)
    }

    #[inline]
    pub fn list_meta(&self, key: &str) -> String {
        self.compose_meta_str(KeyTag::ListMeta.as_slice(), key)
    }

    #[inline]
    pub fn list_item(&self, key: &str, idx: u64) -> String {
        let mut buf = Vec::with_capacity(8 + self.ns.len() + key.len() + 16);
        self.compose_prefix_into(KeyTag::ListData.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(&u64_to_hex_16(idx));
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[inline]
    pub fn list_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::ListData.as_slice(), key)
    }

    #[inline]
    pub fn set_meta(&self, key: &str) -> String {
        self.compose_meta_str(KeyTag::SetMeta.as_slice(), key)
    }

    #[inline]
    pub fn set_key(&self, key: &str, member: &str) -> String {
        self.compose_sub_key_str(KeyTag::SetData.as_slice(), 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 {
        let m = String::from_utf8_lossy(member);
        self.compose_sub_key_str(KeyTag::SetData.as_slice(), key, &m)
    }

    #[inline]
    pub fn set_key_bytes(&self, key: &str, member: &[u8]) -> Vec<u8> {
        self.compose_sub_key_vec(KeyTag::SetData.as_slice(), key, member)
    }

    #[inline]
    pub fn set_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::SetData.as_slice(), key)
    }

    #[inline]
    pub fn zset_meta(&self, key: &str) -> String {
        self.compose_meta_str(KeyTag::ZSetMeta.as_slice(), key)
    }

    #[inline]
    pub fn zset_key(&self, key: &str, member: &str) -> String {
        self.compose_sub_key_str(KeyTag::ZSetData.as_slice(), key, member)
    }

    #[inline]
    pub fn zset_key_bytes(&self, key: &str, member: &[u8]) -> Vec<u8> {
        self.compose_sub_key_vec(KeyTag::ZSetData.as_slice(), key, member)
    }

    #[inline]
    pub fn zset_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::ZSetData.as_slice(), key)
    }

    #[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 mut buf = Vec::with_capacity(8 + self.ns.len() + key.len() + 16 + 1 + member.len());
        self.compose_prefix_into(KeyTag::ZSetScore.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(&hex_bytes);
        buf.push(b':');
        buf.extend_from_slice(member.as_bytes());
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[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> {
        self.compose_prefix_vec(KeyTag::ZSetScore.as_slice(), key)
    }

    #[inline]
    pub fn hll_meta(&self, key: &str) -> String {
        self.compose_meta_str(KeyTag::HllMeta.as_slice(), key)
    }

    #[inline]
    pub fn hll_meta_prefix(&self) -> Vec<u8> {
        self.compose_meta_prefix_vec(KeyTag::HllMeta.as_slice())
    }

    #[inline]
    pub fn hll_key(&self, key: &str) -> String {
        self.compose_meta_str(KeyTag::HllRaw.as_slice(), key)
    }

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

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

    #[inline]
    pub fn bf_item(&self, key: &str, filter_idx: u16) -> String {
        let mut buf = Vec::with_capacity(8 + self.ns.len() + key.len() + 4);
        self.compose_prefix_into(KeyTag::BloomData.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(&u16_to_hex_4(filter_idx));
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[inline]
    pub fn bf_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::BloomData.as_slice(), key)
    }

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

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

    #[inline]
    pub fn cf_page(&self, key: &str, filter_idx: u16, page_idx: u32) -> String {
        let mut buf = Vec::with_capacity(8 + self.ns.len() + key.len() + 4 + 1 + 8);
        self.compose_prefix_into(KeyTag::CuckooData.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(&u16_to_hex_4(filter_idx));
        buf.push(b':');
        buf.extend_from_slice(&u32_to_hex_8(page_idx));
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[inline]
    pub fn cf_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::CuckooData.as_slice(), key)
    }

    #[inline]
    pub fn si_meta(&self, key: &str) -> String {
        self.compose_meta_str(KeyTag::SortedIntMeta.as_slice(), key)
    }

    #[inline]
    pub fn si_key(&self, key: &str, id: u64) -> String {
        let mut buf = Vec::with_capacity(8 + self.ns.len() + key.len() + 16);
        self.compose_prefix_into(KeyTag::SortedIntData.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(&u64_to_hex_16(id));
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[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> {
        self.compose_prefix_vec(KeyTag::SortedIntData.as_slice(), key)
    }

    #[inline]
    pub fn bm_meta(&self, key: &str) -> String {
        self.compose_meta_str(KeyTag::BitmapMeta.as_slice(), key)
    }

    #[inline]
    pub fn bm_segment(&self, key: &str, seg_idx: u32) -> String {
        let mut buf = Vec::with_capacity(8 + self.ns.len() + key.len() + 8);
        self.compose_prefix_into(KeyTag::BitmapData.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(&u32_to_hex_8(seg_idx));
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[inline]
    pub fn bm_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::BitmapData.as_slice(), key)
    }

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

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

    #[inline]
    pub fn stream_item(&self, key: &str, ms: u64, seq: u64) -> String {
        let mut buf = Vec::with_capacity(8 + self.ns.len() + key.len() + 16 + 1 + 16);
        self.compose_prefix_into(KeyTag::StreamData.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(&u64_to_hex_16(ms));
        buf.push(b':');
        buf.extend_from_slice(&u64_to_hex_16(seq));
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[inline]
    pub fn stream_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::StreamData.as_slice(), key)
    }

    #[inline]
    pub fn stream_group_meta(&self, key: &str, group: &str) -> String {
        self.compose_sub_key_str(KeyTag::StreamGroup.as_slice(), key, group)
    }

    #[inline]
    pub fn stream_group_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::StreamGroup.as_slice(), key)
    }

    #[inline]
    pub fn stream_consumer_meta(&self, key: &str, group: &str, consumer: &str) -> String {
        let sub = format!("{group}:{consumer}");
        self.compose_sub_key_str(KeyTag::StreamConsumer.as_slice(), key, &sub)
    }

    #[inline]
    pub fn stream_consumer_prefix(&self, key: &str, group: &str) -> Vec<u8> {
        let mut buf = self.compose_prefix_vec(KeyTag::StreamConsumer.as_slice(), key);
        buf.extend_from_slice(group.as_bytes());
        buf.push(b':');
        buf
    }

    #[inline]
    pub fn stream_consumer_prefix_all(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::StreamConsumer.as_slice(), key)
    }

    #[inline]
    pub fn stream_pel_item(&self, key: &str, group: &str, ms: u64, seq: u64) -> String {
        let mut buf =
            Vec::with_capacity(8 + self.ns.len() + key.len() + group.len() + 1 + 16 + 1 + 16);
        self.compose_prefix_into(KeyTag::StreamPel.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(group.as_bytes());
        buf.push(b':');
        buf.extend_from_slice(&u64_to_hex_16(ms));
        buf.push(b':');
        buf.extend_from_slice(&u64_to_hex_16(seq));
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[inline]
    pub fn stream_pel_prefix(&self, key: &str, group: &str) -> Vec<u8> {
        let mut buf = self.compose_prefix_vec(KeyTag::StreamPel.as_slice(), key);
        buf.extend_from_slice(group.as_bytes());
        buf.push(b':');
        buf
    }

    #[inline]
    pub fn stream_pel_prefix_all(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::StreamPel.as_slice(), key)
    }

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

    #[inline]
    pub fn json_meta_prefix(&self) -> Vec<u8> {
        self.compose_meta_prefix_vec(KeyTag::JsonMeta.as_slice())
    }

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

    #[inline]
    pub fn ft_schema(&self, index: &str) -> String {
        self.compose_meta_str(KeyTag::FtSchema.as_slice(), index)
    }

    #[inline]
    pub fn ft_schema_prefix(&self) -> Vec<u8> {
        self.compose_meta_prefix_vec(KeyTag::FtSchema.as_slice())
    }

    #[inline]
    pub fn ft_alias(&self, alias: &str) -> String {
        self.compose_meta_str(KeyTag::FtAlias.as_slice(), alias)
    }

    #[inline]
    pub fn ft_alias_prefix(&self) -> Vec<u8> {
        self.compose_meta_prefix_vec(KeyTag::FtAlias.as_slice())
    }

    #[inline]
    pub fn ft_index_key(&self, index: &str, field: &str, term: &str, doc_id: &str) -> String {
        let sub = format!("{field}:{term}:{doc_id}");
        self.compose_sub_key_str(KeyTag::FtIndex.as_slice(), index, &sub)
    }

    #[inline]
    pub fn ft_index_prefix(&self, index: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::FtIndex.as_slice(), index)
    }

    #[inline]
    pub fn ft_index_field_prefix(&self, index: &str, field: &str) -> Vec<u8> {
        let mut buf = self.compose_prefix_vec(KeyTag::FtIndex.as_slice(), index);
        buf.extend_from_slice(field.as_bytes());
        buf.push(b':');
        buf
    }

    #[inline]
    pub fn ft_index_term_prefix(&self, index: &str, field: &str, term: &str) -> Vec<u8> {
        let mut buf = self.compose_prefix_vec(KeyTag::FtIndex.as_slice(), index);
        buf.extend_from_slice(field.as_bytes());
        buf.push(b':');
        buf.extend_from_slice(term.as_bytes());
        buf.push(b':');
        buf
    }

    #[inline]
    pub fn ft_index_term_scan_prefix(
        &self,
        index: &str,
        field: &str,
        term_prefix: &str,
    ) -> Vec<u8> {
        let mut buf = self.compose_prefix_vec(KeyTag::FtIndex.as_slice(), index);
        buf.extend_from_slice(field.as_bytes());
        buf.push(b':');
        buf.extend_from_slice(term_prefix.as_bytes());
        buf
    }

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

    #[inline]
    pub fn tdigest_meta_prefix(&self) -> Vec<u8> {
        self.compose_meta_prefix_vec(KeyTag::TDigestMeta.as_slice())
    }

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

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

    #[inline]
    pub fn ts_meta_prefix(&self) -> Vec<u8> {
        self.compose_meta_prefix_vec(KeyTag::TimeSeriesMeta.as_slice())
    }

    #[inline]
    pub fn ts_item(&self, key: &str, timestamp: u64) -> String {
        let mut buf = Vec::with_capacity(8 + self.ns.len() + key.len() + 16);
        self.compose_prefix_into(KeyTag::TimeSeriesData.as_slice(), key.as_bytes(), &mut buf);
        buf.extend_from_slice(&u64_to_hex_16(timestamp));
        unsafe { String::from_utf8_unchecked(buf) }
    }

    #[inline]
    pub fn ts_prefix(&self, key: &str) -> Vec<u8> {
        self.compose_prefix_vec(KeyTag::TimeSeriesData.as_slice(), key)
    }

    #[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（单字节 KeyTag 分派，极致速度）
    #[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;
    }
    KeyTag::from_u8(remain[0]).is_some()
}

#[inline]
fn strip_last_colons(bytes: &[u8], count: usize) -> Option<&[u8]> {
    let mut cur = bytes;
    for _ in 0..count {
        let pos = memrchr(b':', cur)?;
        cur = &cur[..pos];
    }
    Some(cur)
}

#[inline]
fn parse_tag_and_key(remain: &[u8]) -> Option<&[u8]> {
    if remain.is_empty() {
        return Some(remain);
    }
    let Some(tag) = KeyTag::from_u8(remain[0]) else {
        return Some(remain);
    };
    let body = &remain[1..];
    let Some(rest) = body.strip_prefix(b":") else {
        return Some(remain);
    };
    match tag {
        KeyTag::HashMeta
        | KeyTag::ListMeta
        | KeyTag::SetMeta
        | KeyTag::ZSetMeta
        | KeyTag::BloomMeta
        | KeyTag::CuckooMeta
        | KeyTag::BitmapMeta
        | KeyTag::HllMeta
        | KeyTag::HllRaw
        | KeyTag::JsonMeta
        | KeyTag::SortedIntMeta
        | KeyTag::StreamMeta
        | KeyTag::TDigestMeta
        | KeyTag::TimeSeriesMeta
        | KeyTag::FtSchema
        | KeyTag::FtAlias => Some(rest),

        KeyTag::HashData
        | KeyTag::ListData
        | KeyTag::SetData
        | KeyTag::ZSetData
        | KeyTag::BloomData
        | KeyTag::BitmapData
        | KeyTag::SortedIntData
        | KeyTag::StreamGroup
        | KeyTag::TDigestData
        | KeyTag::TimeSeriesData => strip_last_colons(rest, 1).or(Some(rest)),

        KeyTag::CuckooData | KeyTag::StreamData | KeyTag::StreamConsumer | KeyTag::ZSetScore => {
            strip_last_colons(rest, 2).or(Some(rest))
        }

        KeyTag::StreamPel | KeyTag::FtIndex => strip_last_colons(rest, 3).or(Some(rest)),

        KeyTag::JsonData | KeyTag::FtData => Some(rest),
    }
}