wedb_embed 0.1.0

use aok::Void;
use wedb_embed::key_composer::{
    InternalKey, KeyComposer, compose_namespace_key, compose_slot_key_prefix,
    compose_slot_key_upper_bound, db_to_namespace, db_to_namespace_kvrocks, extract_namespace_key,
    extract_slot_id, is_namespace_legal, namespace_to_db,
};

fn to_hex(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

#[test]
fn test_kvrocks_namespace_and_select_encoding_golden() -> Void {
    // 1. 默认命名空间 ComposeNamespaceKey 编码
    let def_ns_key = compose_namespace_key(b"default", b"mykey", None);
    assert_eq!(&def_ns_key[..], b"\x07defaultmykey");
    assert_eq!(to_hex(&def_ns_key), "0764656661756c746d796b6579");
    let (ns, slot, key) = extract_namespace_key(&def_ns_key, false).unwrap();
    assert_eq!(ns, b"default");
    assert_eq!(slot, None);
    assert_eq!(key, b"mykey");

    // 2. SELECT 0 与 SELECT 1 命名空间编码
    let db0_ns_key = compose_namespace_key(b"0", b"mykey", None);
    assert_eq!(&db0_ns_key[..], b"\x010mykey");
    assert_eq!(to_hex(&db0_ns_key), "01306d796b6579");
    let (ns0, _, key0) = extract_namespace_key(&db0_ns_key, false).unwrap();
    assert_eq!(ns0, b"0");
    assert_eq!(key0, b"mykey");

    let db1_ns_key = compose_namespace_key(b"1", b"mykey", None);
    assert_eq!(&db1_ns_key[..], b"\x011mykey");
    assert_eq!(to_hex(&db1_ns_key), "01316d796b6579");
    let (ns1, _, key1) = extract_namespace_key(&db1_ns_key, false).unwrap();
    assert_eq!(ns1, b"1");
    assert_eq!(key1, b"mykey");

    // 3. 自定义多租户命名空间编码
    let tenant_ns_key = compose_namespace_key(b"tenant_a", b"mykey", None);
    assert_eq!(&tenant_ns_key[..], b"\x08tenant_amykey");
    assert_eq!(to_hex(&tenant_ns_key), "0874656e616e745f616d796b6579");
    let (ns_t, _, key_t) = extract_namespace_key(&tenant_ns_key, false).unwrap();
    assert_eq!(ns_t, b"tenant_a");
    assert_eq!(key_t, b"mykey");

    // 4. 自定义租户下的 InternalKey (Subkey / Data Key) 编码
    let int_key = InternalKey::new(b"tenant_a", b"user", b"email", 100, None);
    let int_bytes = int_key.encode();
    assert_eq!(
        to_hex(&int_bytes),
        "0874656e616e745f6100000004757365720000000000000064656d61696c"
    );
    let decoded = InternalKey::decode(&int_bytes, false).unwrap();
    assert_eq!(decoded.namespace, b"tenant_a");
    assert_eq!(decoded.key, b"user");
    assert_eq!(decoded.version, 100);
    assert_eq!(decoded.sub_key, b"email");

    // 5. 带 Slot ID 的集群内部键编码验证
    let slot_key = InternalKey::new(b"default", b"user", b"age", 1, Some(1024));
    let slot_bytes = slot_key.encode();
    assert_eq!(
        to_hex(&slot_bytes),
        "0764656661756c74040000000004757365720000000000000001616765"
    );
    let decoded_slot = InternalKey::decode(&slot_bytes, true).unwrap();
    assert_eq!(decoded_slot.slot_id, Some(1024));
    assert_eq!(decoded_slot.key, b"user");
    assert_eq!(extract_slot_id(&slot_bytes), Some(1024));

    Ok(())
}

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

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

    // 2. Hash
    assert_ne!(kc_def.hash_meta("h"), kc_t1.hash_meta("h"));
    assert_ne!(kc_t1.hash_key("h", "f"), kc_t2.hash_key("h", "f"));
    assert_ne!(kc_t1.hash_prefix("h"), kc_t2.hash_prefix("h"));

    // 3. List
    assert_ne!(kc_def.list_meta("l"), kc_t1.list_meta("l"));
    assert_ne!(kc_t1.list_item("l", 1), kc_t2.list_item("l", 1));

    // 4. Set
    assert_ne!(kc_def.set_meta("s"), kc_t1.set_meta("s"));
    assert_ne!(kc_t1.set_key("s", "m"), kc_t2.set_key("s", "m"));

    // 5. ZSet
    assert_ne!(kc_def.zset_meta("z"), kc_t1.zset_meta("z"));
    assert_ne!(kc_t1.zset_key("z", "m"), kc_t2.zset_key("z", "m"));
    assert_ne!(
        kc_t1.zset_score_key("z", 1.5, "m"),
        kc_t2.zset_score_key("z", 1.5, "m")
    );

    // 6. Bitmap
    assert_ne!(kc_def.bm_meta("b"), kc_t1.bm_meta("b"));
    assert_ne!(kc_t1.bm_segment("b", 0), kc_t2.bm_segment("b", 0));

    // 7. Bloom Filter
    assert_ne!(kc_def.bf_meta("bf"), kc_t1.bf_meta("bf"));
    assert_ne!(kc_t1.bf_item("bf", 1), kc_t2.bf_item("bf", 1));

    // 8. Cuckoo Filter
    assert_ne!(kc_def.cf_meta("cf"), kc_t1.cf_meta("cf"));
    assert_ne!(kc_t1.cf_page("cf", 0, 1), kc_t2.cf_page("cf", 0, 1));

    // 9. HyperLogLog
    assert_ne!(kc_def.hll_meta("hll"), kc_t1.hll_meta("hll"));
    assert_ne!(kc_t1.hll_key("hll"), kc_t2.hll_key("hll"));

    // 10. JSON
    assert_ne!(kc_def.json_meta("j"), kc_t1.json_meta("j"));

    // 11. Search (FT)
    assert_ne!(kc_def.ft_schema("idx"), kc_t1.ft_schema("idx"));
    assert_ne!(kc_def.ft_alias("a"), kc_t1.ft_alias("a"));
    assert_ne!(
        kc_t1.ft_index_key("idx", "f", "term", "doc1"),
        kc_t2.ft_index_key("idx", "f", "term", "doc1")
    );

    // 12. SortedInt
    assert_ne!(kc_def.si_meta("si"), kc_t1.si_meta("si"));
    assert_ne!(kc_t1.si_item("si", 42), kc_t2.si_item("si", 42));

    // 13. Stream
    assert_ne!(kc_def.stream_meta("str"), kc_t1.stream_meta("str"));
    assert_ne!(
        kc_t1.stream_item("str", 100, 1),
        kc_t2.stream_item("str", 100, 1)
    );
    assert_ne!(
        kc_t1.stream_group_meta("str", "g1"),
        kc_t2.stream_group_meta("str", "g1")
    );
    assert_ne!(
        kc_t1.stream_consumer_meta("str", "g1", "c1"),
        kc_t2.stream_consumer_meta("str", "g1", "c1")
    );
    assert_ne!(
        kc_t1.stream_pel_item("str", "g1", 100, 1),
        kc_t2.stream_pel_item("str", "g1", 100, 1)
    );

    // 14. TDigest
    assert_ne!(kc_def.tdigest_meta("td"), kc_t1.tdigest_meta("td"));

    // 15. TimeSeries
    assert_ne!(kc_def.ts_meta("ts"), kc_t1.ts_meta("ts"));
    assert_ne!(kc_t1.ts_item("ts", 1000), kc_t2.ts_item("ts", 1000));

    Ok(())
}

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

    // 验证所有 15 种数据结构的 meta / data / subkey 的 extract_user_key 均能正确提取出原 key
    assert_eq!(
        kc_t.extract_user_key(kc_t.raw_key(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.hash_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.list_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.set_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.zset_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.bm_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.bf_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.cf_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.hll_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.json_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.si_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.stream_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.tdigest_meta(key).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.ts_meta(key).as_bytes()),
        Some(key.as_bytes())
    );

    // 验证子键提取
    assert_eq!(
        kc_t.extract_user_key(kc_t.bm_segment(key, 0).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.si_item(key, 1234).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.ts_item(key, 9999).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.bf_item(key, 1).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.cf_page(key, 0, 1).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.stream_item(key, 100, 2).as_bytes()),
        Some(key.as_bytes())
    );
    assert_eq!(
        kc_t.extract_user_key(kc_t.stream_group_meta(key, "grp").as_bytes()),
        Some(key.as_bytes())
    );

    // 验证系统元数据前缀（_meta:, _ttl:, _raft:）被严格过滤
    assert_eq!(kc_t.extract_user_key(b"_meta:ns_name:foo"), None);
    assert_eq!(kc_t.extract_user_key(b"_ttl:key"), None);
    assert_eq!(kc_t.extract_user_key(b"_raft:log:1"), None);
    assert!(!kc_t.is_key_in_ns(b"_meta:ns_token:xxx"));

    Ok(())
}

#[test]
fn test_transform_key_across_namespaces() -> Void {
    let kc_def = KeyComposer::new("default");
    let kc_db1 = KeyComposer::new("db1");
    let kc_db2 = KeyComposer::new("db2");

    // 1. 默认空间 -> db1
    let raw_def = b"my_string_key";
    assert_eq!(
        kc_def.transform_key_to_target(raw_def, &kc_db1),
        Some("\x00ns:db1:my_string_key".to_string())
    );
    let hash_def = b"\x00hm:my_hash";
    assert_eq!(
        kc_def.transform_key_to_target(hash_def, &kc_db1),
        Some("\x00ns:db1:hm:my_hash".to_string())
    );

    // 2. db1 -> 默认空间
    let raw_db1 = b"\x00ns:db1:my_string_key";
    assert_eq!(
        kc_db1.transform_key_to_target(raw_db1, &kc_def),
        Some("my_string_key".to_string())
    );
    let hash_db1 = b"\x00ns:db1:hm:my_hash";
    assert_eq!(
        kc_db1.transform_key_to_target(hash_db1, &kc_def),
        Some("\x00hm:my_hash".to_string())
    );

    // 3. db1 -> db2
    assert_eq!(
        kc_db1.transform_key_to_target(raw_db1, &kc_db2),
        Some("\x00ns:db2:my_string_key".to_string())
    );
    assert_eq!(
        kc_db1.transform_key_to_target(hash_db1, &kc_db2),
        Some("\x00ns:db2:hm:my_hash".to_string())
    );

    Ok(())
}

#[test]
fn test_select_db_and_namespace_mappings_suite() -> Void {
    assert_eq!(db_to_namespace(0).as_ref(), "default");
    assert_eq!(db_to_namespace(1).as_ref(), "db1");
    assert_eq!(db_to_namespace(15).as_ref(), "db15");
    assert_eq!(db_to_namespace(16).as_ref(), "db16");
    assert_eq!(db_to_namespace(63).as_ref(), "db63");

    assert_eq!(namespace_to_db("default"), 0);
    assert_eq!(namespace_to_db("__namespace"), 0);
    assert_eq!(namespace_to_db("0"), 0);
    assert_eq!(namespace_to_db("db0"), 0);
    assert_eq!(namespace_to_db("__db_0"), 0);
    assert_eq!(namespace_to_db("db1"), 1);
    assert_eq!(namespace_to_db("__db_5"), 5);
    assert_eq!(namespace_to_db("db15"), 15);
    assert_eq!(namespace_to_db("16"), 16);

    assert_eq!(db_to_namespace_kvrocks(0).as_ref(), "__namespace");
    assert_eq!(db_to_namespace_kvrocks(1).as_ref(), "db1");
    assert_eq!(db_to_namespace_kvrocks(15).as_ref(), "db15");
    assert_eq!(db_to_namespace_kvrocks(16).as_ref(), "db16");

    assert!(is_namespace_legal("valid_namespace").is_ok());
    assert!(is_namespace_legal("").is_err());
    assert!(is_namespace_legal(&"x".repeat(256)).is_err());

    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 {
    use wedb_embed::key_composer::{
        compose_namespace_key_with_slot, get_slot_id_from_key, is_default_namespace,
        is_default_namespace_bytes, is_namespace_legal_bytes,
    };

    // 默认命名空间判定
    assert!(is_default_namespace("default"));
    assert!(is_default_namespace("__namespace"));
    assert!(is_default_namespace("0"));
    assert!(is_default_namespace("db0"));
    assert!(is_default_namespace("__db_0"));
    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"__namespace"));
    assert!(is_default_namespace_bytes(b"0"));
    assert!(is_default_namespace_bytes(b"db0"));
    assert!(is_default_namespace_bytes(b"__db_0"));
    assert!(is_default_namespace_bytes(b""));
    assert!(!is_default_namespace_bytes(b"db1"));
    assert!(!is_default_namespace_bytes(b"tenant_1"));

    // 命名空间字符合法性判定
    assert!(is_namespace_legal_bytes(b"valid_ns").is_ok());
    assert!(is_namespace_legal_bytes(b"").is_err());
    assert!(is_namespace_legal_bytes(&[b'a'; 256]).is_err());
    assert!(is_namespace_legal_bytes(b"bad_char\x7f").is_err());
    assert!(is_namespace_legal_bytes(b"bad_char\xff").is_err());

    // 自动 Slot 编码构造
    let key = b"user:{1000}:profile";
    let slot_id = get_slot_id_from_key(key);
    let auto_key = compose_namespace_key_with_slot(b"tenant_x", key, true);
    let manual_key = compose_namespace_key(b"tenant_x", key, Some(slot_id));
    assert_eq!(auto_key, manual_key);

    let no_slot_auto = compose_namespace_key_with_slot(b"tenant_x", key, false);
    let no_slot_manual = compose_namespace_key(b"tenant_x", key, None);
    assert_eq!(no_slot_auto, no_slot_manual);

    Ok(())
}

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

    // 任意二进制 Key（包含非 UTF-8 字节 0xFF, 0xFE, 0x00）
    let binary_key = b"\xff\xfe\x00custom_bin_key";
    let raw_bin = kc_def.raw_key_bytes(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);

    // InternalKey::encode_to 缓冲区复用验证
    let ik = InternalKey::new(b"t1", b"bin_key", b"sub", 12345, Some(999));
    let mut buf = Vec::new();
    ik.encode_to(&mut buf);
    assert_eq!(buf, ik.encode());

    let decoded = InternalKey::decode(&buf, true).expect("decode failed");
    assert_eq!(decoded, ik);

    Ok(())
}