wedb_embed 0.1.2

use crate::key_composer::{
  ns::is_default_namespace,
  oppv::{
    decode_oppv_u64, encode_oppv_u64, encode_oppv_u64_fixed, encode_oppv_u64_slice, oppv_len_u64,
  },
  small_key::SmallKey,
  tag::{KeyTag, ScopeModeTag},
};

/// 通用子键复用构建器（零堆分配，支持高频循环迭代）
#[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_be(&mut self, val: u64) -> &[u8] {
    self.buf.truncate(self.prefix_len);
    self.buf.extend_from_slice(&val.to_be_bytes());
    &self.buf
  }

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

/// 统一的物理键编排器（负责多租户隔离、复合数据结构 Key 与子键前缀构造，全面支持纯数字 OPPV 变长保序编码与前缀无关帧化）
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct KeyComposer<'a> {
  ns: &'a str,
  ns_id: u64,
  db: u64,
}

impl<'a> KeyComposer<'a> {
  /// 构造默认命名空间的 KeyComposer（ns_id = 0, db = 0）
  ///
  /// 非默认命名空间必须通过 `new_named()` 传入数据库分配的持久化自增 ID
  #[inline]
  pub fn new(ns: &'a str) -> Self {
    debug_assert!(
      is_default_namespace(ns),
      "KeyComposer::new() only accepts default namespace, use new_named() with a database-allocated ns_id for '{ns}'"
    );
    Self {
      ns,
      ns_id: 0,
      db: 0,
    }
  }

  #[inline]
  pub const fn new_db(db: u64) -> Self {
    Self {
      ns: "default",
      ns_id: 0,
      db,
    }
  }

  #[inline]
  pub const fn new_named(ns: &'a str, ns_id: u64, db: u64) -> Self {
    Self { ns, ns_id, db }
  }

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

  #[inline(always)]
  pub const fn ns_id(&self) -> u64 {
    self.ns_id
  }

  #[inline(always)]
  pub const fn db(&self) -> u64 {
    self.db
  }

  #[inline(always)]
  pub const fn is_default(&self) -> bool {
    self.ns_id == 0 && self.db == 0
  }

  /// 栈上定长编码作用域前缀到 24 字节数组（零堆分配，零 SmallVec 开销，返回写入字节数）
  ///
  /// 模式 0: 空; 模式 1: \x00\x01[oppv(db)]; 模式 2: \x00\x02[oppv(ns_id)]; 模式 3: \x00\x03[oppv(ns_id)][oppv(db)]
  #[inline(always)]
  pub fn encode_scope_prefix_fixed(&self, buf: &mut [u8; 24]) -> usize {
    if self.is_default() {
      return 0;
    }
    if self.ns_id == 0 {
      buf[0..2].copy_from_slice(ScopeModeTag::DefaultDb.as_prefix_slice());
      let len = encode_oppv_u64_slice(self.db, &mut buf[2..]);
      2 + len
    } else if self.db == 0 {
      buf[0..2].copy_from_slice(ScopeModeTag::Tenant.as_prefix_slice());
      let len = encode_oppv_u64_slice(self.ns_id, &mut buf[2..]);
      2 + len
    } else {
      buf[0..2].copy_from_slice(ScopeModeTag::TenantDb.as_prefix_slice());
      let len1 = encode_oppv_u64_slice(self.ns_id, &mut buf[2..]);
      let len2 = encode_oppv_u64_slice(self.db, &mut buf[2 + len1..]);
      2 + len1 + len2
    }
  }

  /// 编码作用域物理前缀到 Vec<u8>
  #[inline]
  pub fn encode_scope_prefix(&self, buf: &mut Vec<u8>) {
    let mut tmp = [0u8; 24];
    let len = self.encode_scope_prefix_fixed(&mut tmp);
    buf.extend_from_slice(&tmp[..len]);
  }

  /// 编码作用域物理前缀到 SmallKey（栈上零堆分配）
  #[inline]
  pub fn encode_scope_prefix_small(&self, sk: &mut SmallKey) {
    let mut buf = [0u8; 24];
    let len = self.encode_scope_prefix_fixed(&mut buf);
    sk.extend_from_slice(&buf[..len]);
  }

  /// 计算作用域前缀的物理字节长度（纯 CPU 算术判断，零堆内存分配）
  #[inline(always)]
  pub const fn scope_prefix_len(&self) -> usize {
    if self.is_default() {
      0
    } else if self.ns_id == 0 {
      2 + oppv_len_u64(self.db)
    } else if self.db == 0 {
      2 + oppv_len_u64(self.ns_id)
    } else {
      2 + oppv_len_u64(self.ns_id) + oppv_len_u64(self.db)
    }
  }

  /// 构造复合结构元数据存储键（格式：`[scope_prefix][tag][key]`）
  #[inline]
  pub fn compose_meta_key_into(&self, tag: &[u8], key_bytes: &[u8], buf: &mut Vec<u8>) {
    buf.clear();
    buf.reserve(self.scope_prefix_len() + tag.len() + key_bytes.len());
    self.encode_scope_prefix(buf);
    buf.extend_from_slice(tag);
    buf.extend_from_slice(key_bytes);
  }

  #[inline]
  pub fn compose_meta_key_stack(&self, tag: &[u8], key_bytes: &[u8]) -> SmallKey {
    let mut sk = SmallKey::new();
    let mut buf = [0u8; 24];
    let len = self.encode_scope_prefix_fixed(&mut buf);
    sk.extend_from_slice(&buf[..len]);
    sk.extend_from_slice(tag);
    sk.extend_from_slice(key_bytes);
    sk
  }

  /// 构造复合结构子键数据前缀，并额外预留 extra_len 容量（避免二次内存重分配）
  #[inline]
  pub fn compose_prefix_into_with_extra(
    &self,
    tag: &[u8],
    key_bytes: &[u8],
    extra_len: usize,
    buf: &mut Vec<u8>,
  ) {
    buf.clear();
    buf.reserve(
      self.scope_prefix_len()
        + tag.len()
        + oppv_len_u64(key_bytes.len() as u64)
        + key_bytes.len()
        + extra_len,
    );
    self.encode_scope_prefix(buf);
    buf.extend_from_slice(tag);
    encode_oppv_u64(key_bytes.len() as u64, buf);
    buf.extend_from_slice(key_bytes);
  }

  /// 构造复合结构子键数据前缀（前缀无关编码：`[scope_prefix][tag][oppv(len(key))][key]`）
  #[inline]
  pub fn compose_prefix_into(&self, tag: &[u8], key_bytes: &[u8], buf: &mut Vec<u8>) {
    self.compose_prefix_into_with_extra(tag, key_bytes, 0, buf);
  }

  #[inline]
  pub fn compose_prefix(&self, tag: &[u8], key_bytes: &[u8]) -> Vec<u8> {
    let cap =
      self.scope_prefix_len() + tag.len() + oppv_len_u64(key_bytes.len() as u64) + key_bytes.len();
    let mut buf = Vec::with_capacity(cap);
    self.encode_scope_prefix(&mut buf);
    buf.extend_from_slice(tag);
    encode_oppv_u64(key_bytes.len() as u64, &mut buf);
    buf.extend_from_slice(key_bytes);
    buf
  }

  #[inline]
  pub fn compose_prefix_stack(&self, tag: &[u8], key_bytes: &[u8]) -> SmallKey {
    let mut sk = SmallKey::new();
    let mut buf = [0u8; 24];
    let prefix_len = self.encode_scope_prefix_fixed(&mut buf);
    sk.extend_from_slice(&buf[..prefix_len]);
    sk.extend_from_slice(tag);
    let mut tmp = [0u8; 9];
    let len = encode_oppv_u64_fixed(key_bytes.len() as u64, &mut tmp);
    sk.extend_from_slice(&tmp[..len]);
    sk.extend_from_slice(key_bytes);
    sk
  }

  #[inline]
  pub fn compose_subkey_stack(&self, tag: &[u8], key_bytes: &[u8], subkey: &[u8]) -> SmallKey {
    let mut sk = SmallKey::new();
    let mut buf = [0u8; 24];
    let prefix_len = self.encode_scope_prefix_fixed(&mut buf);
    sk.extend_from_slice(&buf[..prefix_len]);
    sk.extend_from_slice(tag);
    let mut tmp = [0u8; 9];
    let len = encode_oppv_u64_fixed(key_bytes.len() as u64, &mut tmp);
    sk.extend_from_slice(&tmp[..len]);
    sk.extend_from_slice(key_bytes);
    sk.extend_from_slice(subkey);
    sk
  }

  /// 栈上定长构造带有两个子键分量的存储键（零堆分配，如 score 索引、时间戳分块、布谷鸟分页）
  #[inline]
  pub fn compose_subkey2_stack(
    &self,
    tag: &[u8],
    key_bytes: &[u8],
    sub1: &[u8],
    sub2: &[u8],
  ) -> SmallKey {
    let mut sk = SmallKey::new();
    let mut buf = [0u8; 24];
    let prefix_len = self.encode_scope_prefix_fixed(&mut buf);
    sk.extend_from_slice(&buf[..prefix_len]);
    sk.extend_from_slice(tag);
    let mut tmp = [0u8; 9];
    let len = encode_oppv_u64_fixed(key_bytes.len() as u64, &mut tmp);
    sk.extend_from_slice(&tmp[..len]);
    sk.extend_from_slice(key_bytes);
    sk.extend_from_slice(sub1);
    sk.extend_from_slice(sub2);
    sk
  }

  /// 栈上定长构造带有 OPPV 长度前缀的层级子键（零堆分配，如 Stream 消费者元数据、PEL 项）
  #[inline]
  pub fn compose_oppv_subkey_stack(
    &self,
    tag: &[u8],
    key_bytes: &[u8],
    mid_bytes: &[u8],
    sub: &[u8],
  ) -> SmallKey {
    let mut sk = SmallKey::new();
    let mut buf = [0u8; 24];
    let prefix_len = self.encode_scope_prefix_fixed(&mut buf);
    sk.extend_from_slice(&buf[..prefix_len]);
    sk.extend_from_slice(tag);
    let mut tmp = [0u8; 9];
    let len1 = encode_oppv_u64_fixed(key_bytes.len() as u64, &mut tmp);
    sk.extend_from_slice(&tmp[..len1]);
    sk.extend_from_slice(key_bytes);
    let len2 = encode_oppv_u64_fixed(mid_bytes.len() as u64, &mut tmp);
    sk.extend_from_slice(&tmp[..len2]);
    sk.extend_from_slice(mid_bytes);
    sk.extend_from_slice(sub);
    sk
  }

  #[inline]
  pub fn compose_meta_prefix_stack(&self, tag: &[u8]) -> SmallKey {
    let mut sk = SmallKey::new();
    let mut buf = [0u8; 24];
    let prefix_len = self.encode_scope_prefix_fixed(&mut buf);
    sk.extend_from_slice(&buf[..prefix_len]);
    sk.extend_from_slice(tag);
    sk
  }

  #[inline]
  pub fn compose_meta_prefix(&self, tag: &[u8]) -> Vec<u8> {
    let mut v = Vec::with_capacity(self.scope_prefix_len() + tag.len());
    self.encode_scope_prefix(&mut v);
    v.extend_from_slice(tag);
    v
  }

  #[inline]
  pub fn namespace_prefix(&self) -> Vec<u8> {
    let mut v = Vec::with_capacity(self.scope_prefix_len());
    self.encode_scope_prefix(&mut v);
    v
  }

  // ==================== 作用域反查与提取 ====================

  /// 从多租户/多库物理键中严格解析作用域前缀（适用于 data_ns / meta_ns 列族中的物理键）
  ///
  /// 返回 `(KeyComposer, prefix_len, remain_slice)`
  #[inline]
  pub fn parse_scoped_prefix(full_key: &[u8]) -> Option<(Self, usize, &[u8])> {
    if full_key.len() < 3 || full_key[0] != 0 {
      return None;
    }
    let scope_tag = ScopeModeTag::from_u8(full_key[1])?;
    match scope_tag {
      ScopeModeTag::DefaultDb => {
        let (db, c) = decode_oppv_u64(&full_key[2..])?;
        let prefix_len = 2 + c;
        if full_key.len() > prefix_len {
          Some((Self::new_db(db), prefix_len, &full_key[prefix_len..]))
        } else {
          None
        }
      }
      ScopeModeTag::Tenant => {
        let (ns_id, c) = decode_oppv_u64(&full_key[2..])?;
        let prefix_len = 2 + c;
        if full_key.len() > prefix_len {
          Some((
            Self::new_named("", ns_id, 0),
            prefix_len,
            &full_key[prefix_len..],
          ))
        } else {
          None
        }
      }
      ScopeModeTag::TenantDb => {
        let (ns_id, c1) = decode_oppv_u64(full_key.get(2..)?)?;
        let (db, c2) = decode_oppv_u64(full_key.get(2 + c1..)?)?;
        let prefix_len = 2 + c1 + c2;
        if full_key.len() > prefix_len {
          Some((
            Self::new_named("", ns_id, db),
            prefix_len,
            &full_key[prefix_len..],
          ))
        } else {
          None
        }
      }
    }
  }

  /// 判断物理键是否属于当前 KeyComposer 所在的作用域
  #[inline(always)]
  pub fn is_key_in_ns(&self, full_key: &[u8]) -> bool {
    if self.is_default() {
      !full_key.is_empty() && KeyTag::from_u8(full_key[0]).is_some()
    } else {
      let mut buf = [0u8; 24];
      let len = self.encode_scope_prefix_fixed(&mut buf);
      let prefix_slice = &buf[..len];
      full_key.starts_with(prefix_slice)
        && full_key.len() > prefix_slice.len()
        && KeyTag::from_u8(full_key[prefix_slice.len()]).is_some()
    }
  }

  /// 高性能零分配提取用户 Key（基于物理类型前缀与 OPPV 长度编码精准分帧，100% 免疫任何特殊字节与冒号）
  #[inline]
  pub fn extract_user_key<'b>(&self, full_key: &'b [u8]) -> Option<&'b [u8]> {
    let remain = if self.is_default() {
      full_key
    } else {
      let mut buf = [0u8; 24];
      let len = self.encode_scope_prefix_fixed(&mut buf);
      let prefix_slice = &buf[..len];
      if !full_key.starts_with(prefix_slice) {
        return None;
      }
      &full_key[len..]
    };

    if remain.is_empty() {
      return None;
    }

    let tag = KeyTag::from_u8(remain[0])?;

    match tag {
      KeyTag::RawString
      | 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(&remain[1..]),

      KeyTag::HashData
      | KeyTag::ListData
      | KeyTag::SetData
      | KeyTag::ZSetData
      | KeyTag::ZSetScore
      | KeyTag::BloomData
      | KeyTag::CuckooData
      | KeyTag::BitmapData
      | KeyTag::JsonData
      | KeyTag::SortedIntData
      | KeyTag::StreamData
      | KeyTag::StreamGroup
      | KeyTag::StreamConsumer
      | KeyTag::StreamPel
      | KeyTag::TDigestData
      | KeyTag::TimeSeriesData
      | KeyTag::FtIndex
      | KeyTag::FtData => {
        let (key_len, consumed) = decode_oppv_u64(&remain[1..])?;
        let start = 1 + consumed;
        let end = start.checked_add(key_len as usize)?;
        remain.get(start..end)
      }
    }
  }

  /// 将当前命名空间的底层键转换为目标命名空间的底层键（零拷贝/二进制安全）
  #[inline]
  pub fn transform_key_to_target_bytes(
    &self,
    full_key: &[u8],
    target_kc: &KeyComposer<'_>,
  ) -> Option<Vec<u8>> {
    let remain = if self.is_default() {
      if full_key.is_empty() || KeyTag::from_u8(full_key[0]).is_none() {
        return None;
      }
      full_key
    } else {
      let mut buf = [0u8; 24];
      let len = self.encode_scope_prefix_fixed(&mut buf);
      let prefix_slice = &buf[..len];
      if !full_key.starts_with(prefix_slice) {
        return None;
      }
      let rem = &full_key[len..];
      if rem.is_empty() || KeyTag::from_u8(rem[0]).is_none() {
        return None;
      }
      rem
    };

    let mut out = Vec::with_capacity(target_kc.scope_prefix_len() + remain.len());
    target_kc.encode_scope_prefix(&mut out);
    out.extend_from_slice(remain);
    Some(out)
  }
}