wedb_embed 0.1.2

use std::{thread, time::Duration};

use aok::{OK, Void};
use wedb_embed::{
  WeDb,
  api::{
    bitmap::{compose_bitmap_meta_key, compose_bitmap_segment},
    bloom::{
      compose_bloom_item, compose_bloom_meta_key, compose_cuckoo_meta_key, compose_cuckoo_page,
    },
    hash::{compose_hash_key, compose_hash_meta_key, compose_hash_prefix},
    hll::compose_hll_meta_key,
    json::compose_json_meta_key,
    list::{compose_list_item, compose_list_meta_key},
    set::{compose_set_key, compose_set_meta_key, compose_set_prefix},
    sortedint::{compose_si_key, compose_si_meta_key},
    stream::{
      compose_stream_consumer_meta, compose_stream_group_meta, compose_stream_item,
      compose_stream_meta_key, compose_stream_pel_item,
    },
    string::{Set as StringSet, compose_string_key, compose_string_key_bytes},
    tdigest::compose_tdigest_meta_key,
    timeseries::{compose_ts_item, compose_ts_meta_key},
    zset::{compose_zset_key, compose_zset_meta_key, compose_zset_prefix, compose_zset_score_key},
  },
  key_composer::{
    KeyComposer, SubkeyComposer, compose_slot_key_prefix, compose_slot_key_upper_bound,
    is_default_namespace, is_default_namespace_bytes,
  },
  prelude::*,
};
fn to_hex(bytes: &[u8]) -> String {
  bytes.iter().map(|b| format!("{b:02x}")).collect()
}

#[test]
fn test_namespace_and_select_db_scope_encoding_golden() -> Void {
  // 1. 默认命名空间 (Mode 0: 空前缀)
  let kc_def = KeyComposer::new("default");
  assert!(kc_def.is_default());
  assert_eq!(kc_def.scope_prefix_len(), 0);
  assert_eq!(kc_def.namespace_prefix(), b"");
  assert_eq!(&*compose_string_key(&kc_def, b"mykey"), b"\x00mykey");

  // 2. 默认租户多库模式 (Mode 1: \x00\x01[oppv(db)])
  let kc_db1 = KeyComposer::new_db(1);
  assert!(!kc_db1.is_default());
  assert_eq!(kc_db1.ns_id(), 0);
  assert_eq!(kc_db1.db(), 1);
  assert_eq!(kc_db1.namespace_prefix(), b"\x00\x01\x01");
  assert_eq!(to_hex(&kc_db1.namespace_prefix()), "000101");
  assert_eq!(
    &*compose_string_key(&kc_db1, b"mykey"),
    b"\x00\x01\x01\x00mykey"
  );

  // 3. 自定义租户单库模式 (Mode 2: \x00\x02[oppv(ns_id)])
  let kc_t1 = KeyComposer::new_named("tenant_a", 1, 0);
  assert!(!kc_t1.is_default());
  assert_eq!(kc_t1.ns_id(), 1);
  assert_eq!(kc_t1.db(), 0);
  assert_eq!(kc_t1.namespace_prefix(), b"\x00\x02\x01");
  assert_eq!(to_hex(&kc_t1.namespace_prefix()), "000201");
  assert_eq!(
    &*compose_string_key(&kc_t1, b"mykey"),
    b"\x00\x02\x01\x00mykey"
  );

  // 4. 自定义租户多库模式 (Mode 3: \x00\x03[oppv(ns_id)][oppv(db)])
  let kc_t1_db2 = KeyComposer::new_named("tenant_a", 1, 2);
  assert_eq!(kc_t1_db2.namespace_prefix(), b"\x00\x03\x01\x02");
  assert_eq!(to_hex(&kc_t1_db2.namespace_prefix()), "00030102");
  assert_eq!(
    &*compose_string_key(&kc_t1_db2, b"mykey"),
    b"\x00\x03\x01\x02\x00mykey"
  );

  // 5. 复合结构 Subkey 编码验证
  let hash_k = compose_hash_key(&kc_t1, b"user", b"email");
  assert_eq!(kc_t1.extract_user_key(&hash_k), Some(b"user".as_slice()));

  Ok(())
}

#[test]
fn test_key_composer_isolation_all_15_types() -> Void {
  let kc_def = KeyComposer::new("default");
  let kc_t1 = KeyComposer::new_named("tenant_1", 1, 0);
  let kc_t2 = KeyComposer::new_named("tenant_2", 2, 0);

  // 1. String (Raw key)
  assert_ne!(
    compose_string_key(&kc_def, b"k"),
    compose_string_key(&kc_t1, b"k")
  );
  assert_ne!(
    compose_string_key(&kc_t1, b"k"),
    compose_string_key(&kc_t2, b"k")
  );

  // 2. Hash
  assert_ne!(
    compose_hash_meta_key(&kc_def, b"h"),
    compose_hash_meta_key(&kc_t1, b"h")
  );
  assert_ne!(
    compose_hash_key(&kc_t1, b"h", b"f"),
    compose_hash_key(&kc_t2, b"h", b"f")
  );
  assert_ne!(
    compose_hash_prefix(&kc_t1, b"h"),
    compose_hash_prefix(&kc_t2, b"h")
  );

  // 3. List
  assert_ne!(
    compose_list_meta_key(&kc_def, b"l"),
    compose_list_meta_key(&kc_t1, b"l")
  );
  assert_ne!(
    compose_list_item(&kc_t1, b"l", 1),
    compose_list_item(&kc_t2, b"l", 1)
  );

  // 4. Set
  assert_ne!(
    compose_set_meta_key(&kc_def, b"s"),
    compose_set_meta_key(&kc_t1, b"s")
  );
  assert_ne!(
    compose_set_key(&kc_t1, b"s", b"m"),
    compose_set_key(&kc_t2, b"s", b"m")
  );

  // 5. ZSet
  assert_ne!(
    compose_zset_meta_key(&kc_def, b"z"),
    compose_zset_meta_key(&kc_t1, b"z")
  );
  assert_ne!(
    compose_zset_key(&kc_t1, b"z", b"m"),
    compose_zset_key(&kc_t2, b"z", b"m")
  );
  assert_ne!(
    compose_zset_score_key(&kc_t1, b"z", 1.5, b"m"),
    compose_zset_score_key(&kc_t2, b"z", 1.5, b"m")
  );

  // 6. Bitmap
  assert_ne!(
    compose_bitmap_meta_key(&kc_def, b"b"),
    compose_bitmap_meta_key(&kc_t1, b"b")
  );
  assert_ne!(
    compose_bitmap_segment(&kc_t1, b"b", 0),
    compose_bitmap_segment(&kc_t2, b"b", 0)
  );

  // 7. Bloom Filter
  assert_ne!(
    compose_bloom_meta_key(&kc_def, b"bf"),
    compose_bloom_meta_key(&kc_t1, b"bf")
  );
  assert_ne!(
    compose_bloom_item(&kc_t1, b"bf", 1),
    compose_bloom_item(&kc_t2, b"bf", 1)
  );

  // 8. Cuckoo Filter
  assert_ne!(
    compose_cuckoo_meta_key(&kc_def, b"cf"),
    compose_cuckoo_meta_key(&kc_t1, b"cf")
  );
  assert_ne!(
    compose_cuckoo_page(&kc_t1, b"cf", 0, 1),
    compose_cuckoo_page(&kc_t2, b"cf", 0, 1)
  );

  // 9. HyperLogLog
  assert_ne!(
    compose_hll_meta_key(&kc_def, b"hll"),
    compose_hll_meta_key(&kc_t1, b"hll")
  );
  assert_ne!(
    compose_hll_meta_key(&kc_t1, b"hll"),
    compose_hll_meta_key(&kc_t2, b"hll")
  );

  // 10. JSON
  assert_ne!(
    compose_json_meta_key(&kc_def, b"j"),
    compose_json_meta_key(&kc_t1, b"j")
  );

  // 11. SortedInt
  assert_ne!(
    compose_si_meta_key(&kc_def, b"si"),
    compose_si_meta_key(&kc_t1, b"si")
  );
  assert_ne!(
    compose_si_key(&kc_t1, b"si", 42),
    compose_si_key(&kc_t2, b"si", 42)
  );

  // 13. Stream
  assert_ne!(
    compose_stream_meta_key(&kc_def, b"str"),
    compose_stream_meta_key(&kc_t1, b"str")
  );
  assert_ne!(
    compose_stream_item(&kc_t1, b"str", 100, 1),
    compose_stream_item(&kc_t2, b"str", 100, 1)
  );
  assert_ne!(
    compose_stream_group_meta(&kc_t1, b"str", b"g1"),
    compose_stream_group_meta(&kc_t2, b"str", b"g1")
  );
  assert_ne!(
    compose_stream_consumer_meta(&kc_t1, b"str", b"g1", b"c1"),
    compose_stream_consumer_meta(&kc_t2, b"str", b"g1", b"c1")
  );
  assert_ne!(
    compose_stream_pel_item(&kc_t1, b"str", b"g1", 100, 1),
    compose_stream_pel_item(&kc_t2, b"str", b"g1", 100, 1)
  );

  // 14. TDigest
  assert_ne!(
    compose_tdigest_meta_key(&kc_def, b"td"),
    compose_tdigest_meta_key(&kc_t1, b"td")
  );

  // 15. TimeSeries
  assert_ne!(
    compose_ts_meta_key(&kc_def, b"ts"),
    compose_ts_meta_key(&kc_t1, b"ts")
  );
  assert_ne!(
    compose_ts_item(&kc_t1, b"ts", 1000),
    compose_ts_item(&kc_t2, b"ts", 1000)
  );

  Ok(())
}

#[test]
fn test_user_key_extraction_and_in_ns_all_types() -> Void {
  let kc_t = KeyComposer::new_named("app_team", 42, 0);
  let key = b"my:unique:user_key";

  // 验证所有 15 种数据结构的 meta / data / subkey 的 extract_user_key 均能正确提取出原 key
  assert_eq!(
    kc_t.extract_user_key(&compose_string_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_hash_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_list_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_set_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_zset_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_bitmap_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_bloom_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_cuckoo_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_hll_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_json_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_si_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_stream_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_tdigest_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_ts_meta_key(&kc_t, key)),
    Some(key.as_slice())
  );

  // 验证子键提取
  assert_eq!(
    kc_t.extract_user_key(&compose_bitmap_segment(&kc_t, key, 0)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_si_key(&kc_t, key, 1234)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_ts_item(&kc_t, key, 9999)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_bloom_item(&kc_t, key, 1)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_cuckoo_page(&kc_t, key, 0, 1)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_stream_item(&kc_t, key, 100, 2)),
    Some(key.as_slice())
  );
  assert_eq!(
    kc_t.extract_user_key(&compose_stream_group_meta(&kc_t, key, b"grp")),
    Some(key.as_slice())
  );

  // 验证默认命名空间下首字节为 0x70~0x7F 的常见用户键（password, user, test 等）绝不会被错误拦截
  let kc_def = KeyComposer::new("default");
  assert_eq!(
    kc_def.extract_user_key(&compose_string_key(&kc_def, b"password")),
    Some(b"password".as_slice())
  );
  assert_eq!(
    kc_def.extract_user_key(&compose_string_key(&kc_def, b"user")),
    Some(b"user".as_slice())
  );
  assert_eq!(
    kc_def.extract_user_key(&compose_string_key(&kc_def, b"token")),
    Some(b"token".as_slice())
  );
  assert_eq!(
    kc_def.extract_user_key(&compose_string_key(&kc_def, b"query")),
    Some(b"query".as_slice())
  );
  assert_eq!(
    kc_def.extract_user_key(&compose_string_key(&kc_def, b"zone")),
    Some(b"zone".as_slice())
  );
  assert!(kc_def.is_key_in_ns(&compose_string_key(&kc_def, b"password")));
  assert!(kc_def.is_key_in_ns(&compose_string_key(&kc_def, b"user")));

  // 验证租户对系统管理域（\x00\x70 ~ \x00\x7F）与非法前缀被严格隔离与过滤
  assert_eq!(kc_t.extract_user_key(b"\x00\x70:ns:name:foo"), None);
  assert_eq!(kc_t.extract_user_key(b"\x00\x70:ns:token:xxx"), None);
  assert_eq!(kc_t.extract_user_key(b"\x00\x71:tenant1:db:\x01"), None);
  assert_eq!(kc_t.extract_user_key(b"\x00\x72:ttl:123456"), None);
  assert_eq!(kc_t.extract_user_key(b"\x00\x73:raft:log:1"), None);
  assert!(!kc_t.is_key_in_ns(b"\x00\x70:ns:token:xxx"));
  assert!(!kc_t.is_key_in_ns(b"\x00\x71:tenant1:db:\x01"));

  Ok(())
}

#[test]
fn test_transform_key_across_namespaces() -> Void {
  let kc_def = KeyComposer::new("default");
  let kc_db1 = KeyComposer::new_db(1);
  let kc_db2 = KeyComposer::new_db(2);
  let kc_t1 = KeyComposer::new_named("tenant1", 1, 0);

  // 1. 默认空间 -> db1 (纯 OPPV 紧凑模式 1)
  let raw_def = b"\x00my_string_key";
  let transformed_db1 = kc_def
    .transform_key_to_target_bytes(raw_def, &kc_db1)
    .unwrap();
  assert_eq!(transformed_db1, b"\x00\x01\x01\x00my_string_key");

  let hash_def = b"\x01:my_hash";
  let transformed_h_db1 = kc_def
    .transform_key_to_target_bytes(hash_def, &kc_db1)
    .unwrap();
  assert_eq!(transformed_h_db1, b"\x00\x01\x01\x01:my_hash");

  // 2. db1 -> 默认空间
  assert_eq!(
    kc_db1.transform_key_to_target_bytes(&transformed_db1, &kc_def),
    Some(b"\x00my_string_key".to_vec())
  );
  assert_eq!(
    kc_db1.transform_key_to_target_bytes(&transformed_h_db1, &kc_def),
    Some(b"\x01:my_hash".to_vec())
  );

  // 3. db1 -> db2
  assert_eq!(
    kc_db1.transform_key_to_target_bytes(&transformed_db1, &kc_db2),
    Some(b"\x00\x01\x02\x00my_string_key".to_vec())
  );

  // 4. db1 -> tenant1 (模式 2 纯数字 ns_id 前缀)
  let expected_t1 = b"\x00\x02\x01\x00my_string_key";
  assert_eq!(
    kc_db1.transform_key_to_target_bytes(&transformed_db1, &kc_t1),
    Some(expected_t1.to_vec())
  );

  Ok(())
}

#[test]
fn test_select_db_and_namespace_mappings_suite() -> Void {
  let prefix = compose_slot_key_prefix(b"ns1", 100);
  assert!(!prefix.is_empty());
  let upper = compose_slot_key_upper_bound(b"ns1", 100);
  assert_eq!(upper, compose_slot_key_prefix(b"ns1", 101));

  Ok(())
}

#[test]
fn test_default_namespace_and_legal_checks_extended() -> Void {
  // 默认命名空间判定
  assert!(is_default_namespace("default"));
  assert!(is_default_namespace("0"));
  assert!(is_default_namespace("db0"));
  assert!(is_default_namespace(""));
  assert!(!is_default_namespace("db1"));
  assert!(!is_default_namespace("tenant_1"));

  assert!(is_default_namespace_bytes(b"default"));
  assert!(is_default_namespace_bytes(b"0"));
  assert!(is_default_namespace_bytes(b"db0"));
  assert!(is_default_namespace_bytes(b""));
  assert!(!is_default_namespace_bytes(b"db1"));
  assert!(!is_default_namespace_bytes(b"tenant_1"));

  // 命名空间字符合法性判定
  Ok(())
}

#[test]
fn test_binary_safe_key_transformation_and_buffer_reuse() -> Void {
  let kc_def = KeyComposer::new("default");
  let kc_t = KeyComposer::new_named("t1", 99, 0);

  // 任意二进制 Key（包含非 UTF-8 字节 0xFF, 0xFE, 0x00）
  let binary_key = b"\xff\xfe\x00custom_bin_key";
  let raw_bin = compose_string_key_bytes(&kc_def, binary_key);
  let trans_bytes = kc_def
    .transform_key_to_target_bytes(&raw_bin, &kc_t)
    .expect("transform failed");
  assert!(kc_t.is_key_in_ns(&trans_bytes));
  let extracted = kc_t.extract_user_key(&trans_bytes).expect("extract failed");
  assert_eq!(extracted, binary_key);

  // SubkeyComposer 原地复用零堆分配测试
  let hash_p = compose_hash_prefix(&kc_t, binary_key);
  let mut composer = SubkeyComposer::from_slice(&hash_p);
  let k1 = composer.compose_sub(b"field1");
  assert_eq!(kc_t.extract_user_key(k1), Some(binary_key.as_slice()));
  let k2 = composer.compose_sub(b"field2");
  assert_eq!(kc_t.extract_user_key(k2), Some(binary_key.as_slice()));

  Ok(())
}

#[test]
fn test_wedb_namespace_object_and_traits_lifecycle() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  // 1. 获取命名空间句柄
  let ns_apple = db.namespace("tenant_apple");
  let ns_banana = db.namespace("tenant_banana");
  let ns_def = db.default_ns();

  assert_eq!(ns_apple.name(), "tenant_apple");
  assert!(!ns_apple.is_default());
  assert_eq!(ns_def.name(), "default");
  assert!(ns_def.is_default());

  // 2. KvOps (String 接口) 与多租户完全隔离
  ns_apple.set("device", "macbook", &[])?;
  ns_banana.set("device", "thinkpad", &[])?;
  ns_def.set("device", "server_pc", &[])?;

  assert_eq!(ns_apple.get("device")?, Some(b"macbook".to_vec()));
  assert_eq!(ns_banana.get("device")?, Some(b"thinkpad".to_vec()));
  assert_eq!(ns_def.get("device")?, Some(b"server_pc".to_vec()));

  // INCR / DECR
  assert_eq!(ns_apple.incr("counter")?, 1);
  assert_eq!(ns_apple.incrby("counter", 5)?, 6);
  assert_eq!(ns_apple.decr("counter")?, 5);

  // MGET / MSET
  ns_apple.mset(&[("k1", "v1"), ("k2", "v2")])?;
  let mget_res = ns_apple.mget(&["k1", "k2", "k3"])?;
  assert_eq!(mget_res[0], Some(b"v1".to_vec()));
  assert_eq!(mget_res[1], Some(b"v2".to_vec()));
  assert_eq!(mget_res[2], None);

  // 3. HashOps (哈希结构接口)
  ns_apple.hset("user:10", &[("name", "steve"), ("role", "admin")])?;
  assert_eq!(ns_apple.hget("user:10", "name")?, Some(b"steve".to_vec()));
  assert_eq!(ns_apple.hlen("user:10")?, 2);
  assert!(ns_apple.hexists("user:10", "role")?);
  assert_eq!(ns_banana.hget("user:10", "name")?, None);

  // 4. ListOps (列表结构接口)
  ns_apple.rpush("queue", &["task1", "task2", "task3"])?;
  assert_eq!(ns_apple.llen("queue")?, 3);
  assert_eq!(ns_apple.lpop("queue", 1)?, vec![b"task1".to_vec()]);

  // 5. SetOps (集合结构接口)
  ns_apple.sadd("tags", &["rust", "database", "kvrocks"])?;
  assert_eq!(ns_apple.scard("tags")?, 3);
  assert!(ns_apple.sismember("tags", "rust")?);
  assert!(!ns_banana.sismember("tags", "rust")?);

  // 6. ZSetOps (有序集合结构接口)
  ns_apple.zadd(
    "ranks",
    &[(100.0, "alice"), (95.5, "bob"), (80.0, "charlie")],
    [],
  )?;
  assert_eq!(ns_apple.zcard("ranks")?, 3);
  assert_eq!(ns_apple.zscore("ranks", "alice")?, Some(100.0));
  assert_eq!(ns_apple.zrank("ranks", "alice")?, Some(2));
  assert_eq!(ns_apple.zrevrank("ranks", "alice")?, Some(0));

  // 7. Namespace Keys 模式匹配
  let apple_keys = ns_apple.keys("*")?;
  assert!(apple_keys.contains(&b"device".to_vec()));
  assert!(apple_keys.contains(&b"user:10".to_vec()));

  // 8. Namespace CLEAR (清空当前租户数据)
  let cleared = ns_apple.clear()?;
  assert!(cleared > 0);
  assert_eq!(ns_apple.get("device")?, None);
  assert_eq!(ns_apple.hget("user:10", "name")?, None);
  assert_eq!(ns_apple.zcard("ranks")?, 0);

  // 验证不影响其他租户
  assert_eq!(ns_banana.get("device")?, Some(b"thinkpad".to_vec()));
  assert_eq!(ns_def.get("device")?, Some(b"server_pc".to_vec()));

  OK
}

#[test]
fn test_wedb_namespace_skip_seek_iterator() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  // 初始状态
  let init_ns: Vec<String> = (&db).into_iter().collect();
  assert_eq!(init_ns, vec!["default".to_string()]);

  // 写入多个命名空间的数据
  let ns_apple = db.namespace("tenant_apple");
  let ns_banana = db.namespace("tenant_banana");
  let ns_db1 = db.select_db(1)?;

  ns_apple.set("k1", "v1", &[])?;
  ns_banana.set("k2", "v2", &[])?;
  ns_db1.set("k3", "v3", &[])?;

  // 1. &db 的 IntoIterator 惰性跳跃遍历（列出所有独立命名的租户命名空间）
  let mut iterated = Vec::new();
  for ns_name in &db {
    iterated.push(ns_name);
  }
  assert_eq!(iterated, vec!["default", "tenant_apple", "tenant_banana"]);

  // 2. 分页与短路测试
  let top2: Vec<String> = (&db).into_iter().take(2).collect();
  assert_eq!(top2, vec!["default", "tenant_apple"]);
  assert!((&db).into_iter().any(|ns| ns == "tenant_apple"));

  // 3. ns 的 IntoIterator 遍历
  let mut apple_iter_dbs = Vec::new();
  for db_idx in &ns_apple {
    apple_iter_dbs.push(db_idx);
  }
  assert_eq!(apple_iter_dbs, vec![0]);

  let mut db1_iter_dbs = Vec::new();
  for db_idx in &ns_db1 {
    db1_iter_dbs.push(db_idx);
  }
  assert_eq!(db1_iter_dbs, vec![0, 1]);

  OK
}

#[test]
fn test_catalog_and_large_dbs_numerical_order() -> Void {
  use wedb_embed::WeDb;

  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let ns_apple = db.namespace("tenant_apple");

  // 激活多个乱序 DB，包含 u64 极限数值 (如 0, 1, 2, 10, 100, 1000, 999999, u64::MAX)
  db.activate_db("tenant_apple", 1000)?;
  db.activate_db("tenant_apple", 2)?;
  db.activate_db("tenant_apple", 10)?;
  db.activate_db("tenant_apple", 999999)?;
  db.activate_db("tenant_apple", 1)?;
  db.activate_db("tenant_apple", 100)?;
  db.activate_db("tenant_apple", u64::MAX - 1)?;

  // 迭代器输出必须保持严格的数值升序（8 字节大端序保序）
  let apple_dbs: Vec<u64> = (&ns_apple).into_iter().collect();
  assert_eq!(
    apple_dbs,
    vec![0, 1, 2, 10, 100, 1000, 999999, u64::MAX - 1]
  );

  // 清空租户后，Catalog 也同步清理
  ns_apple.clear()?;
  let cleared_dbs: Vec<u64> = (&ns_apple).into_iter().collect();
  assert_eq!(cleared_dbs, vec![0]);

  OK
}

#[test]
fn test_numerical_namespace_id_and_instant_rename() -> Void {
  use wedb_embed::WeDb;

  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  // 默认租户 ID 为 0
  assert_eq!(db.ns_id("default")?, 0);
  assert_eq!(db.ns_name(0)?, Some("default".to_string()));

  // 分配租户数字 ID
  let apple_id = db.ns_id("tenant_apple")?;
  let google_id = db.ns_id("tenant_google")?;
  assert_eq!(apple_id, 1);
  assert_eq!(google_id, 2);

  // 幂等查询
  assert_eq!(db.query_ns_id("tenant_apple")?, Some(1));
  assert_eq!(db.ns_name(1)?, Some("tenant_apple".to_string()));
  assert_eq!(db.ns_name(2)?, Some("tenant_google".to_string()));

  // Namespace 句柄上的 id 和 rename 方法
  let ns_apple = db.namespace("tenant_apple");
  assert_eq!(ns_apple.id()?, 1);

  // O(1) 租户重命名测试 (WeDb 与 Namespace 句柄调用)
  db.rename_namespace("tenant_apple", "tenant_apple_v2")?;
  let ns_apple_v2 = db.namespace("tenant_apple_v2");
  assert_eq!(ns_apple_v2.id()?, 1);
  ns_apple_v2.rename("tenant_apple_v3")?;
  assert_eq!(db.query_ns_id("tenant_apple_v2")?, None);
  assert_eq!(db.query_ns_id("tenant_apple_v3")?, Some(1));
  assert_eq!(db.ns_name(1)?, Some("tenant_apple_v3".to_string()));

  OK
}

#[test]
fn test_namespace_id_persistence_across_restarts() -> Void {
  let dir = tempfile::tempdir()?;
  let path = dir.path();

  // 1. 第一次打开数据库，分配两个租户并写入数据
  {
    let db = WeDb::open(path, [])?;
    let id1 = db.ns_id("tenant_apple")?;
    let id2 = db.ns_id("tenant_google")?;
    assert_eq!(id1, 1);
    assert_eq!(id2, 2);

    let ns1 = db.namespace("tenant_apple");
    ns1.set("key1", "val1", &[])?;
    let ns2 = db.namespace("tenant_google");
    ns2.set("key2", "val2", &[])?;
  } // 彻底 drop 关闭数据库实例

  // 2. 第二次重新打开数据库，验证旧租户保持旧 ID，新租户接着自增
  {
    let db = WeDb::open(path, [])?;
    // 旧租户 ID 保持原状
    let id1 = db.ns_id("tenant_apple")?;
    let id2 = db.ns_id("tenant_google")?;
    assert_eq!(id1, 1);
    assert_eq!(id2, 2);

    // 新租户接着之前的发号器分配 ID = 3
    let id3 = db.ns_id("tenant_microsoft")?;
    assert_eq!(id3, 3);

    // 验证先前的数据依然通过租户句柄正常可读
    let ns1 = db.namespace("tenant_apple");
    assert_eq!(ns1.get("key1")?.unwrap(), b"val1");
    let ns2 = db.namespace("tenant_google");
    assert_eq!(ns2.get("key2")?.unwrap(), b"val2");

    let ns3 = db.namespace("tenant_microsoft");
    ns3.set("key3", "val3", &[])?;
    assert_eq!(ns3.get("key3")?.unwrap(), b"val3");
  }

  // 3. 第三次重新打开数据库，验证新租户分配 ID = 4
  {
    let db = WeDb::open(path, [])?;
    let id4 = db.ns_id("tenant_amazon")?;
    assert_eq!(id4, 4);
    assert_eq!(db.ns_id("tenant_microsoft")?, 3);
  }

  OK
}

#[test]
fn test_anti_penetration_and_binary_isolation() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let default_ns = db.default_ns();
  let db1 = db.namespace("db1");

  // 1. 验证恶意二进制 Key 绝不穿透到其他租户
  let malicious_bin_key = b"\x00ns:db1:secret";
  default_ns.set(malicious_bin_key, b"hacked_val", &[])?;
  db1.set(b"secret", b"real_val", &[])?;

  // 两者必须绝对正交物理隔离
  assert_eq!(default_ns.get(malicious_bin_key)?.unwrap(), b"hacked_val");
  assert_eq!(db1.get(b"secret")?.unwrap(), b"real_val");

  // 2. 验证恶意字符串 Key 绝不覆盖破坏同名复合结构的 KeyMeta 元数据
  let malicious_hash_meta_key = b"\x00\x01:my_hash";
  default_ns.set(malicious_hash_meta_key, b"fake_meta_str", &[])?;

  default_ns.hset(b"my_hash", &[(b"f1", b"v1")])?;
  assert_eq!(default_ns.hget(b"my_hash", b"f1")?.unwrap(), b"v1");
  assert_eq!(
    default_ns.get(malicious_hash_meta_key)?.unwrap(),
    b"fake_meta_str"
  );

  // 3. 验证默认空间的独立 clear 与 keys 查询
  assert!(default_ns.keys("*")?.contains(&malicious_bin_key.to_vec()));
  assert!(
    default_ns
      .keys("*")?
      .contains(&malicious_hash_meta_key.to_vec())
  );
  assert!(default_ns.keys("*")?.contains(&b"my_hash".to_vec()));

  // 清空默认空间，不影响 db1 租户
  default_ns.clear()?;
  assert_eq!(default_ns.get(malicious_bin_key)?, None);
  assert_eq!(default_ns.hget(b"my_hash", b"f1")?, None);
  assert_eq!(db1.get(b"secret")?.unwrap(), b"real_val");

  OK
}

#[test]
fn test_exhaustive_anti_penetration_matrix() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let t1 = db.namespace("tenant_alpha");
  let t2 = db.namespace("tenant_beta");
  let def = db.default_ns();

  // 1. 跨租户伪造前缀穿透测试：tenant_alpha 写入以 tenant_beta 开头的恶意 Key
  let crafted_key = b"\x00ns:tenant_beta:\x00admin_token";
  t1.set(crafted_key, b"alpha_spoofed_val", &[])?;
  t2.set(b"admin_token", b"beta_genuine_val", &[])?;
  def.set(crafted_key, b"default_spoofed_val", &[])?;

  // 验证严格隔离：三者值互不影响
  assert_eq!(t1.get(crafted_key)?.unwrap(), b"alpha_spoofed_val");
  assert_eq!(t2.get(b"admin_token")?.unwrap(), b"beta_genuine_val");
  assert_eq!(def.get(crafted_key)?.unwrap(), b"default_spoofed_val");

  // 2. 跨类型冲突与类型防护测试 (WRONGTYPE 校验)
  t1.hset("hkey:1", &[("field", "val1")])?;
  assert!(t1.sadd("hkey:1", &["member"]).is_err()); // HASH 无法执行 SADD
  assert!(t1.rpush("hkey:1", &["item"]).is_err()); // HASH 无法执行 RPUSH
  assert!(t1.zadd("hkey:1", &[(1.0, "z")], []).is_err()); // HASH 无法执行 ZADD

  // 3. 复合类型子键前缀穿透边界测试：user:1 绝不污染 user:10
  t1.hset("user:1", &[("field", "val1")])?;
  t1.hset("user:10", &[("field", "val10")])?;
  t1.sadd("set:1", &["member1"])?;
  t1.sadd("set:10", &["member10"])?;
  t1.zadd("zset:1", &[(10.0, "z1")], [])?;
  t1.zadd("zset:10", &[(20.0, "z10")], [])?;
  t1.rpush("list:1", &["list1"])?;
  t1.rpush("list:10", &["list10"])?;

  assert_eq!(t1.hget("user:1", "field")?.unwrap(), b"val1");
  assert_eq!(t1.hget("user:10", "field")?.unwrap(), b"val10");
  assert_eq!(t1.scard("set:1")?, 1);
  assert_eq!(t1.scard("set:10")?, 1);
  assert_eq!(t1.zcard("zset:1")?, 1);
  assert_eq!(t1.zcard("zset:10")?, 1);
  assert_eq!(t1.llen("list:1")?, 1);
  assert_eq!(t1.llen("list:10")?, 1);

  // 删除 user:1, set:1, zset:1, list:1，验证后缀为 10 的键完好无损
  t1.del(&["user:1", "set:1", "zset:1", "list:1"])?;
  assert_eq!(t1.hget("user:1", "field")?, None);
  assert_eq!(t1.scard("set:1")?, 0);
  assert_eq!(t1.zcard("zset:1")?, 0);
  assert_eq!(t1.llen("list:1")?, 0);

  assert_eq!(t1.hget("user:10", "field")?.unwrap(), b"val10");
  assert_eq!(t1.scard("set:10")?, 1);
  assert_eq!(t1.zcard("zset:10")?, 1);
  assert_eq!(t1.llen("list:10")?, 1);

  // 3. 多租户下复合结构主动过期与全局 flushall
  db.active_expire_cycle(100)?;
  assert_eq!(t1.hget("user:10", "field")?.unwrap(), b"val10");

  db.flushall()?;
  assert_eq!(t1.hget("user:10", "field")?, None);
  assert_eq!(t2.get(b"admin_token")?, None);
  assert_eq!(def.get(crafted_key)?, None);

  OK
}

#[test]
fn test_arbitrary_colons_and_binary_keys_zero_collision() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  // 1. 租户名本身包含冒号和特殊字符
  let ns_colon = db.namespace("org:team:prod:us-east");
  let ns_db2 = ns_colon.select_db(2)?;
  let ns_def = db.default_ns();

  // 2. 各种极端的冒号 Key
  let key_multi_colons = "::::";
  let key_prefix_colon = ":leading:colon:key";
  let key_suffix_colon = "trailing:colon:key:";
  let key_url_colons = "https://user:pass@api.wedb.io:8443/v1/query?tag=a:b:c";
  let key_binary_colons = b"\x00\x01:::binary\x00\xff:key:";

  // 2.1 测试 String 存储与读取
  ns_db2.set(key_multi_colons, "val_multi", &[])?;
  ns_db2.set(key_prefix_colon, "val_prefix", &[])?;
  ns_db2.set(key_suffix_colon, "val_suffix", &[])?;
  ns_db2.set(key_url_colons, "val_url", &[])?;
  ns_db2.set(key_binary_colons, b"val_bin", &[])?;

  ns_def.set(key_multi_colons, "def_multi", &[])?;
  ns_def.set(key_url_colons, "def_url", &[])?;

  // 验证读取精准无误
  assert_eq!(ns_db2.get(key_multi_colons)?.unwrap(), b"val_multi");
  assert_eq!(ns_db2.get(key_prefix_colon)?.unwrap(), b"val_prefix");
  assert_eq!(ns_db2.get(key_suffix_colon)?.unwrap(), b"val_suffix");
  assert_eq!(ns_db2.get(key_url_colons)?.unwrap(), b"val_url");
  assert_eq!(ns_db2.get(key_binary_colons)?.unwrap(), b"val_bin");

  assert_eq!(ns_def.get(key_multi_colons)?.unwrap(), b"def_multi");
  assert_eq!(ns_def.get(key_url_colons)?.unwrap(), b"def_url");

  // 2.2 测试 Hash 复合结构：Key 和 Field 同时包含海量冒号
  let hash_key = "hash:main:user:1001";
  let field_url = "http://schema.org:80/field:sub:1";
  let field_colons = "::nested::field::";

  ns_db2.hset(hash_key, &[(field_url, "val1"), (field_colons, "val2")])?;

  assert_eq!(ns_db2.hget(hash_key, field_url)?.unwrap(), b"val1");
  assert_eq!(ns_db2.hget(hash_key, field_colons)?.unwrap(), b"val2");

  let all_fields = ns_db2.hgetall(hash_key)?;
  assert_eq!(all_fields.len(), 2);
  assert!(
    all_fields
      .iter()
      .any(|(f, v)| f == field_url.as_bytes() && v == b"val1")
  );
  assert!(
    all_fields
      .iter()
      .any(|(f, v)| f == field_colons.as_bytes() && v == b"val2")
  );

  // 2.3 测试 ZSet 复合结构：Member 包含冒号
  let zset_key = "zset:score:index:2026";
  let member_with_colon = "member:id:888:sub:99";
  ns_db2.zadd(zset_key, &[(100.5, member_with_colon)], [])?;
  assert_eq!(ns_db2.zscore(zset_key, member_with_colon)?.unwrap(), 100.5);

  // 2.4 测试 Keys 查询与 Pattern 匹配能够精准提取包含冒号的原始 Key
  let keys = ns_db2.keys("*")?;
  assert!(keys.contains(&key_multi_colons.as_bytes().to_vec()));
  assert!(keys.contains(&key_prefix_colon.as_bytes().to_vec()));
  assert!(keys.contains(&key_suffix_colon.as_bytes().to_vec()));
  assert!(keys.contains(&key_url_colons.as_bytes().to_vec()));
  assert!(keys.contains(&key_binary_colons.to_vec()));
  assert!(keys.contains(&hash_key.as_bytes().to_vec()));
  assert!(keys.contains(&zset_key.as_bytes().to_vec()));

  Ok(())
}

#[test]
fn test_keys_with_colon_subkeys_exact_recovery() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let ns = db.namespace("tenant_colons");

  // 1. 写入普通 String 键
  ns.set("user", "normal_val", &[])?;

  // 2. 写入 Hash，且 field 包含连续多段冒号
  ns.hset(
    "app",
    &[("a:b:c:d", "hash_val1"), ("config:sub:prop", "hash_val2")],
  )?;

  // 3. 写入 Set，且 member 包含冒号
  ns.sadd("tags", &["lang:rust:v1", "db:kvrocks:rocksdb"])?;

  // 4. 获取所有 Key
  let all_keys = ns.keys("*")?;
  let key_strings: Vec<String> = all_keys
    .iter()
    .map(|k| String::from_utf8_lossy(k).to_string())
    .collect();

  // 必须精准只包含 3 个顶层用户键，绝不能出现因反解子键冒号而泄露的虚假幽灵键
  assert_eq!(all_keys.len(), 3);
  assert!(key_strings.contains(&"user".to_string()));
  assert!(key_strings.contains(&"app".to_string()));
  assert!(key_strings.contains(&"tags".to_string()));

  assert!(!key_strings.contains(&"app:a".to_string()));
  assert!(!key_strings.contains(&"app:a:b".to_string()));
  assert!(!key_strings.contains(&"tags:lang".to_string()));

  Ok(())
}

#[test]
fn test_tenant_multi_db_clear_cascade() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let ns0 = db.namespace("tenant_multi");
  let ns1 = db.namespace("tenant_multi").select_db(1)?;
  let ns2 = db.namespace("tenant_multi").select_db(2)?;

  ns0.set("k0", "v0", &[])?;
  ns1.set("k1", "v1", &[])?;
  ns2.set("k2", "v2", &[])?;

  assert_eq!(ns0.get("k0")?, Some(b"v0".to_vec()));
  assert_eq!(ns1.get("k1")?, Some(b"v1".to_vec()));
  assert_eq!(ns2.get("k2")?, Some(b"v2".to_vec()));

  // 在租户级别调用 clear (db 0)
  let cleared = ns0.clear()?;
  assert!(cleared >= 3);

  // 验证该租户下所有 DB (db 0, db 1, db 2) 的数据均被彻底级联物理清理
  assert_eq!(ns0.get("k0")?, None);
  assert_eq!(ns1.get("k1")?, None);
  assert_eq!(ns2.get("k2")?, None);

  Ok(())
}

#[test]
fn test_cross_key_prefix_collision_isolation() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;
  let ns = db.namespace("test_cross_prefix");

  // 1. 验证 Hash 交叉键碰撞隔离（Hash A: key="user", field="info:name" vs Hash B: key="user:info", field="name"）
  ns.hset("user", &[("info:name", "Alice"), ("info:age", "30")])?;
  ns.hset("user:info", &[("name", "Bob"), ("country", "US")])?;

  // 两个 Hash 的数据必须互不干扰，绝不覆写
  assert_eq!(ns.hget("user", "info:name")?.unwrap(), b"Alice");
  assert_eq!(ns.hget("user", "info:age")?.unwrap(), b"30");
  assert_eq!(ns.hget("user:info", "name")?.unwrap(), b"Bob");
  assert_eq!(ns.hget("user:info", "country")?.unwrap(), b"US");

  // HGETALL 扫描必须精确隔离，绝不泄漏对方字段
  let user_all = ns.hgetall("user")?;
  assert_eq!(user_all.len(), 2);
  assert!(
    user_all
      .iter()
      .any(|(f, v)| f == b"info:name" && v == b"Alice")
  );
  assert!(user_all.iter().any(|(f, v)| f == b"info:age" && v == b"30"));

  let user_info_all = ns.hgetall("user:info")?;
  assert_eq!(user_info_all.len(), 2);
  assert!(
    user_info_all
      .iter()
      .any(|(f, v)| f == b"name" && v == b"Bob")
  );
  assert!(
    user_info_all
      .iter()
      .any(|(f, v)| f == b"country" && v == b"US")
  );

  // 删除 Hash A (user)，不能影响 Hash B (user:info)
  assert_eq!(ns.del(&["user"])?, 1);
  assert_eq!(ns.hget("user", "info:name")?, None);
  assert_eq!(ns.hget("user:info", "name")?.unwrap(), b"Bob");
  assert_eq!(ns.hget("user:info", "country")?.unwrap(), b"US");

  // 2. 验证 Set 交叉键隔离（Set A: key="group", member="admin:super" vs Set B: key="group:admin", member="super"）
  ns.sadd("group", &["admin:super", "guest"])?;
  ns.sadd("group:admin", &["super", "root"])?;

  assert!(ns.sismember("group", "admin:super")?);
  assert!(ns.sismember("group:admin", "super")?);
  assert!(!ns.sismember("group", "super")?);
  assert!(!ns.sismember("group:admin", "admin:super")?);

  let group_members = ns.smembers("group")?;
  assert_eq!(group_members.len(), 2);
  assert!(group_members.contains(&b"admin:super".to_vec()));
  assert!(group_members.contains(&b"guest".to_vec()));

  let group_admin_members = ns.smembers("group:admin")?;
  assert_eq!(group_admin_members.len(), 2);
  assert!(group_admin_members.contains(&b"super".to_vec()));
  assert!(group_admin_members.contains(&b"root".to_vec()));

  // 3. 验证 ZSet 交叉键隔离
  ns.zadd("rank", &[(100.0, "vip:1")], [])?;
  ns.zadd("rank:vip", &[(200.0, "1")], [])?;

  assert_eq!(ns.zscore("rank", "vip:1")?.unwrap(), 100.0);
  assert_eq!(ns.zscore("rank:vip", "1")?.unwrap(), 200.0);
  assert_eq!(ns.zscore("rank", "1")?, None);

  Ok(())
}

#[test]
fn test_arbitrary_binary_keys_and_null_bytes() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;
  let ns = db.namespace("bin_ns");

  // 构造含 0x00, 0xFF, 冒号 0x3A, 各种不可见字节的极端键
  let bin_str_key: &[u8] = b"\x00\x00\xff\xfe:str\x00key:1";
  let bin_hash_key_1: &[u8] = b"\x00\x00\xff\xfe:hash\x00key:1";
  let bin_hash_key_2: &[u8] = b"\x00\x00\xff\xfe:hash\x00key";
  let bin_set_key: &[u8] = b"\x00\x00\xff\xfe:set\x00key:1";
  let bin_field_1: &[u8] = b"\x00\x01\x02:::field:\xff";
  let bin_field_2: &[u8] = b":field:\xff";

  // 1. String 存储
  ns.set(bin_str_key, b"\xde\xad\xbe\xef", &[])?;
  assert_eq!(ns.get(bin_str_key)?.unwrap(), b"\xde\xad\xbe\xef");

  // 2. Hash 存储
  ns.hset(
    bin_hash_key_1,
    &[
      (bin_field_1, b"hval1".as_slice()),
      (bin_field_2, b"hval2".as_slice()),
    ],
  )?;
  ns.hset(bin_hash_key_2, &[(bin_field_1, b"hval3".as_slice())])?;

  assert_eq!(ns.hget(bin_hash_key_1, bin_field_1)?.unwrap(), b"hval1");
  assert_eq!(ns.hget(bin_hash_key_1, bin_field_2)?.unwrap(), b"hval2");
  assert_eq!(ns.hget(bin_hash_key_2, bin_field_1)?.unwrap(), b"hval3");

  let all = ns.hgetall(bin_hash_key_1)?;
  assert_eq!(all.len(), 2);

  // 3. Set 存储
  ns.sadd(bin_set_key, &[bin_field_1, bin_field_2])?;
  assert!(ns.sismember(bin_set_key, bin_field_1)?);
  assert!(ns.sismember(bin_set_key, bin_field_2)?);

  // 4. keys("*") 提取
  let all_keys = ns.keys("*")?;
  assert!(all_keys.contains(&bin_str_key.to_vec()));
  assert!(all_keys.contains(&bin_hash_key_1.to_vec()));
  assert!(all_keys.contains(&bin_hash_key_2.to_vec()));
  assert!(all_keys.contains(&bin_set_key.to_vec()));

  Ok(())
}

#[test]
fn test_namespace_select_db_and_catalog_helpers() -> Void {
  use wedb_embed::key_composer::{
    CATALOG_PREFIX, NS_ID_PREFIX, NS_NAME_PREFIX, NS_NEXT_ID_KEY, NS_TOKEN_PREFIX, catalog_db_key,
    catalog_ns_prefix, ns_id_key, ns_name_key, ns_token_key,
  };

  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  // 1. 常量校验 (基于纯 enum u8 结构)
  assert_eq!(NS_NAME_PREFIX, b"\x00\x70\x01");
  assert_eq!(NS_ID_PREFIX, b"\x00\x70\x02");
  assert_eq!(NS_NEXT_ID_KEY, b"\x00\x70\x03");
  assert_eq!(NS_TOKEN_PREFIX, b"\x00\x70\x04");
  assert_eq!(CATALOG_PREFIX, b"\x00\x71");

  assert_eq!(ns_name_key("alpha"), b"\x00\x70\x01alpha");
  assert_eq!(ns_token_key("tok"), b"\x00\x70\x04tok");
  assert_eq!(ns_id_key(1), b"\x00\x70\x02\x01");
  assert_eq!(catalog_ns_prefix("alpha"), b"\x00\x71\x05alpha");
  assert_eq!(catalog_db_key("alpha", 2), b"\x00\x71\x05alpha\x02");

  // 2. Namespace 上的 select_db 链式调用
  let ns = db.namespace("tenant_gamma");
  let ns_db3 = ns.select_db(3)?;
  assert_eq!(ns_db3.name(), "tenant_gamma");
  assert_eq!(ns_db3.db_index(), 3);

  ns_db3.set("gamma_key", "gamma_val", &[])?;
  assert_eq!(ns_db3.get("gamma_key")?.unwrap(), b"gamma_val");
  assert_eq!(ns.get("gamma_key")?, None);

  // 3. 测试 flushdb 单库清除与 rename_namespace 目录同步
  ns.set("gamma_db0_key", "gamma_db0_val", &[])?;
  assert_eq!(ns.get("gamma_db0_key")?.unwrap(), b"gamma_db0_val");

  let dbs_before: Vec<u64> = ns.iter().collect();
  assert_eq!(dbs_before, vec![0, 3]);

  // 重命名 tenant_gamma -> tenant_gamma_renamed
  ns.rename("tenant_gamma_renamed")?;
  let ns_renamed = db.namespace("tenant_gamma_renamed");
  let dbs_after: Vec<u64> = ns_renamed.iter().collect();
  assert_eq!(dbs_after, vec![0, 3]);

  let ns_renamed_db3 = ns_renamed.select_db(3)?;
  assert_eq!(ns_renamed_db3.get("gamma_key")?.unwrap(), b"gamma_val");
  assert_eq!(ns_renamed.get("gamma_db0_key")?.unwrap(), b"gamma_db0_val");

  // flushdb 仅清除 db 3，不影响 db 0
  ns_renamed_db3.flushdb()?;
  assert_eq!(ns_renamed_db3.get("gamma_key")?, None);
  assert_eq!(ns_renamed.get("gamma_db0_key")?.unwrap(), b"gamma_db0_val");

  Ok(())
}

#[test]
fn test_default_ns_multi_db_cascade_clear_and_binary_patterns() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let ns_def0 = db.default_ns();
  let ns_def1 = db.select_db(1)?;
  let ns_t0 = db.namespace("tenant_xyz");
  let ns_t1 = ns_t0.select_db(1)?;

  // 1. 写入 4 种作用域下的同名键（完全隔离）
  let bin_key = b"\x00\x01\x02:my:raw\x00key:\xff";
  ns_def0.set(bin_key, "val_def0", &[])?;
  ns_def1.set(bin_key, "val_def1", &[])?;
  ns_t0.set(bin_key, "val_t0", &[])?;
  ns_t1.set(bin_key, "val_t1", &[])?;

  assert_eq!(ns_def0.get(bin_key)?.unwrap(), b"val_def0");
  assert_eq!(ns_def1.get(bin_key)?.unwrap(), b"val_def1");
  assert_eq!(ns_t0.get(bin_key)?.unwrap(), b"val_t0");
  assert_eq!(ns_t1.get(bin_key)?.unwrap(), b"val_t1");

  // 2. 二进制通配符测试 (AsRef<[u8]>)
  let def0_keys_str = ns_def0.keys("*")?;
  let def0_keys_bytes = ns_def0.keys(b"\x00\x01*")?;
  assert_eq!(def0_keys_str.len(), 1);
  assert_eq!(def0_keys_bytes.len(), 1);
  assert_eq!(&def0_keys_bytes[0], bin_key);

  // 3. 测试 default_ns flushall / clear 级联清理 default 下所有 db，但不影响其他 tenant
  ns_def0.clear()?;
  assert_eq!(ns_def0.get(bin_key)?, None);
  assert_eq!(ns_def1.get(bin_key)?, None);
  // tenant_xyz 完全不受影响
  assert_eq!(ns_t0.get(bin_key)?.unwrap(), b"val_t0");
  assert_eq!(ns_t1.get(bin_key)?.unwrap(), b"val_t1");

  Ok(())
}

#[test]
fn test_concurrent_auto_increment_ns_id_allocation() -> Void {
  use std::{collections::HashSet, sync::Arc, thread};

  let dir = tempfile::tempdir()?;
  let db = Arc::new(WeDb::open(dir.path(), [])?);

  let num_threads = 16;
  let num_tenants_per_thread = 50;

  let mut handles = Vec::new();
  for t in 0..num_threads {
    let db_clone = db.clone();
    handles.push(thread::spawn(move || {
      let mut ids = Vec::new();
      for i in 0..num_tenants_per_thread {
        let name = format!("tenant_{}_{}", t, i);
        let ns = db_clone.namespace(&name);
        let id = ns.kc.ns_id();
        ids.push((name, id));
      }
      ids
    }));
  }

  let mut all_ids = HashSet::new();
  let total_expected = num_threads * num_tenants_per_thread;

  for h in handles {
    let ids = h.join().unwrap();
    for (name, id) in ids {
      assert!(
        id >= 1,
        "Tenant {name} must have positive auto-increment id >= 1"
      );
      assert!(
        all_ids.insert(id),
        "Duplicate auto-increment ID detected: {id} for tenant {name}"
      );
    }
  }

  assert_eq!(all_ids.len(), total_expected);
  // 验证分配的 ID 必须连续覆盖 1..=total_expected
  for expected_id in 1..=(total_expected as u64) {
    assert!(
      all_ids.contains(&expected_id),
      "Missing expected auto-increment ID {expected_id}"
    );
  }

  Ok(())
}

#[test]
fn test_colon_in_namespace_catalog_collision_freedom() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  // 两个具有前缀包含关系的命名空间
  let ns_short = db.namespace("alpha");
  let ns_long = db.namespace("alpha:db:1");

  ns_short.select_db(1)?.set("k1", "v_short_db1", &[])?;
  ns_short.select_db(2)?.set("k2", "v_short_db2", &[])?;
  ns_long.select_db(5)?.set("k5", "v_long_db5", &[])?;

  // 验证 short 的 DB 列表绝不包含 long 的 DB 5
  let short_dbs: Vec<u64> = ns_short.iter().collect();
  assert_eq!(short_dbs, vec![0, 1, 2]);

  let long_dbs: Vec<u64> = ns_long.iter().collect();
  assert_eq!(long_dbs, vec![0, 5]);

  // 重命名 short 绝不影响 long
  ns_short.rename("alpha_renamed")?;
  let ns_short_renamed = db.namespace("alpha_renamed");
  assert_eq!(
    ns_short_renamed.select_db(1)?.get("k1")?.unwrap(),
    b"v_short_db1"
  );
  assert_eq!(ns_long.select_db(5)?.get("k5")?.unwrap(), b"v_long_db5");

  // 验证重复 rename 会被拒绝
  assert!(ns_short_renamed.rename("alpha:db:1").is_err());

  Ok(())
}

#[test]
fn test_tenant_subdb_flushall_and_small_key_types() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let ns_tenant = db.namespace("tenant_flush_test");
  let ns_db1 = ns_tenant.select_db(1)?;
  let ns_db2 = ns_tenant.select_db(2)?;

  ns_tenant.set("k0", "v0", &[])?;
  ns_db1.set("k1", "v1", &[])?;
  ns_db2.set("k2", "v2", &[])?;

  // 从子库 ns_db2 触发 flushall()，应当清空该租户下的全部 DB (db0, db1, db2)
  let flushed = ns_db2.flushall()?;
  assert!(flushed >= 3);

  assert_eq!(ns_tenant.get("k0")?, None);
  assert_eq!(ns_db1.get("k1")?, None);
  assert_eq!(ns_db2.get("k2")?, None);

  // 验证 SmallKey 的从超长切片构建与 Hash/Ord 特性
  use wedb_embed::key_composer::SmallKey;
  let short_sk = SmallKey::from_slice(b"hello");
  assert_eq!(short_sk.as_bytes(), b"hello");

  let long_bytes = [0x42u8; 200];
  let long_sk = SmallKey::from_slice(&long_bytes);
  assert_eq!(long_sk.as_bytes(), &long_bytes);

  use std::collections::HashSet;
  let mut set = HashSet::new();
  set.insert(short_sk.clone());
  set.insert(long_sk.clone());
  assert!(set.contains(&short_sk));
  assert!(set.contains(&long_sk));

  Ok(())
}

#[test]
fn test_active_expire_cycle_cursor_progression_and_composite_subkeys_cascade_purge() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let ns_def = db.default_ns();
  let ns_tenant = db.namespace("tenant_gc");
  let ns_tenant_db2 = ns_tenant.select_db(2)?;

  // 1. 写入默认空间带 TTL 的复合类型与 String
  ns_def.set("str_def", "val_def", &[StringSet::Px(1)])?;
  ns_def.hset("hash_def", &[("f1", "v1"), ("f2", "v2")])?;
  ns_def.expire("hash_def", 1)?;
  ns_def.sadd("set_def", &["m1", "m2", "m3"])?;
  ns_def.expire("set_def", 1)?;
  ns_def.zadd("zset_def", &[(1.0, "zm1"), (2.0, "zm2")], [])?;
  ns_def.expire("zset_def", 1)?;

  // 2. 写入租户带 TTL 的复合类型与 String
  ns_tenant.set("str_t", "val_t", &[StringSet::Px(1)])?;
  ns_tenant.hset("hash_t", &[("f1", "v1")])?;
  ns_tenant.expire("hash_t", 1)?;

  ns_tenant_db2.set("str_t_db2", "val_t_db2", &[StringSet::Px(1)])?;
  ns_tenant_db2.zadd("zset_t_db2", &[(10.0, "member_a")], [])?;
  ns_tenant_db2.expire("zset_t_db2", 1)?;

  // 写入无 TTL 永不过期的键
  ns_def.set("str_persist", "val_persist", &[])?;
  ns_tenant.set("str_t_persist", "val_t_persist", &[])?;

  // 等待过期
  thread::sleep(Duration::from_millis(1500));

  // 3. 执行多次小批量主动过期采样扫描（验证游标推进与全表轮询）
  let mut total_cleaned = 0;
  for _ in 0..10 {
    total_cleaned += db.active_expire_cycle(3)?;
  }

  assert!(
    total_cleaned >= 7,
    "Total cleaned {total_cleaned} should be >= 7"
  );

  // 4. 验证过期键全部不可见
  assert_eq!(ns_def.get("str_def")?, None);
  assert_eq!(ns_def.exists(&["hash_def"])?, 0);
  assert_eq!(ns_def.exists(&["set_def"])?, 0);
  assert_eq!(ns_def.exists(&["zset_def"])?, 0);
  assert_eq!(ns_tenant.get("str_t")?, None);
  assert_eq!(ns_tenant.exists(&["hash_t"])?, 0);
  assert_eq!(ns_tenant_db2.get("str_t_db2")?, None);
  assert_eq!(ns_tenant_db2.exists(&["zset_t_db2"])?, 0);

  // 验证持久键完全完好
  assert_eq!(ns_def.get("str_persist")?.unwrap(), b"val_persist");
  assert_eq!(ns_tenant.get("str_t_persist")?.unwrap(), b"val_t_persist");

  // 5. 深度验证底层 data / data_ns 列族中的子键数据已被物理级联彻底清除（无僵尸残留）
  let hash_def_prefix = compose_hash_prefix(&ns_def.kc, b"hash_def");
  assert_eq!(db.data.prefix(&hash_def_prefix).count(), 0);

  let set_def_prefix = compose_set_prefix(&ns_def.kc, b"set_def");
  assert_eq!(db.data.prefix(&set_def_prefix).count(), 0);

  let zset_def_prefix = compose_zset_prefix(&ns_def.kc, b"zset_def");
  assert_eq!(db.data.prefix(&zset_def_prefix).count(), 0);

  let hash_t_prefix = compose_hash_prefix(&ns_tenant.kc, b"hash_t");
  assert_eq!(db.data_ns.prefix(&hash_t_prefix).count(), 0);

  let zset_t_db2_prefix = compose_zset_prefix(&ns_tenant_db2.kc, b"zset_t_db2");
  assert_eq!(db.data_ns.prefix(&zset_t_db2_prefix).count(), 0);

  Ok(())
}

#[test]
fn test_binary_key_scope_fallback_and_adversarial_patterns() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let ns_def = db.default_ns();
  let ns_db1 = db.select_db(1)?;
  let kc_def = KeyComposer::new("default");

  // 1. 测试在默认空间下，用户 key 首字节为 \x01, \x02, \x03（即与 ScopeModeTag 相同）的极端场景
  let raw_bin_key1 = b"\x01\x01\x01\x01";
  let full_str_key1 = compose_string_key(&kc_def, raw_bin_key1);
  assert!(kc_def.is_key_in_ns(&full_str_key1));
  assert_eq!(
    kc_def.extract_user_key(&full_str_key1),
    Some(raw_bin_key1.as_slice())
  );

  // 2. 测试在默认空间下，用户 key 包含 \x01\x01:attack（看似符合模式 1 结构但后随非 KeyTag 字节）
  let raw_bin_key2 = b"\x01\x01:attack";
  let full_str_key2 = compose_string_key(&kc_def, raw_bin_key2);
  assert!(kc_def.is_key_in_ns(&full_str_key2));
  assert_eq!(
    kc_def.extract_user_key(&full_str_key2),
    Some(raw_bin_key2.as_slice())
  );

  // 3. 写入默认空间与 db 1 空间，验证绝对隔离与 keys 查询
  ns_def.set(raw_bin_key1, "val_bin1", &[])?;
  ns_def.set(raw_bin_key2, "val_bin2", &[])?;
  ns_db1.set(b"attack", "val_db1_attack", &[])?;

  assert_eq!(ns_def.get(raw_bin_key1)?.unwrap(), b"val_bin1");
  assert_eq!(ns_def.get(raw_bin_key2)?.unwrap(), b"val_bin2");
  assert_eq!(ns_db1.get(b"attack")?.unwrap(), b"val_db1_attack");

  let def_keys = ns_def.keys("*")?;
  assert!(def_keys.contains(&raw_bin_key1.to_vec()));
  assert!(def_keys.contains(&raw_bin_key2.to_vec()));
  assert!(!def_keys.contains(&b"attack".to_vec()));

  let db1_keys = ns_db1.keys("*")?;
  assert_eq!(db1_keys, vec![b"attack".to_vec()]);

  Ok(())
}

#[test]
fn test_round2_extreme_binary_boundary_and_corrupted_key_parsing() -> Void {
  let dir = tempfile::tempdir()?;
  let db = WeDb::open(dir.path(), [])?;

  let ns_def = db.default_ns();
  let ns_t1 = db.namespace("tenant_r2_1");
  let ns_t1_db5 = ns_t1.select_db(5)?;

  // 1. 极端二进制 Key 与 Subkey（空字节、全 0xFF、包含 \x00 与控制字符）
  let extreme_keys: Vec<&[u8]> = vec![
    b"",
    b"\x00",
    b"\xff",
    b"\x00\x00\x00",
    b"\xff\xff\xff\xff",
    b"\x00\xff\x00\xff\x01\x02\x03\x04",
    b"\x1f\x7f\x00\n\r\t\x00\xff",
  ];

  for &k in &extreme_keys {
    // 1.1 String 存储与读取
    ns_def.set(k, b"def_val", &[])?;
    assert_eq!(ns_def.get(k)?.unwrap(), b"def_val");

    ns_t1.set(k, b"t1_val", &[])?;
    assert_eq!(ns_t1.get(k)?.unwrap(), b"t1_val");

    ns_t1_db5.set(k, b"t1_db5_val", &[])?;
    assert_eq!(ns_t1_db5.get(k)?.unwrap(), b"t1_db5_val");

    // 1.2 严格验证 WRONGTYPE 防穿透：在已存在 String 的键上执行 HSET / SADD / ZADD 必须被拦截报错
    let sub_keys: Vec<&[u8]> = vec![b"", b"\x00", b"\xff", b"\x00\xff:nested:field:\x00"];
    assert!(
      ns_def
        .hset(k, &[(sub_keys[0], b"payload".as_slice())])
        .is_err()
    );
    assert!(ns_def.sadd(k, &[b"\x00".as_slice()]).is_err());
    assert!(ns_def.zadd(k, &[(1.0, b"zm".as_slice())], []).is_err());

    // 删除 String 键后，写入 Hash 结构
    ns_def.del(&[k])?;
    ns_t1.del(&[k])?;
    for &sub in &sub_keys {
      ns_def.hset(k, &[(sub, b"hash_def_payload".as_slice())])?;
      assert_eq!(ns_def.hget(k, sub)?.unwrap(), b"hash_def_payload");

      ns_t1.hset(k, &[(sub, b"hash_t1_payload".as_slice())])?;
      assert_eq!(ns_t1.hget(k, sub)?.unwrap(), b"hash_t1_payload");
    }

    // 删除 Hash 键后，写入 Set 结构
    ns_def.del(&[k])?;
    ns_def.sadd(k, &[b"\x00".as_slice(), b"\xff".as_slice()])?;
    assert!(ns_def.sismember(k, b"\x00")?);
    assert!(ns_def.sismember(k, b"\xff")?);

    // 删除 Set 键后，写入 ZSet 结构
    ns_def.del(&[k])?;
    ns_def.zadd(k, &[(123.456, b"\x00\xff_member".as_slice())], [])?;
    assert_eq!(ns_def.zscore(k, b"\x00\xff_member")?, Some(123.456));
  }

  // 2. 畸变/损坏字节序列防越界与防 Panic 审计
  let kc = KeyComposer::new("default");
  let corrupted_payloads: Vec<&[u8]> = vec![
    b"",
    b"\x00",
    b"\x01",
    b"\x02",
    b"\x00\x01",
    b"\x00\x02",
    b"\x00\x03",
    b"\x00\x03\x80",
    b"\x00\x03\xf8",
    b"\x00\x03\xf8\x00\x00\x00\x00\x00\x00\x00",
    b"\x02\xf8\xff\xff\xff\xff\xff\xff\xff\xff", // 超大 OPPV 长度
    b"\x02\xf0",                                 // 非法首字节
    b"\x02\xff",                                 // 非法首字节
  ];

  for &corrupted in &corrupted_payloads {
    // 必须安全返回 None，绝对杜绝 panic 或越界切片
    let _ = kc.extract_user_key(corrupted);
    let _ = KeyComposer::parse_scoped_prefix(corrupted);
  }

  // 3. WeDb::flushall 后内存 ns_cache 与磁盘严格同步
  assert_eq!(db.query_ns_id("tenant_r2_1")?, Some(1));
  db.flushall()?;
  assert_eq!(db.query_ns_id("tenant_r2_1")?, None);

  Ok(())
}