storage-engines 0.1.0

//! bplus-tree 功能演示
//!
//! 覆盖：基础 set/get/delete · 大 value(blob) · 批量 · multi_get
//!       计数器 incr / incr_with_ttl · 范围 scan / prefix / seek
//!       模糊搜索分页 · 元数据 · TTL

use storage_engines::bplus_tree::{kv_ops, MVCC, SearchQuery};
use std::{format, fs};
use std::path::{Path, PathBuf};
use std::time::Duration;

/// demo 数据目录（所有 data.db* 落在此目录下，不污染仓库根）
const DEMO_DIR: &str = "./data/bplus_tree";
const DEMO_DB_NAME: &str = "./data.db";

fn demo_db_path() -> PathBuf {
    Path::new(DEMO_DIR).join(DEMO_DB_NAME)
}

/// 清理某一库路径及其 sidecar（.wal / .dblwr / .lock / .blob / .freelist）
fn remove_db_files(db_path: &Path) {
    for suf in ["", ".dblwr", ".wal", ".freelist", ".lock", ".blob"] {
        let p = if suf.is_empty() {
            db_path.to_path_buf()
        } else {
            PathBuf::from(format!("{}{suf}", db_path.display()))
        };
        let _ = fs::remove_file(&p);
    }
}

/// 准备 demo 目录：创建 data/，清空其中旧库，并尽量删掉仓库根上遗留的旧文件
fn prepare_demo_dir() -> PathBuf {
    let _ = fs::create_dir_all(DEMO_DIR);
    let db = demo_db_path();
    remove_db_files(&db);

    // 根目录历史遗留：旧 demo 的 data.db*、一次性导入的 from_lsm.*、临时快照
    remove_db_files(Path::new("data.db"));
    remove_db_files(Path::new("from_lsm.db"));
    for junk in [
        "snap_mig.bin",
        "snapshot.bin",
        "snapshot.json",
        "verify.bin",
    ] {
        let _ = fs::remove_file(junk);
    }
    db
}

fn show_key(k: &[u8]) -> String {
    if k.len() > 40 {
        format!("{}...({}B)", String::from_utf8_lossy(&k[..24]), k.len())
    } else {
        format!("{:?}", String::from_utf8_lossy(k))
    }
}

fn show_val(v: Option<&[u8]>) -> String {
    match v {
        None => "<none>".into(),
        Some(val) if val.len() > 32 => {
            format!("{}...({}B)", String::from_utf8_lossy(&val[..16]), val.len())
        }
        Some(val) => format!("{:?}", String::from_utf8_lossy(val)),
    }
}

#[test]
fn test_bplus_tree() {
    let db_path = prepare_demo_dir();
    let db_path_str = db_path.to_string_lossy().into_owned();
    println!("demo 数据目录: {DEMO_DIR}/  库文件: {db_path_str}");

    let large_key = {
        let mut k = b"user:profile:".to_vec();
        k.extend(std::iter::repeat(b'x').take(200));
        k
    };
    let large_val = {
        let mut v = b"BLOB:".to_vec();
        v.extend(std::iter::repeat(b'A').take(4_000));
        v
    };

    // ══════════════════════════════════════════════════════════════
    // 1. 基础 CRUD + 大 KV + 批量 / 扫描 / 搜索 / TTL
    // ══════════════════════════════════════════════════════════════
    {
        let mvcc = MVCC::open(&db_path, 16, 64);

        // ── 1.1 基础 set / get / delete / 覆盖 ──
        {
            println!("\n========== 1. 基础 set / get / delete ==========");
            let tx = mvcc.begin_transaction();

            assert!(tx.set(b"test", b"test1".to_vec()));
            println!("[set]  test = test1");

            assert!(tx.set(&large_key, large_val.clone()));
            println!(
                "[set]  large_key(len={}) / large_val(len={}) → blob",
                large_key.len(),
                large_val.len()
            );

            assert!(tx.set(b"to_delete", b"will_gone".to_vec()));
            assert_eq!(tx.get(b"test").as_deref(), Some(b"test1".as_slice()));
            let got_big = tx.get(&large_key).expect("大 value");
            assert_eq!(got_big, large_val);
            println!(
                "[get]  large 读回 len={} 前缀 {:?}",
                got_big.len(),
                String::from_utf8_lossy(&got_big[..8])
            );

            assert!(tx.delete(b"to_delete"));
            assert_eq!(tx.get(b"to_delete"), None);
            println!("[del]  to_delete → get=None");

            assert!(tx.set(b"test", b"test2".to_vec()));
            assert_eq!(tx.get(b"test").as_deref(), Some(b"test2".as_slice()));
            println!("[set]  test 覆盖为 test2");

            tx.commit();
            println!("commit ok");
        }

        // ── 1.2 批量写入 / multi_get / batch_delete ──
        {
            println!("\n========== 2. 批量 batch_set / multi_get / batch_delete ==========");
            let tx = mvcc.begin_transaction();

            let items: Vec<(Vec<u8>, Vec<u8>)> = (0..8u8)
                .map(|i| {
                    (
                        format!("order:{i:02}").into_bytes(),
                        format!("order-val-{i}").into_bytes(),
                    )
                })
                .collect();
            tx.batch_set(&items).expect("batch_set");
            println!("[batch_set] 写入 {} 条 order:*", items.len());

            let keys: Vec<Vec<u8>> = items.iter().map(|(k, _)| k.clone()).collect();
            let got = tx.multi_get(&keys);
            println!(
                "[multi_get] 命中 {}/{}",
                got.iter().filter(|v| v.is_some()).count(),
                got.len()
            );
            for (k, v) in keys.iter().zip(got.iter()) {
                println!("  {} => {}", show_key(k), show_val(v.as_deref()));
            }

            // 删掉后半
            let del: Vec<Vec<u8>> = (4..8u8)
                .map(|i| format!("order:{i:02}").into_bytes())
                .collect();
            tx.batch_delete(&del).expect("batch_delete");
            println!("[batch_delete] 删除 order:04..07");
            assert_eq!(tx.get(b"order:04"), None);
            assert_eq!(tx.get(b"order:03").as_deref(), Some(b"order-val-3".as_slice()));

            // 再插一些用于扫描/搜索
            let more: Vec<(Vec<u8>, Vec<u8>)> = vec![
                (b"apple".to_vec(), b"fruit".to_vec()),
                (b"apply".to_vec(), b"verb".to_vec()),
                (b"banana".to_vec(), b"yellow".to_vec()),
                (b"user:01".to_vec(), b"alice".to_vec()),
                (b"user:02".to_vec(), b"bob".to_vec()),
                (b"user:10".to_vec(), b"carol".to_vec()),
                (b"zebra".to_vec(), b"z".to_vec()),
            ];
            tx.batch_set(&more).expect("batch more");
            tx.commit();
            println!("commit ok");
        }

        // ── 1.3 范围 scan / prefix / seek ──
        {
            println!("\n========== 3. 范围 scan / prefix_scan / seek / seek_prev ==========");
            let tx = mvcc.begin_transaction();

            let range = tx.scan(Some(b"order:"), Some(b"order:\x7f"));
            println!("[scan] [order:, order:\\x7f) → {} 条", range.len());
            for r in &range {
                println!(
                    "  {} => {}",
                    show_key(&r.key),
                    show_val(r.value.as_deref())
                );
            }

            let pref = tx.prefix_scan(b"user:");
            println!("[prefix_scan] user: → {} 条", pref.len());
            for r in &pref {
                println!("  {}", show_key(&r.key));
            }

            let rev = tx.reverse_scan(Some(b"a"), Some(b"c"));
            println!("[reverse_scan] [a,c) 逆序 首条 = {}", show_key(&rev[0].key));

            if let Some(s) = tx.seek(b"user:02") {
                println!("[seek]  >= user:02 → {}", show_key(&s.key));
            }
            if let Some(s) = tx.seek(b"user:05") {
                println!("[seek]  >= user:05 → {} (跳到 user:10)", show_key(&s.key));
            }
            if let Some(p) = tx.seek_prev(b"user:05") {
                println!("[seek_prev] <= user:05 → {}", show_key(&p.key));
            }

            println!(
                "[exists] order:01={}  gone={}",
                tx.exists(b"order:01"),
                tx.exists(b"order:04")
            );
            println!("[key_count] prefix order: = {}", tx.key_count(b"order:"));

            if let Some(m) = tx.get_meta(b"order:01") {
                println!(
                    "[get_meta] order:01 len={:?} ver={} expired={} deleted={}",
                    m.value_len, m.version, m.expired, m.deleted
                );
            }
            tx.commit();
        }

        // ── 1.4 模糊搜索 + 分页 ──
        {
            println!("\n========== 4. 模糊 / 前缀搜索 + 分页 ==========");

            let page = mvcc.search_keys(&SearchQuery::contains(b"pp", 0, 10));
            println!(
                "[search contains \"pp\"] total={} page={}/{} items={}",
                page.total,
                page.page + 1,
                page.total_pages.max(1),
                page.items.len()
            );
            for r in &page.items {
                println!(
                    "  {} => {}",
                    show_key(&r.key),
                    show_val(r.value.as_deref())
                );
            }

            println!("[search prefix \"user:\" page_size=2]");
            let mut page_idx = 0usize;
            loop {
                let p = mvcc.search_keys(&SearchQuery::prefix(b"user:", page_idx, 2));
                println!(
                    "  [page {}/{}] count={} has_next={}",
                    p.page + 1,
                    p.total_pages.max(1),
                    p.items.len(),
                    p.has_next()
                );
                for r in &p.items {
                    println!("    {}", show_key(&r.key));
                }
                if !p.has_next() {
                    break;
                }
                page_idx += 1;
            }
        }

        // ── 1.5 计数器 incr / incr_with_ttl ──
        {
            println!("\n========== 5. 计数器 incr / decr / incr_with_ttl ==========");
            let tx = mvcc.begin_transaction();

            // 不存在从 0 起
            let n = tx.incr(b"seq", 1).expect("incr");
            println!("[incr] seq +1 → {n}");
            let n = tx.incr(b"seq", 10).expect("incr");
            println!("[incr] seq +10 → {n}");
            let n = tx.decr(b"seq", 3).expect("decr");
            println!("[decr] seq -3 → {n}");
            assert_eq!(tx.get(b"seq"), Some(b"8".to_vec()));
            println!("[get]  seq = {:?}", String::from_utf8_lossy(&tx.get(b"seq").unwrap()));

            // 带 TTL 的计数器（序列号 / 限流计数等）
            let n = tx
                .incr_with_ttl(b"hits", 1, Duration::from_secs(120))
                .expect("incr_with_ttl");
            println!("[incr_with_ttl] hits +1 → {n} (TTL 120s)");
            let n = tx
                .incr_with_ttl(b"hits", 5, Duration::from_secs(120))
                .expect("incr_with_ttl");
            println!("[incr_with_ttl] hits +5 → {n}");
            assert_eq!(tx.get(b"hits"), Some(b"6".to_vec()));
            match tx.get_ttl(b"hits") {
                Some(Some(d)) => println!("[get_ttl] hits 剩余 ≈ {}s", d.as_secs()),
                other => println!("[get_ttl] hits = {other:?}"),
            }

            // 非整数 value 会失败
            assert!(tx.set(b"not_num", kv_ops::pack_plain(b"abc")));
            match tx.incr(b"not_num", 1) {
                Err(e) => println!("[incr] not_num +1 → Err({e}) （预期 NotInteger）"),
                Ok(v) => panic!("should fail, got {v}"),
            }

            tx.commit();
            println!("commit ok");
        }

        // ── 1.6 TTL ──
        {
            println!("\n========== 6. TTL set_with_ttl / get_ttl / persist / purge ==========");
            let tx = mvcc.begin_transaction();

            assert!(tx.set_with_ttl(b"cache:hot", b"payload".to_vec(), Duration::from_secs(120)));
            assert!(tx.set(b"cache:forever", kv_ops::pack_plain(b"keep")));

            // 直接写一条「已过期」
            let stale = kv_ops::pack_ttl(b"old", kv_ops::now_unix_secs().saturating_sub(5));
            assert!(tx.set(b"cache:stale", stale));

            println!(
                "[get] cache:hot={:?}  cache:stale={:?} (过期应 None)",
                tx.get(b"cache:hot").map(|v| String::from_utf8_lossy(&v).into_owned()),
                tx.get(b"cache:stale")
            );
            println!(
                "[exists] hot={} stale={} forever={}",
                tx.exists(b"cache:hot"),
                tx.exists(b"cache:stale"),
                tx.exists(b"cache:forever")
            );

            match tx.get_ttl(b"cache:hot") {
                Some(Some(d)) => println!("[get_ttl] cache:hot 剩余 ≈ {}s", d.as_secs()),
                other => println!("[get_ttl] cache:hot = {other:?}"),
            }
            match tx.get_ttl(b"cache:forever") {
                Some(None) => println!("[get_ttl] cache:forever = 永久"),
                other => println!("[get_ttl] cache:forever = {other:?}"),
            }

            assert!(tx.refresh_ttl(b"cache:hot", Duration::from_secs(300)));
            println!("[refresh_ttl] cache:hot → 300s");
            assert!(tx.persist(b"cache:hot"));
            println!("[persist] cache:hot → 永久, get_ttl={:?}", tx.get_ttl(b"cache:hot"));

            let n = tx.purge_expired();
            println!("[purge_expired] 清理 {n} 条过期 key");
            assert!(!tx.exists(b"cache:stale"));

            if let Some(m) = tx.get_meta(b"cache:hot") {
                println!(
                    "[get_meta] cache:hot len={:?} expire={:?} expired={}",
                    m.value_len, m.expire_unix_secs, m.expired
                );
            }
            tx.commit();
        }

        let blob_path = PathBuf::from(format!("{db_path_str}.blob"));
        if let Ok(meta) = fs::metadata(&blob_path) {
            println!(
                "\n[blob] {} = {} B（大 value 外置）",
                blob_path.display(),
                meta.len()
            );
        }
    } // drop → 释放锁

    // ══════════════════════════════════════════════════════════════
    // 2. 模拟重启：校验持久化
    // ══════════════════════════════════════════════════════════════
    {
        println!("\n========== 7. 重启后校验 ==========");
        let mvcc = MVCC::open(&db_path, 16, 64);
        let tx = mvcc.begin_transaction();

        assert_eq!(tx.get(b"test").as_deref(), Some(b"test2".as_slice()));
        println!(
            "[reopen] test = {:?}",
            String::from_utf8_lossy(&tx.get(b"test").unwrap())
        );
        assert_eq!(tx.get(b"to_delete"), None);
        println!("[reopen] to_delete 仍不存在");

        let got_big = tx.get(&large_key).expect("重启后大 value");
        assert_eq!(got_big, large_val);
        println!("[reopen] large_key len={} OK", got_big.len());

        // 批量/扫描数据仍在
        assert_eq!(tx.get(b"order:01").as_deref(), Some(b"order-val-1".as_slice()));
        assert_eq!(tx.get(b"order:04"), None);
        assert_eq!(tx.key_count(b"order:"), 4);
        println!("[reopen] order:* 存活 {} 条", tx.key_count(b"order:"));

        // 计数器仍在
        assert_eq!(tx.get(b"seq"), Some(b"8".to_vec()));
        let n = tx.incr(b"seq", 1).expect("reopen incr");
        assert_eq!(n, 9);
        println!("[reopen] seq 仍为 8，再 incr → {n}");
        assert_eq!(tx.get(b"hits"), Some(b"6".to_vec()));
        println!("[reopen] hits = 6（带 TTL 计数器仍在）");

        assert!(tx.exists(b"cache:hot"));
        assert!(!tx.exists(b"cache:stale"));
        println!("[reopen] TTL: hot 在 / stale 已 purge");

        // 搜索仍可用
        let page = mvcc.search_keys(&SearchQuery::contains(b"pp", 0, 5));
        println!(
            "[reopen] search \"pp\" → {:?}",
            page.items
                .iter()
                .map(|r| String::from_utf8_lossy(&r.key).into_owned())
                .collect::<Vec<_>>()
        );

        assert!(tx.delete(&large_key));
        assert_eq!(tx.get(&large_key), None);
        println!("[reopen] 删除 large_key 后 get=None");
        tx.commit();
    }

    println!("\n—— 磁盘文件（均在 {DEMO_DIR}/ 下）——");
    for suf in ["", ".wal", ".dblwr", ".lock", ".blob", ".freelist"] {
        let p = if suf.is_empty() {
            db_path.clone()
        } else {
            PathBuf::from(format!("{db_path_str}{suf}"))
        };
        if let Ok(m) = fs::metadata(&p) {
            println!("  {:28} {:>8} B", p.display(), m.len());
        }
    }
    println!("\ndemo 完成。数据文件在 `./{DEMO_DIR}/`，不会再写到仓库根目录。");
}