wedb_embed 0.1.2

use core::str::from_utf8;

use crate::{
  db::WeDb,
  error::{Error, Result},
  key::Key,
  key_composer::{
    KeyComposer,
    ns::{
      DEFAULT_NAMESPACE, NS_NAME_PREFIX, NS_NEXT_ID_KEY, catalog_db_key, catalog_ns_prefix,
      is_default_namespace, ns_id_key, ns_name_key,
    },
    oppv::{decode_oppv_u64, encode_oppv_u64, encode_oppv_u64_fixed},
    tag::ScopeModeTag,
  },
};

/// 命名空间句柄对象（零堆分配，内部自动完成物理键编排与作用域隔离）
#[derive(Clone, Copy)]
pub struct Namespace<'a> {
  pub db: &'a WeDb,
  pub kc: KeyComposer<'a>,
}

impl WeDb {
  /// 根据租户名称构造携带持久化自增 ns_id 的 KeyComposer
  #[inline]
  pub fn kc_for_ns<'a>(&self, name: &'a str) -> KeyComposer<'a> {
    if is_default_namespace(name) {
      KeyComposer::new("default")
    } else {
      let ns_id = self.ns_id(name).unwrap_or(0);
      KeyComposer::new_named(name, ns_id, 0)
    }
  }

  /// 根据租户名称与 db 索引构造携带持久化自增 ns_id 的 KeyComposer
  #[inline]
  pub fn kc_for_ns_db<'a>(&self, name: &'a str, db: u64) -> KeyComposer<'a> {
    if is_default_namespace(name) {
      KeyComposer::new_db(db)
    } else {
      let ns_id = self.ns_id(name).unwrap_or(0);
      KeyComposer::new_named(name, ns_id, db)
    }
  }

  /// 获取指定租户命名空间的操作句柄
  #[inline]
  pub fn namespace<'a>(&'a self, name: &'a str) -> Namespace<'a> {
    let kc = self.kc_for_ns(name);
    Namespace { db: self, kc }
  }

  /// 获取指定数据库编号的操作句柄（对标 SELECT db，支持纯数值 u64）
  #[inline]
  pub fn select_db(&self, db_index: u64) -> Result<Namespace<'_>> {
    if db_index > 0 {
      self.activate_db("default", db_index)?;
    }
    Ok(Namespace {
      db: self,
      kc: KeyComposer::new_db(db_index),
    })
  }

  /// 获取默认命名空间操作句柄
  #[inline]
  pub fn default_ns(&self) -> Namespace<'_> {
    self.namespace("default")
  }

  /// 纯流式零堆分配迭代所有已激活的命名空间（O(1) 辅助内存）
  #[inline]
  pub fn iter(&self) -> Namespaces {
    Namespaces {
      iter: self.meta_ns.prefix(NS_NAME_PREFIX),
      emitted_default: false,
    }
  }

  /// 在 Catalog 紧凑目录中持久化激活指定命名空间的数据库（采用 OPPV 保序变长整型编码）
  pub fn activate_db(&self, ns: &str, db_idx: u64) -> Result<()> {
    let key = catalog_db_key(ns, db_idx);
    self.meta_ns.insert(key, b"")?;
    Ok(())
  }

  /// 获取或创建租户对应的数字 ID（无锁并发内存缓存 + 双重检查持久化互斥锁，0 为 default）
  pub fn ns_id(&self, name: &str) -> Result<u64> {
    if is_default_namespace(name) {
      return Ok(0);
    }
    // 1. 无锁快速读取缓存 (Lock-free Fast Path ~5ns)
    let pin = self.ns_cache.pin();
    if let Some(&id) = pin.get(name) {
      return Ok(id);
    }

    // 2. 慢路径：并发互斥分配锁 (parking_lot Adaptive Spin + Futex Mutex)
    let _guard = self.ns_lock.lock();

    // 2.1 再次检查缓存（可能已被其他并发线程分配）
    let pin = self.ns_cache.pin();
    if let Some(&id) = pin.get(name) {
      return Ok(id);
    }

    let key = ns_name_key(name);

    // 2.2 检查磁盘是否已存在该命名空间映射
    if let Some(val) = self.meta_ns.get(&key)?
      && let Some((id, _)) = decode_oppv_u64(&val)
    {
      pin.insert(hipstr::HipStr::from(name), id);
      return Ok(id);
    }

    // 2.3 分配全新的全局自增 ID
    let current_id = match self.meta_ns.get(NS_NEXT_ID_KEY)? {
      Some(val) => decode_oppv_u64(&val).map(|(v, _)| v).unwrap_or(1),
      None => 1,
    };
    let new_id = current_id;
    let next_val = current_id + 1;

    let mut batch = self.db.batch();
    let mut val_buf = [0u8; 9];
    let val_len = encode_oppv_u64_fixed(new_id, &mut val_buf);
    batch.insert(&self.meta_ns, &key, &val_buf[..val_len]);

    let id_key = ns_id_key(new_id);
    batch.insert(&self.meta_ns, id_key, name.as_bytes());

    let mut next_val_buf = [0u8; 9];
    let next_len = encode_oppv_u64_fixed(next_val, &mut next_val_buf);
    batch.insert(&self.meta_ns, NS_NEXT_ID_KEY, &next_val_buf[..next_len]);

    batch.commit()?;
    pin.insert(hipstr::HipStr::from(name), new_id);
    Ok(new_id)
  }

  /// 查询租户名称对应的数字 ID
  pub fn query_ns_id(&self, name: &str) -> Result<Option<u64>> {
    if is_default_namespace(name) {
      return Ok(Some(0));
    }
    let pin = self.ns_cache.pin();
    if let Some(&id) = pin.get(name) {
      return Ok(Some(id));
    }
    let key = ns_name_key(name);
    if let Some(val) = self.meta_ns.get(&key)?
      && let Some((id, _)) = decode_oppv_u64(&val)
    {
      pin.insert(hipstr::HipStr::from(name), id);
      return Ok(Some(id));
    }
    Ok(None)
  }

  /// 根据租户数字 ID 反查租户名称
  pub fn ns_name(&self, ns_id: u64) -> Result<Option<String>> {
    if ns_id == 0 {
      return Ok(Some("default".to_string()));
    }
    let id_key = ns_id_key(ns_id);
    if let Some(val) = self.meta_ns.get(&id_key)? {
      return Ok(String::from_utf8(val.to_vec()).ok());
    }
    Ok(None)
  }

  /// O(1) 租户瞬间重命名（仅需原子更新元数据映射，底层业务键零重写）
  pub fn rename_namespace(&self, old_name: &str, new_name: &str) -> Result<()> {
    if is_default_namespace(old_name) || is_default_namespace(new_name) {
      return Err(Error::invalid_data("cannot rename default namespace"));
    }
    let _guard = self.ns_lock.lock();
    let ns_id = match self.query_ns_id(old_name)? {
      Some(id) => id,
      None => return Err(Error::invalid_data("namespace not found")),
    };

    if self.query_ns_id(new_name)?.is_some() {
      return Err(Error::invalid_data("target namespace already exists"));
    }

    let mut batch = self.db.batch();
    // 删除旧 name 映射
    let old_key = ns_name_key(old_name);
    batch.remove(&self.meta_ns, old_key);

    // 插入新 name 映射
    let new_key = ns_name_key(new_name);
    let mut val_buf = [0u8; 9];
    let val_len = encode_oppv_u64_fixed(ns_id, &mut val_buf);
    batch.insert(&self.meta_ns, &new_key, &val_buf[..val_len]);

    // 更新 id -> name 映射
    let id_key = ns_id_key(ns_id);
    batch.insert(&self.meta_ns, id_key, new_name.as_bytes());

    // 同步原子迁移该租户下的 Catalog 目录索引（若存在多库激活记录）
    let cat_old_prefix = catalog_ns_prefix(old_name);
    for item in self.meta_ns.prefix(&cat_old_prefix) {
      let k = item.key()?;
      if !k.starts_with(&cat_old_prefix) {
        break;
      }
      batch.remove(&self.meta_ns, &*k);
      let remain = &k[cat_old_prefix.len()..];
      let mut new_cat_k = catalog_ns_prefix(new_name);
      new_cat_k.extend_from_slice(remain);
      batch.insert(&self.meta_ns, new_cat_k, b"");
    }

    batch.commit()?;
    let pin = self.ns_cache.pin();
    pin.remove(old_name);
    pin.insert(hipstr::HipStr::from(new_name), ns_id);
    Ok(())
  }

  /// 按通配符模式匹配查询默认命名空间下的所有用户 Key
  #[inline]
  pub fn keys<P: AsRef<[u8]>>(&self, pattern: P) -> Result<Vec<Vec<u8>>> {
    Key::keys(self, pattern)
  }

  /// 统计默认命名空间下的 Key 总数
  #[inline]
  pub fn key_count(&self) -> Result<usize> {
    Key::key_count(self)
  }
}

#[inline]
fn clear_ks_prefix(
  ks: &fjall::Keyspace,
  prefix: &[u8],
  batch: &mut fjall::OwnedWriteBatch,
  count: &mut u64,
) -> Result<()> {
  for item in ks.prefix(prefix) {
    let k = item.key()?;
    if !k.starts_with(prefix) {
      break;
    }
    batch.remove(ks, k);
    *count += 1;
  }
  Ok(())
}

impl<'a> Namespace<'a> {
  /// 命名空间名称
  #[inline]
  pub fn name(&self) -> &str {
    self.kc.ns()
  }

  /// 是否为默认命名空间
  #[inline]
  pub fn is_default(&self) -> bool {
    self.kc.is_default()
  }

  /// 获取当前租户对应的数字 ID（OPPV 编码）
  #[inline]
  pub fn id(&self) -> Result<u64> {
    self.db.ns_id(self.name())
  }

  /// O(1) 重命名当前命名空间（仅需原子更新元数据映射，底层业务键零重写）
  #[inline]
  pub fn rename(&self, new_name: &str) -> Result<()> {
    self.db.rename_namespace(self.name(), new_name)
  }

  /// 切换到当前命名空间下的指定数据库（对标 SELECT db）
  #[inline]
  pub fn select_db(&self, db_index: u64) -> Result<Namespace<'a>> {
    if db_index > 0 {
      self.db.activate_db(self.kc.ns(), db_index)?;
    }
    Ok(Namespace {
      db: self.db,
      kc: KeyComposer::new_named(self.kc.ns(), self.kc.ns_id(), db_index),
    })
  }

  /// 获取当前命名空间对应的 DB 索引（纯数字 u64）
  #[inline]
  pub fn db_index(&self) -> u64 {
    self.kc.db()
  }

  /// 按通配符模式匹配查询当前命名空间下的所有用户 Key
  #[inline]
  pub fn keys<P: AsRef<[u8]>>(&self, pattern: P) -> Result<Vec<Vec<u8>>> {
    Key::keys(self, pattern)
  }

  /// 统计当前命名空间下的 Key 总数
  #[inline]
  pub fn key_count(&self) -> Result<usize> {
    Key::key_count(self)
  }

  /// 批量删除当前命名空间下的指定键
  #[inline]
  pub fn del<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<usize> {
    Key::del(self, keys)
  }

  /// 批量检查当前命名空间下的指定键是否存在
  #[inline]
  pub fn exists<K: AsRef<[u8]>>(&self, keys: &[K]) -> Result<usize> {
    Key::exists(self, keys)
  }

  /// 清空当前命名空间下的所有业务数据与复合结构（物理批量清除）
  pub fn clear(&self) -> Result<u64> {
    let mut count = 0u64;
    let mut batch = self.db.db.batch();

    if self.is_default() {
      // 1. 清空默认 db 0 数据与元数据
      for item in self.db.data.iter() {
        batch.remove(&self.db.data, item.key()?);
        count += 1;
      }
      for item in self.db.meta.iter() {
        batch.remove(&self.db.meta, item.key()?);
        count += 1;
      }

      // 2. 清理默认租户下所有 db > 0 的业务数据 (ScopeModeTag::DefaultDb: \x00\x01)
      let prefix_default_dbs = ScopeModeTag::DefaultDb.as_prefix_slice();
      clear_ks_prefix(&self.db.data_ns, prefix_default_dbs, &mut batch, &mut count)?;
      clear_ks_prefix(&self.db.meta_ns, prefix_default_dbs, &mut batch, &mut count)?;

      // 3. 清理默认租户下的 Catalog 目录索引
      let cat_prefix = catalog_ns_prefix(DEFAULT_NAMESPACE);
      let mut _dummy = 0;
      clear_ks_prefix(&self.db.meta_ns, &cat_prefix, &mut batch, &mut _dummy)?;
    } else if self.kc.db() > 0 {
      // 清空当前特定 db
      let prefix = self.kc.namespace_prefix();
      clear_ks_prefix(&self.db.data_ns, &prefix, &mut batch, &mut count)?;
      clear_ks_prefix(&self.db.meta_ns, &prefix, &mut batch, &mut count)?;

      // 清理 Catalog 目录中该 db 的索引
      let cat_key = catalog_db_key(self.name(), self.kc.db());
      batch.remove(&self.db.meta_ns, cat_key);
    } else {
      // 清空整个租户（包含 db 0 及该租户下所有 db > 0 的业务数据）
      // 1. 清理单库 (db 0): \x00\x02[oppv(ns_id)]:
      let prefix_single = self.kc.namespace_prefix();
      clear_ks_prefix(&self.db.data_ns, &prefix_single, &mut batch, &mut count)?;
      clear_ks_prefix(&self.db.meta_ns, &prefix_single, &mut batch, &mut count)?;

      // 2. 清理多库 (db > 0): \x00\x03[oppv(ns_id)]
      let mut prefix_multi = Vec::with_capacity(2 + 9);
      prefix_multi.extend_from_slice(ScopeModeTag::TenantDb.as_prefix_slice());
      encode_oppv_u64(self.kc.ns_id(), &mut prefix_multi);
      clear_ks_prefix(&self.db.data_ns, &prefix_multi, &mut batch, &mut count)?;
      clear_ks_prefix(&self.db.meta_ns, &prefix_multi, &mut batch, &mut count)?;

      // 3. 清理该租户下的 Catalog 目录索引
      let cat_prefix = catalog_ns_prefix(self.name());
      let mut _dummy = 0;
      clear_ks_prefix(&self.db.meta_ns, &cat_prefix, &mut batch, &mut _dummy)?;
    }

    batch.commit()?;
    Ok(count)
  }

  /// 清空当前选中的数据库（FLUSHDB 语义：仅清除当前 db_index 的数据）
  pub fn flushdb(&self) -> Result<u64> {
    let mut count = 0u64;
    let mut batch = self.db.db.batch();

    if self.is_default() {
      for item in self.db.data.iter() {
        batch.remove(&self.db.data, item.key()?);
        count += 1;
      }
      for item in self.db.meta.iter() {
        batch.remove(&self.db.meta, item.key()?);
        count += 1;
      }
    } else {
      let prefix = self.kc.namespace_prefix();
      clear_ks_prefix(&self.db.data_ns, &prefix, &mut batch, &mut count)?;
      clear_ks_prefix(&self.db.meta_ns, &prefix, &mut batch, &mut count)?;
      if self.kc.db() > 0 {
        let cat_key = catalog_db_key(self.name(), self.kc.db());
        batch.remove(&self.db.meta_ns, cat_key);
      }
    }

    batch.commit()?;
    Ok(count)
  }

  /// 清空当前命名空间下的所有数据库（FLUSHALL 语义：清除该租户下所有 db 的数据）
  #[inline]
  pub fn flushall(&self) -> Result<u64> {
    if self.is_default() || self.kc.db() == 0 {
      self.clear()
    } else {
      Namespace {
        db: self.db,
        kc: KeyComposer::new_named(self.kc.ns(), self.kc.ns_id(), 0),
      }
      .clear()
    }
  }

  /// 纯流式零堆分配升序迭代当前命名空间下已激活的数据库索引列表（支持 u64 与 OPPV 编码）
  #[inline]
  pub fn iter(&self) -> Dbs {
    let cat_prefix = catalog_ns_prefix(self.name());
    let iter = self.db.meta_ns.prefix(&cat_prefix);
    Dbs {
      prefix: cat_prefix,
      iter,
      emitted_zero: false,
    }
  }
}

/// 纯流式零堆分配命名空间名称迭代器（O(1) 内存，零重复 seek）
pub struct Namespaces {
  iter: fjall::Iter,
  emitted_default: bool,
}

impl Iterator for Namespaces {
  type Item = String;

  fn next(&mut self) -> Option<Self::Item> {
    if !self.emitted_default {
      self.emitted_default = true;
      return Some("default".to_string());
    }

    let prefix = NS_NAME_PREFIX;
    for item in self.iter.by_ref() {
      let k = item.key().ok()?;
      if !k.starts_with(prefix) {
        return None;
      }
      if let Ok(ns_name) = from_utf8(&k[prefix.len()..]) {
        return Some(ns_name.to_string());
      }
    }
    None
  }
}

impl IntoIterator for &WeDb {
  type Item = String;
  type IntoIter = Namespaces;

  #[inline]
  fn into_iter(self) -> Self::IntoIter {
    self.iter()
  }
}

impl IntoIterator for WeDb {
  type Item = String;
  type IntoIter = Namespaces;

  #[inline]
  fn into_iter(self) -> Self::IntoIter {
    self.iter()
  }
}

/// 纯流式零堆分配升序数据库索引迭代器（直接流式读取 OPPV 变长保序 Catalog 紧凑目录）
pub struct Dbs {
  prefix: Vec<u8>,
  iter: fjall::Iter,
  emitted_zero: bool,
}

impl Iterator for Dbs {
  type Item = u64;

  fn next(&mut self) -> Option<Self::Item> {
    if !self.emitted_zero {
      self.emitted_zero = true;
      return Some(0);
    }

    for item in self.iter.by_ref() {
      let k = item.key().ok()?;
      if !k.starts_with(&self.prefix) {
        return None;
      }
      let remain = &k[self.prefix.len()..];
      if let Some((db_idx, _consumed)) = decode_oppv_u64(remain)
        && db_idx > 0
      {
        return Some(db_idx);
      }
    }
    None
  }
}

impl IntoIterator for &Namespace<'_> {
  type Item = u64;
  type IntoIter = Dbs;

  #[inline]
  fn into_iter(self) -> Self::IntoIter {
    self.iter()
  }
}

impl IntoIterator for Namespace<'_> {
  type Item = u64;
  type IntoIter = Dbs;

  #[inline]
  fn into_iter(self) -> Self::IntoIter {
    self.iter()
  }
}