wrecord 0.1.0

use core::{
  borrow::Borrow,
  cmp::Ordering,
  hash::{Hash, Hasher},
  mem::{align_of, size_of},
  ops::Deref,
};

use crate::{
  buf::stack_heap_buf,
  error::{Error, Result},
  simd::fast_key_eq,
  tag::KeyTag,
};

/// OPPV 单字节编码上限 (128)
pub const VARINT_1B_MAX: u64 = 128;
/// OPPV 双字节编码上限 (16,512)
pub const VARINT_2B_MAX: u64 = 16_512;
/// OPPV 三字节编码上限 (2,113,664)
pub const VARINT_3B_MAX: u64 = 2_113_664;
/// OPPV 四字节编码上限 (270,549,120)
pub const VARINT_4B_MAX: u64 = 270_549_120;

/// OPPV 九字节满位前缀标志位 (0xFF)
pub const VARINT_9B_MARKER: u8 = 0xFF;

/// OPPV 单字节首字节最大值 (0x7F)
pub const VARINT_1B_FIRST_BYTE_MAX: u8 = 0x7F;
/// OPPV 单字节首字节门限值 (0x80)，小于此门限必定为单字节 OPPV
pub const VARINT_1B_FIRST_BYTE_LIMIT: u8 = 0x80;

/// OPPV 双字节前缀标志位 (0x80) 与首字节最大值 (0xBF)
pub const VARINT_2B_MARKER: u8 = 0x80;
pub const VARINT_2B_FIRST_BYTE_MAX: u8 = 0xBF;
/// OPPV 双字节首字节有效载荷掩码 (6 位载荷, 0x3F)
pub const VARINT_2B_PAYLOAD_MASK: u8 = 0x3F;

/// OPPV 三字节前缀标志位 (0xC0) 与首字节最大值 (0xDF)
pub const VARINT_3B_MARKER: u8 = 0xC0;
pub const VARINT_3B_FIRST_BYTE_MAX: u8 = 0xDF;
/// OPPV 三字节首字节有效载荷掩码 (5 位载荷, 0x1F)
pub const VARINT_3B_PAYLOAD_MASK: u8 = 0x1F;

/// OPPV 四字节前缀标志位 (0xE0) 与首字节最大值 (0xEF)
pub const VARINT_4B_MARKER: u8 = 0xE0;
pub const VARINT_4B_FIRST_BYTE_MAX: u8 = 0xEF;
/// OPPV 四字节首字节有效载荷掩码 (4 位载荷, 0x0F)
pub const VARINT_4B_PAYLOAD_MASK: u8 = 0x0F;

/// 方案 A 最简会话前缀字节长度 (ns=1B + db=1B = 2B，覆盖 99.9% 业务场景)
pub const MIN_SESSION_PREFIX_LEN: usize = 2;

/// 方案 A 子键公共元数据头字节数 (key_id: 8B + version: 8B = 16B)
pub const SUBKEY_META_HEADER_LEN: usize = 16;
/// 方案 A 集合分块子键 chunk_id 长度 (4 字节)
pub const CHUNK_ID_LEN: usize = 4;
/// 单个 64 位大端整型字节数 (8 字节)
pub const U64_BYTE_LEN: usize = 8;
/// 方案 A 最简物理子键最小字节数 (ns: 1B + db: 1B + KeyTag: 1B + key_id: 8B + version: 8B = 19B)
pub const MIN_SUBKEY_LEN: usize = MIN_SESSION_PREFIX_LEN + KeyTag::TAG_LEN + SUBKEY_META_HEADER_LEN;
/// 方案 A 最简集合分块物理子键最小字节数 (19B + chunk_id: 4B = 23B)
pub const MIN_CHUNK_KEY_LEN: usize = MIN_SUBKEY_LEN + CHUNK_ID_LEN;

/// 单个变长数值最大编码字节数 (9 字节)
pub const MAX_VARINT_LEN: usize = 9;
/// 会话前缀最大编码字节数 (NsVarint 9B + DbVarint 9B = 18 字节)
pub const MAX_SESSION_PREFIX_LEN: usize = 18;

/// 栈分配键容量上限 (62 字节，留出 1 字节长度与 1 字节枚举鉴别符，刚好满足 64 字节单缓存行)
pub const STACK_KEY_CAP: usize = 62;

/// 预计算的会话 (Namespace, DB) 变长前缀定长缓冲区
///
/// 结构紧凑（18 字节固定数组 + 1 字节长度 = 19 字节），支持 `Copy`。
/// 在客户端连接或执行 `SELECT <db>` 时单次计算常驻会话，后续热路径彻底免去变长编码开销。
#[derive(Clone, Copy, Debug)]
pub struct SessionPrefixBuf {
  buf: [u8; MAX_SESSION_PREFIX_LEN],
  len: u8,
}

impl SessionPrefixBuf {
  /// 编码命名空间与数据库编号，生成会话前缀 (const fn)
  #[inline]
  pub const fn new(ns: u64, db: u64) -> Self {
    let (buf, len) = NamespaceDbCodec::encode_session_prefix_to_array(ns, db);
    Self {
      buf,
      len: len as u8,
    }
  }

  /// 从已知切片安全解析并验证会话前缀 (const fn)
  #[inline]
  pub const fn from_slice(slice: &[u8]) -> Result<Self> {
    if slice.is_empty() || slice.len() > MAX_SESSION_PREFIX_LEN {
      return Err(Error::BufferTooShort {
        expected: 1,
        actual: slice.len(),
      });
    }
    let (_, ns_len) = match NamespaceDbCodec::decode_varint(slice) {
      Ok(v) => v,
      Err(e) => return Err(e),
    };
    let rest = match slice.split_at_checked(ns_len) {
      Some((_, r)) => r,
      None => {
        return Err(Error::BufferTooShort {
          expected: ns_len,
          actual: slice.len(),
        });
      }
    };
    let (_, db_len) = match NamespaceDbCodec::decode_varint(rest) {
      Ok(v) => v,
      Err(e) => return Err(e),
    };
    let total_len = ns_len + db_len;
    if slice.len() != total_len {
      return Err(Error::NonCanonicalEncoding);
    }
    let mut buf = [0u8; MAX_SESSION_PREFIX_LEN];
    let mut i = 0;
    while i < total_len {
      buf[i] = slice[i];
      i += 1;
    }
    Ok(Self {
      buf,
      len: total_len as u8,
    })
  }

  /// 解码会话前缀中的命名空间与数据库编号
  #[inline]
  pub fn decode(&self) -> Result<(u64, u64)> {
    let slice = self.as_slice();
    let (ns, ns_len) = NamespaceDbCodec::decode_varint(slice)?;
    let (_, rest) = slice
      .split_at_checked(ns_len)
      .ok_or(Error::BufferTooShort {
        expected: ns_len,
        actual: slice.len(),
      })?;
    let (db, _) = NamespaceDbCodec::decode_varint(rest)?;
    Ok((ns, db))
  }

  /// 获取会话前缀只读切片
  #[inline(always)]
  pub fn as_slice(&self) -> &[u8] {
    let len = (self.len as usize).min(MAX_SESSION_PREFIX_LEN);
    // 安全性保证：len <= MAX_SESSION_PREFIX_LEN (18)
    unsafe { self.buf.get_unchecked(..len) }
  }

  /// 获取前缀字节长度 (const fn)
  #[inline(always)]
  pub const fn len(&self) -> usize {
    self.len as usize
  }

  /// 是否为空（理论上最少占用 2 字节，恒为 false）
  #[inline(always)]
  pub const fn is_empty(&self) -> bool {
    self.len == 0
  }
}

impl Default for SessionPrefixBuf {
  #[inline]
  fn default() -> Self {
    Self::new(0, 0)
  }
}

impl Deref for SessionPrefixBuf {
  type Target = [u8];

  #[inline(always)]
  fn deref(&self) -> &Self::Target {
    self.as_slice()
  }
}

impl AsRef<[u8]> for SessionPrefixBuf {
  #[inline(always)]
  fn as_ref(&self) -> &[u8] {
    self.as_slice()
  }
}

impl Borrow<[u8]> for SessionPrefixBuf {
  #[inline(always)]
  fn borrow(&self) -> &[u8] {
    self.as_slice()
  }
}

impl PartialEq for SessionPrefixBuf {
  #[inline(always)]
  fn eq(&self, other: &Self) -> bool {
    fast_key_eq(self.as_slice(), other.as_slice())
  }
}

impl Eq for SessionPrefixBuf {}

impl PartialOrd for SessionPrefixBuf {
  #[inline(always)]
  fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    Some(self.cmp(other))
  }
}

impl Ord for SessionPrefixBuf {
  #[inline(always)]
  fn cmp(&self, other: &Self) -> Ordering {
    self.as_slice().cmp(other.as_slice())
  }
}

impl Hash for SessionPrefixBuf {
  #[inline(always)]
  fn hash<H: Hasher>(&self, state: &mut H) {
    self.as_slice().hash(state);
  }
}

impl PartialEq<[u8]> for SessionPrefixBuf {
  #[inline(always)]
  fn eq(&self, other: &[u8]) -> bool {
    fast_key_eq(self.as_slice(), other)
  }
}

impl PartialEq<&[u8]> for SessionPrefixBuf {
  #[inline(always)]
  fn eq(&self, other: &&[u8]) -> bool {
    fast_key_eq(self.as_slice(), other)
  }
}

// 物理键内部存储排布表示
stack_heap_buf!(KeyBufRepr, STACK_KEY_CAP);

/// 64 字节单缓存行对齐物理键缓冲区
///
/// 严格对齐 CPU L1 缓存行（64 Bytes），彻底消除多核缓存颠簸与伪共享。
/// 方案 A 排布：`[NsVarint] + [DbVarint] + [KeyTag: 1B] + [Payload]`。
/// 在典型 Redis 场景下（ns < 128, db < 128, key <= 59B），键总长 <= 62B，全程零堆分配。
#[repr(C, align(64))]
#[derive(Debug, Clone)]
pub struct TaggedKeyBuf {
  /// 底层物理键存储排布（公开字段供底层引擎与测试直接访问）
  pub inner: KeyBufRepr,
}

// 编译期静态断言：TaggedKeyBuf 严格 64 字节尺寸与 64 字节 L1 缓存行对齐
const _: () = assert!(size_of::<TaggedKeyBuf>() == 64);
const _: () = assert!(align_of::<TaggedKeyBuf>() == 64);
const _: () = assert!(size_of::<KeyBufRepr>() <= 64);
const _: () = assert!(size_of::<SessionPrefixBuf>() == 19);

impl TaggedKeyBuf {
  /// 判断是否使用栈缓冲区存储
  #[inline(always)]
  pub const fn is_stack(&self) -> bool {
    matches!(self.inner, KeyBufRepr::Stack(..))
  }

  /// 判断是否已回退至堆内存存储
  #[inline(always)]
  pub const fn is_heap(&self) -> bool {
    matches!(self.inner, KeyBufRepr::Heap(..))
  }

  /// 构造空键缓冲区 (const fn, 默认栈存储)
  #[inline(always)]
  pub const fn new() -> Self {
    Self::from_stack([0u8; STACK_KEY_CAP], 0)
  }

  /// 从栈分配构造
  #[inline(always)]
  pub const fn from_stack(buf: [u8; STACK_KEY_CAP], len: u8) -> Self {
    Self {
      inner: KeyBufRepr::Stack(buf, len),
    }
  }

  /// 从堆分配构造
  #[inline(always)]
  pub const fn from_heap(vec: Vec<u8>) -> Self {
    Self {
      inner: KeyBufRepr::Heap(vec),
    }
  }

  /// 获取键的只读切片
  #[inline(always)]
  pub fn as_slice(&self) -> &[u8] {
    self.inner.as_slice()
  }

  /// 获取键的字节长度
  #[inline(always)]
  pub fn len(&self) -> usize {
    self.inner.len()
  }

  /// 判断键是否为空
  #[inline(always)]
  pub fn is_empty(&self) -> bool {
    self.inner.is_empty()
  }

  /// 转换为持有所有权的 `Vec<u8>`
  #[inline]
  pub fn into_vec(self) -> Vec<u8> {
    match self.inner {
      KeyBufRepr::Stack(buf, len) => {
        let len = (len as usize).min(STACK_KEY_CAP);
        buf[..len].to_vec()
      }
      KeyBufRepr::Heap(vec) => vec,
    }
  }
}

impl Default for TaggedKeyBuf {
  #[inline(always)]
  fn default() -> Self {
    Self::new()
  }
}

impl Deref for TaggedKeyBuf {
  type Target = [u8];

  #[inline(always)]
  fn deref(&self) -> &Self::Target {
    self.as_slice()
  }
}

impl AsRef<[u8]> for TaggedKeyBuf {
  #[inline(always)]
  fn as_ref(&self) -> &[u8] {
    self.as_slice()
  }
}

impl Borrow<[u8]> for TaggedKeyBuf {
  #[inline(always)]
  fn borrow(&self) -> &[u8] {
    self.as_slice()
  }
}

impl PartialEq for TaggedKeyBuf {
  #[inline(always)]
  fn eq(&self, other: &Self) -> bool {
    fast_key_eq(self.as_slice(), other.as_slice())
  }
}

impl Eq for TaggedKeyBuf {}

impl PartialOrd for TaggedKeyBuf {
  #[inline(always)]
  fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    Some(self.cmp(other))
  }
}

impl Ord for TaggedKeyBuf {
  #[inline(always)]
  fn cmp(&self, other: &Self) -> Ordering {
    self.as_slice().cmp(other.as_slice())
  }
}

impl Hash for TaggedKeyBuf {
  #[inline(always)]
  fn hash<H: Hasher>(&self, state: &mut H) {
    self.as_slice().hash(state);
  }
}

impl PartialEq<[u8]> for TaggedKeyBuf {
  #[inline(always)]
  fn eq(&self, other: &[u8]) -> bool {
    fast_key_eq(self.as_slice(), other)
  }
}

impl PartialEq<&[u8]> for TaggedKeyBuf {
  #[inline(always)]
  fn eq(&self, other: &&[u8]) -> bool {
    fast_key_eq(self.as_slice(), other)
  }
}

impl From<Vec<u8>> for TaggedKeyBuf {
  #[inline]
  fn from(vec: Vec<u8>) -> Self {
    if vec.len() <= STACK_KEY_CAP {
      let mut buf = [0u8; STACK_KEY_CAP];
      buf[..vec.len()].copy_from_slice(&vec);
      Self::from_stack(buf, vec.len() as u8)
    } else {
      Self::from_heap(vec)
    }
  }
}

impl From<&[u8]> for TaggedKeyBuf {
  #[inline]
  fn from(slice: &[u8]) -> Self {
    if slice.len() <= STACK_KEY_CAP {
      let mut buf = [0u8; STACK_KEY_CAP];
      buf[..slice.len()].copy_from_slice(slice);
      Self::from_stack(buf, slice.len() as u8)
    } else {
      Self::from_heap(slice.to_vec())
    }
  }
}

impl From<TaggedKeyBuf> for Vec<u8> {
  #[inline]
  fn from(buf: TaggedKeyBuf) -> Self {
    buf.into_vec()
  }
}

/// 内部高效写入前缀、标签与有效载荷到目标切片
#[inline(always)]
fn write_key_parts(prefix: &[u8], tag: KeyTag, payload: &[u8], dst: &mut [u8]) {
  let prefix_len = prefix.len();
  dst[..prefix_len].copy_from_slice(prefix);
  dst[prefix_len] = tag as u8;
  dst[prefix_len + KeyTag::TAG_LEN..prefix_len + KeyTag::TAG_LEN + payload.len()]
    .copy_from_slice(payload);
}

/// 方案 A（命名空间先行，Namespace-First）统一物理键编解码器
///
/// 全库统一物理键结构：`[NsVarint] + [DbVarint] + [KeyTag: 1B] + [Payload]`
///
/// 特性：
/// 1. 严格大端字典序保序：OPPV 编码保证数值与二进制切片单调一致。
/// 2. 局部性极高：同会话的所有键具有相同前缀，热路径下只需单次 `strip_prefix(session_prefix)`。
/// 3. 全局强正交：String 与 Meta 拥有专属 `KeyTag`（0x00 与 0x01），彻底消除裸键与长度猜测。
pub struct NamespaceDbCodec;

impl NamespaceDbCodec {
  /// 租户命名空间合法上限（18446744073709550591）
  pub const MAX_TENANT_NAMESPACE: u64 = 18446744073709550591;

  /// 计算单个 u64 数值经 OPPV 编码后占用的字节数 (1..=9, const fn)
  #[inline(always)]
  pub const fn varint_len(val: u64) -> usize {
    if val < VARINT_1B_MAX {
      1
    } else if val < VARINT_2B_MAX {
      2
    } else if val < VARINT_3B_MAX {
      3
    } else if val < VARINT_4B_MAX {
      4
    } else {
      MAX_VARINT_LEN
    }
  }

  /// 计算指定 (ns, db) 会话前缀所占用的总字节数 (2..=18, const fn)
  #[inline(always)]
  pub const fn session_prefix_len(ns: u64, db: u64) -> usize {
    Self::varint_len(ns) + Self::varint_len(db)
  }

  /// 计算完整物理键的总长度 (const fn)
  #[inline(always)]
  pub const fn key_len(ns: u64, db: u64, payload_len: usize) -> usize {
    Self::session_prefix_len(ns, db) + KeyTag::TAG_LEN + payload_len
  }

  /// 编码单一 u64 为 OPPV 变长字节数组 (const fn, 零堆分配)
  #[inline]
  pub const fn encode_varint_to_array(val: u64) -> ([u8; MAX_VARINT_LEN], usize) {
    let mut buf = [0u8; MAX_VARINT_LEN];
    if val < VARINT_1B_MAX {
      buf[0] = val as u8;
      (buf, 1)
    } else if val < VARINT_2B_MAX {
      let offset = (val - VARINT_1B_MAX) as u16;
      let bytes = offset.to_be_bytes();
      buf[0] = VARINT_2B_MARKER | bytes[0];
      buf[1] = bytes[1];
      (buf, 2)
    } else if val < VARINT_3B_MAX {
      let offset = (val - VARINT_2B_MAX) as u32;
      let bytes = offset.to_be_bytes();
      buf[0] = VARINT_3B_MARKER | bytes[1];
      buf[1] = bytes[2];
      buf[2] = bytes[3];
      (buf, 3)
    } else if val < VARINT_4B_MAX {
      let offset = (val - VARINT_3B_MAX) as u32;
      let bytes = offset.to_be_bytes();
      buf[0] = VARINT_4B_MARKER | bytes[0];
      buf[1] = bytes[1];
      buf[2] = bytes[2];
      buf[3] = bytes[3];
      (buf, 4)
    } else {
      buf[0] = VARINT_9B_MARKER;
      let bytes = val.to_be_bytes();
      buf[1] = bytes[0];
      buf[2] = bytes[1];
      buf[3] = bytes[2];
      buf[4] = bytes[3];
      buf[5] = bytes[4];
      buf[6] = bytes[5];
      buf[7] = bytes[6];
      buf[8] = bytes[7];
      (buf, MAX_VARINT_LEN)
    }
  }

  /// 编码指定 (ns, db) 为定长数组与有效长度 (const fn)
  #[inline]
  pub const fn encode_session_prefix_to_array(
    ns: u64,
    db: u64,
  ) -> ([u8; MAX_SESSION_PREFIX_LEN], usize) {
    let (ns_buf, ns_len) = Self::encode_varint_to_array(ns);
    let (db_buf, db_len) = Self::encode_varint_to_array(db);
    let mut buf = [0u8; MAX_SESSION_PREFIX_LEN];
    let mut i = 0;
    while i < ns_len {
      buf[i] = ns_buf[i];
      i += 1;
    }
    let mut j = 0;
    while j < db_len {
      buf[ns_len + j] = db_buf[j];
      j += 1;
    }
    (buf, ns_len + db_len)
  }

  /// 编码单一 u64 为 OPPV 变长字节，返回实际写入字节数
  #[inline]
  pub fn encode_varint(val: u64, dst: &mut [u8]) -> usize {
    let (arr, len) = Self::encode_varint_to_array(val);
    debug_assert!(dst.len() >= len, "目标缓冲区空间不足以写入变长整型");
    dst[..len].copy_from_slice(&arr[..len]);
    len
  }

  /// 从首字节判定单个 OPPV 变长整型的预期字节数 (const fn, 零回溯)
  #[inline(always)]
  pub const fn varint_len_from_byte(first: u8) -> Option<usize> {
    match first {
      0..=VARINT_1B_FIRST_BYTE_MAX => Some(1),
      VARINT_2B_MARKER..=VARINT_2B_FIRST_BYTE_MAX => Some(2),
      VARINT_3B_MARKER..=VARINT_3B_FIRST_BYTE_MAX => Some(3),
      VARINT_4B_MARKER..=VARINT_4B_FIRST_BYTE_MAX => Some(4),
      VARINT_9B_MARKER => Some(MAX_VARINT_LEN),
      _ => None,
    }
  }

  /// 单步自定界解码单一 OPPV 数值 (首字节确定预期长度，零回溯且报错精准, const fn)
  #[inline]
  pub const fn decode_varint(slice: &[u8]) -> Result<(u64, usize)> {
    let first = match slice {
      [first, ..] => *first,
      [] => {
        return Err(Error::BufferTooShort {
          expected: 1,
          actual: 0,
        });
      }
    };
    let expected_len = match Self::varint_len_from_byte(first) {
      Some(l) => l,
      None => return Err(Error::NonCanonicalEncoding),
    };

    if slice.len() < expected_len {
      return Err(Error::BufferTooShort {
        expected: expected_len,
        actual: slice.len(),
      });
    }

    match expected_len {
      1 => Ok((first as u64, 1)),
      2 => {
        let offset = (((first & VARINT_2B_PAYLOAD_MASK) as u64) << 8) | (slice[1] as u64);
        Ok((VARINT_1B_MAX + offset, 2))
      }
      3 => {
        let offset = (((first & VARINT_3B_PAYLOAD_MASK) as u64) << 16)
          | ((slice[1] as u64) << 8)
          | (slice[2] as u64);
        Ok((VARINT_2B_MAX + offset, 3))
      }
      4 => {
        let offset = (((first & VARINT_4B_PAYLOAD_MASK) as u64) << 24)
          | ((slice[1] as u64) << 16)
          | ((slice[2] as u64) << 8)
          | (slice[3] as u64);
        Ok((VARINT_3B_MAX + offset, 4))
      }
      MAX_VARINT_LEN => {
        let be_bytes = [
          slice[1], slice[2], slice[3], slice[4], slice[5], slice[6], slice[7], slice[8],
        ];
        let val = u64::from_be_bytes(be_bytes);
        if val < VARINT_4B_MAX {
          return Err(Error::NonCanonicalEncoding);
        }
        Ok((val, MAX_VARINT_LEN))
      }
      _ => unreachable!(),
    }
  }

  /// 编码指定 (ns, db) 会话前缀 (const fn)
  #[inline(always)]
  pub const fn encode_session_prefix(ns: u64, db: u64) -> SessionPrefixBuf {
    SessionPrefixBuf::new(ns, db)
  }

  /// 构造基于指定会话前缀切片的完整物理键缓冲区（复用前缀，零冗余）
  #[inline]
  pub fn encode_with_session_prefix(prefix: &[u8], tag: KeyTag, payload: &[u8]) -> TaggedKeyBuf {
    let total_len = prefix.len() + KeyTag::TAG_LEN + payload.len();
    if total_len <= STACK_KEY_CAP {
      let mut buf = [0u8; STACK_KEY_CAP];
      write_key_parts(prefix, tag, payload, &mut buf);
      TaggedKeyBuf::from_stack(buf, total_len as u8)
    } else {
      let mut vec = Vec::with_capacity(total_len);
      vec.extend_from_slice(prefix);
      vec.push(tag as u8);
      vec.extend_from_slice(payload);
      TaggedKeyBuf::from_heap(vec)
    }
  }

  /// 构造方案 A 物理键缓冲区: `[NsVarint] + [DbVarint] + [KeyTag: 1B] + [Payload]`
  ///
  /// 优先利用预计算的会话前缀 (SessionPrefixBuf)，消除重复分支判定与中间栈数组搬运 (1:1 对标 ns.md 规范)
  #[inline]
  pub fn encode_tagged_key(ns: u64, db: u64, tag: KeyTag, payload: &[u8]) -> TaggedKeyBuf {
    let prefix = SessionPrefixBuf::new(ns, db);
    Self::encode_with_session_prefix(prefix.as_slice(), tag, payload)
  }

  /// 便捷方法：编码普通字符串键 (KeyTag::String = 0x00)
  #[inline(always)]
  pub fn encode_string_key(ns: u64, db: u64, user_key: &[u8]) -> TaggedKeyBuf {
    Self::encode_tagged_key(ns, db, KeyTag::String, user_key)
  }

  /// 便捷方法：编码集合元数据键 (KeyTag::Meta = 0x01)
  #[inline(always)]
  pub fn encode_meta_key(ns: u64, db: u64, user_key: &[u8]) -> TaggedKeyBuf {
    Self::encode_tagged_key(ns, db, KeyTag::Meta, user_key)
  }

  /// 便捷方法：编码方案 A 集合打平子键
  /// 结构：`[NsVarint] + [DbVarint] + [KeyTag: 1B] + [key_id: 8B be] + [version: 8B be] + [field]`
  #[inline]
  pub fn encode_sub_key(
    ns: u64,
    db: u64,
    tag: KeyTag,
    key_id: u64,
    version: u64,
    field: &[u8],
  ) -> TaggedKeyBuf {
    let prefix = SessionPrefixBuf::new(ns, db);
    Self::encode_sub_key_with_prefix(prefix.as_slice(), tag, key_id, version, field)
  }

  /// 便捷方法：基于已知会话前缀编码方案 A 集合打平子键（栈优先零堆分配）
  #[inline]
  pub fn encode_sub_key_with_prefix(
    prefix: &[u8],
    tag: KeyTag,
    key_id: u64,
    version: u64,
    field: &[u8],
  ) -> TaggedKeyBuf {
    let payload_len = SUBKEY_META_HEADER_LEN + field.len();
    let total_len = prefix.len() + KeyTag::TAG_LEN + payload_len;
    let k = key_id.to_be_bytes();
    let v = version.to_be_bytes();

    if total_len <= STACK_KEY_CAP {
      let mut buf = [0u8; STACK_KEY_CAP];
      let p_len = prefix.len();
      buf[..p_len].copy_from_slice(prefix);
      buf[p_len] = tag as u8;
      let hdr_start = p_len + KeyTag::TAG_LEN;
      buf[hdr_start..hdr_start + U64_BYTE_LEN].copy_from_slice(&k);
      buf[hdr_start + U64_BYTE_LEN..hdr_start + SUBKEY_META_HEADER_LEN].copy_from_slice(&v);
      buf[hdr_start + SUBKEY_META_HEADER_LEN..total_len].copy_from_slice(field);
      TaggedKeyBuf::from_stack(buf, total_len as u8)
    } else {
      let mut vec = Vec::with_capacity(total_len);
      vec.extend_from_slice(prefix);
      vec.push(tag as u8);
      vec.extend_from_slice(&k);
      vec.extend_from_slice(&v);
      vec.extend_from_slice(field);
      TaggedKeyBuf::from_heap(vec)
    }
  }

  /// 便捷方法：编码方案 A 集合分块子键
  /// 结构：`[NsVarint] + [DbVarint] + [KeyTag: 1B] + [key_id: 8B be] + [version: 8B be] + [chunk_id: 4B be]`
  #[inline]
  pub fn encode_chunk_key(
    ns: u64,
    db: u64,
    tag: KeyTag,
    key_id: u64,
    version: u64,
    chunk_id: u32,
  ) -> TaggedKeyBuf {
    let prefix = SessionPrefixBuf::new(ns, db);
    Self::encode_chunk_key_with_prefix(prefix.as_slice(), tag, key_id, version, chunk_id)
  }

  /// 便捷方法：基于已知会话前缀编码方案 A 集合分块子键（栈优先零堆分配）
  #[inline]
  pub fn encode_chunk_key_with_prefix(
    prefix: &[u8],
    tag: KeyTag,
    key_id: u64,
    version: u64,
    chunk_id: u32,
  ) -> TaggedKeyBuf {
    Self::encode_sub_key_with_prefix(prefix, tag, key_id, version, &chunk_id.to_be_bytes())
  }

  /// 基于已知会话前缀在栈上构造物理键并执行闭包（短键全程零堆分配）
  #[inline]
  pub fn with_session_prefix<R>(
    prefix: &[u8],
    tag: KeyTag,
    payload: &[u8],
    f: impl FnOnce(&[u8]) -> R,
  ) -> R {
    let buf = Self::encode_with_session_prefix(prefix, tag, payload);
    f(buf.as_slice())
  }

  /// 在栈缓冲区上零堆分配构造物理键并执行闭包（用户键 <= 59B 全程零分配）
  #[inline]
  pub fn with_tagged_key<R>(
    ns: u64,
    db: u64,
    tag: KeyTag,
    payload: &[u8],
    f: impl FnOnce(&[u8]) -> R,
  ) -> R {
    let buf = Self::encode_tagged_key(ns, db, tag, payload);
    f(buf.as_slice())
  }

  /// 便捷闭包方法：普通字符串键
  #[inline(always)]
  pub fn with_string_key<R>(ns: u64, db: u64, user_key: &[u8], f: impl FnOnce(&[u8]) -> R) -> R {
    Self::with_tagged_key(ns, db, KeyTag::String, user_key, f)
  }

  /// 便捷闭包方法：集合元数据键
  #[inline(always)]
  pub fn with_meta_key<R>(ns: u64, db: u64, user_key: &[u8], f: impl FnOnce(&[u8]) -> R) -> R {
    Self::with_tagged_key(ns, db, KeyTag::Meta, user_key, f)
  }

  /// 将物理键原位直接写入目标预分配切片（零堆分配，零中间栈拷贝）
  #[inline]
  pub fn encode_to_slice(
    ns: u64,
    db: u64,
    tag: KeyTag,
    payload: &[u8],
    dst: &mut [u8],
  ) -> Result<usize> {
    let total_len = Self::key_len(ns, db, payload.len());
    if dst.len() < total_len {
      return Err(Error::BufferTooShort {
        expected: total_len,
        actual: dst.len(),
      });
    }
    let ns_len = Self::encode_varint(ns, dst);
    let db_len = Self::encode_varint(db, &mut dst[ns_len..]);
    let prefix_len = ns_len + db_len;
    dst[prefix_len] = tag as u8;
    dst[prefix_len + KeyTag::TAG_LEN..total_len].copy_from_slice(payload);
    Ok(total_len)
  }

  /// 将包含已知前缀的物理键写入目标切片
  #[inline]
  pub fn encode_with_session_prefix_to_slice(
    prefix: &[u8],
    tag: KeyTag,
    payload: &[u8],
    dst: &mut [u8],
  ) -> Result<usize> {
    let total_len = prefix.len() + KeyTag::TAG_LEN + payload.len();
    if dst.len() < total_len {
      return Err(Error::BufferTooShort {
        expected: total_len,
        actual: dst.len(),
      });
    }
    write_key_parts(prefix, tag, payload, dst);
    Ok(total_len)
  }

  /// 从完整物理键中解码出 `(ns, db, tag, payload)` (const fn)
  #[inline]
  pub const fn decode_tagged_key(key: &[u8]) -> Result<(u64, u64, KeyTag, &[u8])> {
    match key {
      [ns_b, db_b, tag_byte, payload @ ..]
        if *ns_b < VARINT_1B_FIRST_BYTE_LIMIT && *db_b < VARINT_1B_FIRST_BYTE_LIMIT =>
      {
        let tag = match KeyTag::from_u8(*tag_byte) {
          Some(t) => t,
          None => return Err(Error::InvalidKeyTag(*tag_byte)),
        };
        Ok((*ns_b as u64, *db_b as u64, tag, payload))
      }
      _ => {
        let (ns, ns_len) = match Self::decode_varint(key) {
          Ok(v) => v,
          Err(e) => return Err(e),
        };
        let rest = match key.split_at_checked(ns_len) {
          Some((_, r)) => r,
          None => {
            return Err(Error::BufferTooShort {
              expected: ns_len,
              actual: key.len(),
            });
          }
        };
        let (db, db_len) = match Self::decode_varint(rest) {
          Ok(v) => v,
          Err(e) => return Err(e),
        };
        let tag_slice = match rest.split_at_checked(db_len) {
          Some((_, r)) => r,
          None => {
            return Err(Error::BufferTooShort {
              expected: db_len,
              actual: rest.len(),
            });
          }
        };
        let (tag_byte, payload) = match tag_slice {
          [first, rest @ ..] => (*first, rest),
          [] => {
            return Err(Error::BufferTooShort {
              expected: 1,
              actual: 0,
            });
          }
        };
        let tag = match KeyTag::from_u8(tag_byte) {
          Some(t) => t,
          None => return Err(Error::InvalidKeyTag(tag_byte)),
        };
        Ok((ns, db, tag, payload))
      }
    }
  }

  /// 从物理键中快速剥离当前会话前缀，提取 `(tag, payload)`
  ///
  /// 若键不属于当前会话或长度不足，返回 `None`。
  /// 热路径极速过滤：无需任何变长整型解码，底层为单次 SIMD 内存比对与单字节标签解析。
  #[inline(always)]
  pub fn strip_session_prefix<'a>(
    key: &'a [u8],
    session_prefix: &[u8],
  ) -> Option<(KeyTag, &'a [u8])> {
    let rest = key.strip_prefix(session_prefix)?;
    let (&tag_byte, payload) = rest.split_first()?;
    let tag = KeyTag::from_u8(tag_byte)?;
    Some((tag, payload))
  }

  /// 剥离指定会话前缀并校验特定标签，若标签或前缀不匹配则安全返回 `None`
  #[inline(always)]
  pub fn strip_session_prefix_with_tag<'a>(
    key: &'a [u8],
    session_prefix: &[u8],
    expected_tag: KeyTag,
  ) -> Option<&'a [u8]> {
    let rest = key.strip_prefix(session_prefix)?;
    let (&tag_byte, payload) = rest.split_first()?;
    if tag_byte == expected_tag as u8 {
      Some(payload)
    } else {
      None
    }
  }

  /// 便捷方法：剥离当前会话前缀并提取普通字符串键的用户键
  #[inline(always)]
  pub fn strip_string_key<'a>(key: &'a [u8], session_prefix: &[u8]) -> Option<&'a [u8]> {
    Self::strip_session_prefix_with_tag(key, session_prefix, KeyTag::String)
  }

  /// 便捷方法：剥离当前会话前缀并提取集合元数据键的用户键
  #[inline(always)]
  pub fn strip_meta_key<'a>(key: &'a [u8], session_prefix: &[u8]) -> Option<&'a [u8]> {
    Self::strip_session_prefix_with_tag(key, session_prefix, KeyTag::Meta)
  }

  /// 从物理键中提取属于当前会话的存活用户逻辑键（String 或 Meta）
  ///
  /// 若物理键不属于当前会话，或属于集合内部打平子键（Hash/Set/List等），安全返回 `None`。
  /// 彻底消除黑名单与长度探测 hack，供 `KEYS`、`SCAN`、`FLUSHDB`、`DBSIZE` 毫秒级流式过滤。
  #[inline(always)]
  pub fn extract_live_user_key<'a>(
    key: &'a [u8],
    session_prefix: &[u8],
  ) -> Option<(KeyTag, &'a [u8])> {
    match Self::strip_session_prefix(key, session_prefix) {
      Some((tag, user_key)) if tag.is_user_visible() => Some((tag, user_key)),
      _ => None,
    }
  }

  /// 从方案 A 物理键切片中快速解析出会话前缀 (ns + db) 的总字节数 (const fn, 零堆分配零回溯)
  /// 若键长度不足或首字节前缀非法则安全返回 None
  #[inline(always)]
  pub const fn session_prefix_len_from_slice(key: &[u8]) -> Option<usize> {
    match key {
      [ns_b, db_b, ..]
        if *ns_b < VARINT_1B_FIRST_BYTE_LIMIT && *db_b < VARINT_1B_FIRST_BYTE_LIMIT =>
      {
        Some(MIN_SESSION_PREFIX_LEN)
      }
      _ => {
        let first = match key {
          [first, ..] => *first,
          [] => return None,
        };
        let ns_len = match Self::varint_len_from_byte(first) {
          Some(l) => l,
          None => return None,
        };
        let rest = match key.split_at_checked(ns_len) {
          Some((_, r)) => r,
          None => return None,
        };
        let db_first = match rest {
          [first, ..] => *first,
          [] => return None,
        };
        let db_len = match Self::varint_len_from_byte(db_first) {
          Some(l) => l,
          None => return None,
        };
        let prefix_len = ns_len + db_len;
        if key.len() >= prefix_len {
          Some(prefix_len)
        } else {
          None
        }
      }
    }
  }

  /// 从方案 A 物理键中快速解析提取 KeyTag（const fn，零回溯零分配）
  #[inline(always)]
  pub const fn decode_tag(key: &[u8]) -> Option<KeyTag> {
    match key {
      [ns_b, db_b, tag_byte, ..]
        if *ns_b < VARINT_1B_FIRST_BYTE_LIMIT && *db_b < VARINT_1B_FIRST_BYTE_LIMIT =>
      {
        KeyTag::from_u8(*tag_byte)
      }
      _ => {
        let prefix_len = match Self::session_prefix_len_from_slice(key) {
          Some(l) => l,
          None => return None,
        };
        if let Some((_, rest)) = key.split_at_checked(prefix_len) {
          match rest {
            [tag_byte, ..] => KeyTag::from_u8(*tag_byte),
            [] => None,
          }
        } else {
          None
        }
      }
    }
  }

  /// 从方案 A 集合元数据物理键中快速提取逻辑用户键 (const fn，零回溯零分配)
  #[inline(always)]
  pub const fn decode_meta_user_key(key: &[u8]) -> Option<&[u8]> {
    match key {
      [ns_b, db_b, tag_byte, payload @ ..]
        if *ns_b < VARINT_1B_FIRST_BYTE_LIMIT && *db_b < VARINT_1B_FIRST_BYTE_LIMIT =>
      {
        if *tag_byte == KeyTag::Meta as u8 {
          Some(payload)
        } else {
          None
        }
      }
      _ => {
        let prefix_len = match Self::session_prefix_len_from_slice(key) {
          Some(l) => l,
          None => return None,
        };
        if let Some((_, rest)) = key.split_at_checked(prefix_len) {
          match rest {
            [tag_byte, payload @ ..] if *tag_byte == KeyTag::Meta as u8 => Some(payload),
            _ => None,
          }
        } else {
          None
        }
      }
    }
  }

  /// 从方案 A 物理子键中快速提取 (tag, key_id, version)（const fn，零分配，供紧缩器极速判定）
  #[inline(always)]
  pub const fn decode_subkey_id_version(key: &[u8]) -> Option<(KeyTag, u64, u64)> {
    if key.len() < MIN_SUBKEY_LEN {
      return None;
    }
    match key {
      [
        ns_b,
        db_b,
        tag_byte,
        k0,
        k1,
        k2,
        k3,
        k4,
        k5,
        k6,
        k7,
        v0,
        v1,
        v2,
        v3,
        v4,
        v5,
        v6,
        v7,
        ..,
      ] if *ns_b < VARINT_1B_FIRST_BYTE_LIMIT && *db_b < VARINT_1B_FIRST_BYTE_LIMIT => {
        let tag = match KeyTag::from_u8(*tag_byte) {
          Some(t) if t.is_subkey() => t,
          _ => return None,
        };
        Some((
          tag,
          u64::from_be_bytes([*k0, *k1, *k2, *k3, *k4, *k5, *k6, *k7]),
          u64::from_be_bytes([*v0, *v1, *v2, *v3, *v4, *v5, *v6, *v7]),
        ))
      }
      _ => {
        let prefix_len = match Self::session_prefix_len_from_slice(key) {
          Some(l) => l,
          None => return None,
        };
        let rest = match key.split_at_checked(prefix_len) {
          Some((_, r)) => r,
          None => return None,
        };
        match rest {
          [
            tag_byte,
            k0,
            k1,
            k2,
            k3,
            k4,
            k5,
            k6,
            k7,
            v0,
            v1,
            v2,
            v3,
            v4,
            v5,
            v6,
            v7,
            ..,
          ] => {
            let tag = match KeyTag::from_u8(*tag_byte) {
              Some(t) if t.is_subkey() => t,
              _ => return None,
            };
            Some((
              tag,
              u64::from_be_bytes([*k0, *k1, *k2, *k3, *k4, *k5, *k6, *k7]),
              u64::from_be_bytes([*v0, *v1, *v2, *v3, *v4, *v5, *v6, *v7]),
            ))
          }
          _ => None,
        }
      }
    }
  }
}

/// 解码物理子键后的各分量元组 `(ns, db, tag, key_id, version, payload)`
pub type DecodedSubKey<'a> = (u64, u64, KeyTag, u64, u64, &'a [u8]);

impl NamespaceDbCodec {
  /// 从方案 A 物理子键中一次性解码出 `(ns, db, tag, key_id, version, payload)` (const fn)
  #[inline]
  pub const fn decode_sub_key(key: &[u8]) -> Result<DecodedSubKey<'_>> {
    match key {
      [
        ns_b,
        db_b,
        tag_byte,
        k0,
        k1,
        k2,
        k3,
        k4,
        k5,
        k6,
        k7,
        v0,
        v1,
        v2,
        v3,
        v4,
        v5,
        v6,
        v7,
        subkey_payload @ ..,
      ] if *ns_b < VARINT_1B_FIRST_BYTE_LIMIT && *db_b < VARINT_1B_FIRST_BYTE_LIMIT => {
        let tag = match KeyTag::from_u8(*tag_byte) {
          Some(t) if t.is_subkey() => t,
          Some(t) => return Err(Error::InvalidKeyTag(t as u8)),
          None => return Err(Error::InvalidKeyTag(*tag_byte)),
        };
        let key_id = u64::from_be_bytes([*k0, *k1, *k2, *k3, *k4, *k5, *k6, *k7]);
        let version = u64::from_be_bytes([*v0, *v1, *v2, *v3, *v4, *v5, *v6, *v7]);
        Ok((
          *ns_b as u64,
          *db_b as u64,
          tag,
          key_id,
          version,
          subkey_payload,
        ))
      }
      _ => {
        let (ns, db, tag, payload) = match Self::decode_tagged_key(key) {
          Ok(v) => v,
          Err(e) => return Err(e),
        };
        if !tag.is_subkey() {
          return Err(Error::InvalidKeyTag(tag as u8));
        }
        match payload {
          [
            k0,
            k1,
            k2,
            k3,
            k4,
            k5,
            k6,
            k7,
            v0,
            v1,
            v2,
            v3,
            v4,
            v5,
            v6,
            v7,
            subkey_payload @ ..,
          ] => {
            let key_id = u64::from_be_bytes([*k0, *k1, *k2, *k3, *k4, *k5, *k6, *k7]);
            let version = u64::from_be_bytes([*v0, *v1, *v2, *v3, *v4, *v5, *v6, *v7]);
            Ok((ns, db, tag, key_id, version, subkey_payload))
          }
          _ => Err(Error::BufferTooShort {
            expected: SUBKEY_META_HEADER_LEN,
            actual: payload.len(),
          }),
        }
      }
    }
  }

  /// 从方案 A 集合分块物理子键中一次性解码出 `(ns, db, tag, key_id, version, chunk_id)` (const fn)
  #[inline]
  pub const fn decode_chunk_key(key: &[u8]) -> Result<(u64, u64, KeyTag, u64, u64, u32)> {
    if key.len() < MIN_CHUNK_KEY_LEN {
      return Err(Error::BufferTooShort {
        expected: MIN_CHUNK_KEY_LEN,
        actual: key.len(),
      });
    }
    match key {
      [
        ns_b,
        db_b,
        tag_byte,
        k0,
        k1,
        k2,
        k3,
        k4,
        k5,
        k6,
        k7,
        v0,
        v1,
        v2,
        v3,
        v4,
        v5,
        v6,
        v7,
        c0,
        c1,
        c2,
        c3,
      ] if *ns_b < VARINT_1B_FIRST_BYTE_LIMIT && *db_b < VARINT_1B_FIRST_BYTE_LIMIT => {
        let tag = match KeyTag::from_u8(*tag_byte) {
          Some(t) if t.is_subkey() => t,
          Some(t) => return Err(Error::InvalidKeyTag(t as u8)),
          None => return Err(Error::InvalidKeyTag(*tag_byte)),
        };
        Ok((
          *ns_b as u64,
          *db_b as u64,
          tag,
          u64::from_be_bytes([*k0, *k1, *k2, *k3, *k4, *k5, *k6, *k7]),
          u64::from_be_bytes([*v0, *v1, *v2, *v3, *v4, *v5, *v6, *v7]),
          u32::from_be_bytes([*c0, *c1, *c2, *c3]),
        ))
      }
      _ => {
        let (ns, db, tag, key_id, version, payload) = match Self::decode_sub_key(key) {
          Ok(v) => v,
          Err(e) => return Err(e),
        };
        match payload {
          [c0, c1, c2, c3] => Ok((
            ns,
            db,
            tag,
            key_id,
            version,
            u32::from_be_bytes([*c0, *c1, *c2, *c3]),
          )),
          _ => Err(Error::BufferTooShort {
            expected: CHUNK_ID_LEN,
            actual: payload.len(),
          }),
        }
      }
    }
  }
}