trine-kv 0.6.0

use super::{BranchLifecycle, RegistryEntry, data_bucket, fork_checkpoint, registry_bucket};
use crate::storage::{
    StorageObjectKind,
    fault_injection::{StorageFaultGuard, StorageFaultPoint},
};
use crate::{Db, DbOptions, Error, InMemoryObjectStore, KeyRange};

fn memory_db() -> Db {
    Db::open_sync(DbOptions::memory()).expect("open in-memory db")
}

#[test]
fn object_store_branch_has_a_complete_async_lifecycle() {
    futures::executor::block_on(async {
        let client = std::sync::Arc::new(InMemoryObjectStore::new());
        let db = Db::open_object_store(client.clone(), DbOptions::object_store())
            .await
            .expect("object-store database opens");
        let data = db.bucket("data").await.expect("data bucket opens");
        data.put(b"base", b"root")
            .await
            .expect("root write commits");
        let fork = db.latest_read_version();
        let branch_blob = vec![7_u8; crate::BucketOptions::DEFAULT_BLOB_THRESHOLD_BYTES + 1];

        db.create_branch("dev", fork).await.expect("branch creates");
        let mut branch = db.open_branch("dev").await.expect("branch opens");
        branch
            .put("data", b"base", b"branch")
            .await
            .expect("branch override commits");
        branch
            .put("data", b"private", branch_blob.clone())
            .await
            .expect("branch insert commits");
        assert_eq!(
            branch.get("data", b"base").await.expect("branch reads"),
            Some(b"branch".to_vec()),
        );
        let mut range = branch
            .range("data", &KeyRange::all())
            .await
            .expect("branch range opens");
        let mut rows = Vec::new();
        while let Some(row) = range.next().await.expect("branch row reads") {
            rows.push(row);
        }
        assert_eq!(
            rows.into_iter()
                .map(|row| (row.key, row.value))
                .collect::<Vec<_>>(),
            vec![
                (b"base".to_vec(), b"branch".to_vec()),
                (b"private".to_vec(), branch_blob.clone()),
            ],
        );
        drop(branch);
        assert_eq!(
            db.list_branches().await.expect("branches list"),
            vec!["dev".to_owned()],
        );
        assert_eq!(
            db.branch_info("dev")
                .await
                .expect("branch info reads")
                .expect("branch is active")
                .fork(),
            fork,
        );
        db.close().await.expect("database closes");
        drop(db);

        let reopened = Db::open_object_store(client, DbOptions::object_store())
            .await
            .expect("object-store database reopens");
        let branch = reopened.open_branch("dev").await.expect("branch reopens");
        assert_eq!(
            branch
                .get("data", b"private")
                .await
                .expect("reopened branch reads"),
            Some(branch_blob.clone()),
        );
        let mut rows = branch
            .range("data", &KeyRange::all())
            .await
            .expect("reopened branch range opens");
        assert_eq!(
            rows.next()
                .await
                .expect("first branch row reads")
                .expect("first branch row exists")
                .value,
            b"branch",
        );
        assert_eq!(
            rows.next()
                .await
                .expect("blob branch row reads")
                .expect("blob branch row exists")
                .value,
            branch_blob,
        );
        drop(branch);
        reopened.delete_branch("dev").await.expect("branch deletes");
        assert!(
            reopened
                .list_branches()
                .await
                .expect("branches list")
                .is_empty(),
        );
        reopened.close().await.expect("reopened database closes");
    });
}

#[test]
fn registry_entries_require_current_format_marker() {
    assert!(matches!(
        RegistryEntry::decode(&[0; 32]),
        Err(Error::Corruption { .. })
    ));
}

#[test]
fn branch_reads_parent_then_shadows_with_local_writes() {
    let db = memory_db();
    let bucket = db.bucket_sync("data").expect("bucket");
    bucket.put_sync(b"k1".to_vec(), b"v1".to_vec()).expect("p1");
    bucket.put_sync(b"k2".to_vec(), b"v2".to_vec()).expect("p2");

    let mut branch = db.branch_from_latest().expect("branch");
    assert_eq!(
        branch.get_sync("data", b"k1").expect("get"),
        Some(b"v1".to_vec())
    );

    branch
        .put_sync("data", b"k1", b"v1-branch".to_vec())
        .expect("put");
    branch.delete_sync("data", b"k2").expect("delete");
    assert_eq!(
        branch.get_sync("data", b"k1").expect("get"),
        Some(b"v1-branch".to_vec())
    );
    assert_eq!(branch.get_sync("data", b"k2").expect("get"), None);

    // The parent is untouched.
    assert_eq!(bucket.get_sync(b"k1").expect("get"), Some(b"v1".to_vec()));
    assert_eq!(bucket.get_sync(b"k2").expect("get"), Some(b"v2".to_vec()));
}

#[test]
fn branch_pins_its_fork_while_the_parent_diverges() {
    let db = memory_db();
    let bucket = db.bucket_sync("data").expect("bucket");
    bucket.put_sync(b"k".to_vec(), b"v1".to_vec()).expect("p1");

    let branch = db.branch_from_latest().expect("branch");
    bucket.put_sync(b"k".to_vec(), b"v2".to_vec()).expect("p2");

    assert_eq!(
        branch.get_sync("data", b"k").expect("get"),
        Some(b"v1".to_vec()),
        "the branch stays frozen at its fork while the parent diverges"
    );
    assert_eq!(bucket.get_sync(b"k").expect("get"), Some(b"v2".to_vec()));
}

#[test]
fn branch_at_a_retained_past_version_time_travels() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(8))
        .expect("open with retention");
    let bucket = db.bucket_sync("data").expect("bucket");
    bucket.put_sync(b"k".to_vec(), b"v1".to_vec()).expect("p1");
    let v1 = db.latest_read_version();
    bucket.put_sync(b"k".to_vec(), b"v2".to_vec()).expect("p2");

    let old = db.branch_at(v1).expect("branch at v1");
    assert_eq!(
        old.get_sync("data", b"k").expect("get"),
        Some(b"v1".to_vec())
    );
    let now = db.branch_from_latest().expect("branch now");
    assert_eq!(
        now.get_sync("data", b"k").expect("get"),
        Some(b"v2".to_vec())
    );
}

#[test]
fn ephemeral_branch_range_merges_overlay_over_parent() {
    let db = memory_db();
    let bucket = db.bucket_sync("data").expect("bucket");
    for (k, v) in [(b"a", b"1"), (b"b", b"2"), (b"c", b"3")] {
        bucket.put_sync(k.to_vec(), v.to_vec()).expect("seed");
    }

    let mut branch = db.branch_from_latest().expect("branch");
    branch
        .put_sync("data", b"b", b"2-branch".to_vec())
        .expect("override b");
    branch.delete_sync("data", b"c").expect("delete c");
    branch.put_sync("data", b"d", b"4".to_vec()).expect("add d");

    let rows = branch.range_sync("data", &KeyRange::all()).expect("range");
    let got: Vec<(Vec<u8>, Vec<u8>)> = rows
        .map(|kv| {
            let kv = kv.expect("row");
            (kv.key, kv.value)
        })
        .collect();
    assert_eq!(
        got,
        vec![
            (b"a".to_vec(), b"1".to_vec()),
            (b"b".to_vec(), b"2-branch".to_vec()),
            (b"d".to_vec(), b"4".to_vec()),
        ]
    );
}

#[test]
fn durable_branch_persists_writes_and_shadows_parent() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    let bucket = db.bucket_sync("data").expect("bucket");
    bucket
        .put_sync(b"k1".to_vec(), b"parent".to_vec())
        .expect("p1");
    bucket
        .put_sync(b"k2".to_vec(), b"parent".to_vec())
        .expect("p2");

    db.create_branch_sync("dev", db.latest_read_version())
        .expect("create");
    {
        let mut dev = db.open_branch_sync("dev").expect("open");
        dev.put_sync("data", b"k1", b"dev".to_vec()).expect("put");
        dev.delete_sync("data", b"k2").expect("delete");
    }

    // A freshly opened handle sees the persisted branch writes; the parent is
    // untouched.
    let dev = db.open_branch_sync("dev").expect("reopen");
    assert_eq!(
        dev.get_sync("data", b"k1").expect("get"),
        Some(b"dev".to_vec())
    );
    assert_eq!(
        dev.get_sync("data", b"k2").expect("get"),
        None,
        "branch tombstone hides parent"
    );
    assert_eq!(dev.get_sync("data", b"k3").expect("get"), None);
    assert_eq!(
        bucket.get_sync(b"k1").expect("get"),
        Some(b"parent".to_vec())
    );
    assert_eq!(
        bucket.get_sync(b"k2").expect("get"),
        Some(b"parent".to_vec())
    );

    assert_eq!(
        db.list_branches_sync().expect("list"),
        vec!["dev".to_string()]
    );
    assert!(dev.is_durable());
}

#[test]
fn durable_branch_storage_names_cannot_collide_at_component_boundaries() {
    let db = memory_db();
    let first_bucket = "b\u{1}c";
    let second_branch = "a\u{1}b";
    db.bucket_sync(first_bucket).expect("first parent bucket");
    db.bucket_sync("c").expect("second parent bucket");
    let fork = db.latest_read_version();
    db.create_branch_sync("a", fork).expect("first branch");
    db.create_branch_sync(second_branch, fork)
        .expect("second branch");

    let mut first = db.open_branch_sync("a").expect("open first branch");
    first
        .put_sync(first_bucket, b"k", b"first".to_vec())
        .expect("write first branch");
    let mut second = db
        .open_branch_sync(second_branch)
        .expect("open second branch");
    second
        .put_sync("c", b"k", b"second".to_vec())
        .expect("write second branch");

    drop(first);
    drop(second);
    assert_eq!(
        db.open_branch_sync("a")
            .expect("reopen first")
            .get_sync(first_bucket, b"k")
            .expect("read first"),
        Some(b"first".to_vec())
    );
    assert_eq!(
        db.open_branch_sync(second_branch)
            .expect("reopen second")
            .get_sync("c", b"k")
            .expect("read second"),
        Some(b"second".to_vec())
    );

    db.delete_branch_sync("a").expect("delete first branch");
    assert_eq!(
        db.open_branch_sync(second_branch)
            .expect("second survives first deletion")
            .get_sync("c", b"k")
            .expect("read surviving branch"),
        Some(b"second".to_vec())
    );
}

#[test]
fn durable_branch_write_commits_data_and_registry_together() {
    let dir =
        std::env::temp_dir().join(format!("trine-branch-atomic-write-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    let db = Db::open_sync(&dir).expect("open");
    db.bucket_sync("data").expect("bucket");
    db.create_branch_sync("dev", db.latest_read_version())
        .expect("create");
    let mut dev = db.open_branch_sync("dev").expect("open");

    let fault = StorageFaultGuard::install(
        &dir,
        StorageFaultPoint::WalAppend,
        Some(StorageObjectKind::Wal),
        1,
    );
    assert!(dev.put_sync("data", b"k", b"v".to_vec()).is_err());
    assert_eq!(fault.calls(), 1);
    drop(fault);
    drop(dev);
    drop(db);

    let db = Db::open_sync(&dir).expect("reopen after uncertain WAL outcome");
    let registry = futures::executor::block_on(db.read_registry("dev"))
        .expect("registry")
        .expect("entry");
    assert!(registry.written_buckets.is_empty());
    assert_eq!(
        db.internal_bucket_sync(data_bucket("dev", "data"))
            .expect("data bucket")
            .get_sync(b"k")
            .expect("data read"),
        None
    );

    drop(db);

    // An append error is fail-stop because the OS may have written a partial
    // frame. Reopen performs recovery before admitting any later write.
    let db = Db::open_sync(&dir).expect("database reopens after WAL failure");
    let mut reopened = db.open_branch_sync("dev").expect("branch reopens");
    reopened
        .put_sync("data", b"k", b"v".to_vec())
        .expect("write succeeds after recovery");
    assert_eq!(
        reopened.get_sync("data", b"k").expect("read"),
        Some(b"v".to_vec())
    );
    drop(reopened);
    drop(db);
    std::fs::remove_dir_all(dir).expect("test database removes");
}

#[test]
fn deleting_marker_hides_branch_and_delete_resumes() {
    let db = memory_db();
    db.bucket_sync("data").expect("bucket");
    db.create_branch_sync("dev", db.latest_read_version())
        .expect("create");
    let mut stale = db.open_branch_sync("dev").expect("open");
    stale
        .put_sync("data", b"k", b"v".to_vec())
        .expect("branch data");

    let mut entry = futures::executor::block_on(db.read_registry("dev"))
        .expect("registry")
        .expect("entry");
    entry.lifecycle = BranchLifecycle::Deleting;
    let encoded = entry.encode().expect("encode interrupted delete marker");
    db.internal_bucket_sync(registry_bucket())
        .expect("registry bucket")
        .put_sync(b"dev".to_vec(), encoded)
        .expect("persist interrupted delete marker");

    assert!(db.list_branches_sync().expect("list").is_empty());
    assert!(db.branch_info_sync("dev").expect("info").is_none());
    assert!(db.open_branch_sync("dev").is_err());
    assert!(stale.get_sync("data", b"k").is_err());
    assert!(stale.put_sync("data", b"k2", b"v2".to_vec()).is_err());

    db.delete_branch_sync("dev").expect("delete resumes");
    db.delete_branch_sync("dev")
        .expect("completed delete is idempotent");
    assert!(
        futures::executor::block_on(db.read_registry("dev"))
            .expect("registry")
            .is_none()
    );
}

#[test]
fn stale_handle_cannot_write_recreated_branch_generation() {
    let db = memory_db();
    db.bucket_sync("data").expect("bucket");
    let fork = db.latest_read_version();
    db.create_branch_sync("dev", fork).expect("create");
    let mut stale = db.open_branch_sync("dev").expect("open old generation");
    db.delete_branch_sync("dev").expect("delete old generation");
    db.create_branch_sync("dev", fork).expect("recreate");

    assert!(
        stale.put_sync("data", b"stale", b"value".to_vec()).is_err(),
        "an old handle must not mutate a replacement with the same name and fork"
    );
    assert_eq!(
        db.open_branch_sync("dev")
            .expect("open replacement")
            .get_sync("data", b"stale")
            .expect("read replacement"),
        None
    );
}

#[test]
fn durable_branch_survives_reopen_with_default_retention() {
    let dir = std::env::temp_dir().join(format!("trine-branch-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    // Default retention keeps only the latest version, yet the branch's fork
    // checkpoint pins its fork history durably — so it reopens with no manual
    // retention configuration (slice 3).
    {
        let db = Db::open_sync(&dir).expect("open");
        let bucket = db.bucket_sync("data").expect("bucket");
        bucket
            .put_sync(b"k".to_vec(), b"parent".to_vec())
            .expect("seed");
        db.create_branch_sync("dev", db.latest_read_version())
            .expect("create");
        let mut dev = db.open_branch_sync("dev").expect("open");
        dev.put_sync("data", b"k", b"dev".to_vec()).expect("put");
        db.flush_sync().expect("flush");
    }
    // Reopen: the durable branch, its fork, and its writes all survive.
    let db = Db::open_sync(&dir).expect("reopen");
    assert_eq!(
        db.list_branches_sync().expect("list"),
        vec!["dev".to_string()]
    );
    let dev = db.open_branch_sync("dev").expect("open after reopen");
    assert_eq!(
        dev.get_sync("data", b"k").expect("get"),
        Some(b"dev".to_vec())
    );
    let _ = std::fs::remove_dir_all(&dir);
}

#[test]
fn durable_branch_fork_is_pinned_against_aggressive_gc() {
    // keep_last_read_versions(1) retains only the latest version, so without a
    // pin the fork would expire after the next parent write.
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(1)).expect("open");
    let bucket = db.bucket_sync("data").expect("bucket");
    bucket
        .put_sync(b"k".to_vec(), b"forked".to_vec())
        .expect("seed");
    db.create_branch_sync("dev", db.latest_read_version())
        .expect("create");

    // Hammer the parent well past the fork; the fork checkpoint keeps that
    // history retained.
    for i in 0..50 {
        bucket
            .put_sync(b"k".to_vec(), format!("v{i}").into_bytes())
            .expect("churn");
    }

    let dev = db.open_branch_sync("dev").expect("fork still openable");
    assert_eq!(
        dev.get_sync("data", b"k").expect("get"),
        Some(b"forked".to_vec()),
        "the branch still reads its fork value despite aggressive parent GC"
    );
}

#[test]
fn delete_branch_releases_the_fork_pin() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(1)).expect("open");
    let bucket = db.bucket_sync("data").expect("bucket");
    bucket
        .put_sync(b"k".to_vec(), b"forked".to_vec())
        .expect("seed");
    let fork = db.latest_read_version();
    db.create_branch_sync("dev", fork).expect("create");

    // While the branch lives, the fork stays pinned even past parent writes.
    bucket
        .put_sync(b"k".to_vec(), b"after".to_vec())
        .expect("write");
    assert!(
        db.branch_at(fork).is_ok(),
        "fork pinned while branch exists"
    );

    db.delete_branch_sync("dev").expect("delete");
    assert!(
        db.open_branch_sync("dev").is_err(),
        "deleted branch cannot be opened"
    );
    // The pin is released: with only the latest retained, a further write
    // pushes the floor past the fork, which is now expired.
    bucket
        .put_sync(b"k".to_vec(), b"later".to_vec())
        .expect("write");
    assert!(
        db.branch_at(fork).is_err(),
        "the fork is no longer pinned after the branch is deleted"
    );
}

#[test]
fn durable_branch_range_merges_over_parent() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    let bucket = db.bucket_sync("data").expect("bucket");
    for (k, v) in [(b"a", b"1"), (b"b", b"2"), (b"c", b"3")] {
        bucket.put_sync(k.to_vec(), v.to_vec()).expect("seed");
    }
    db.create_branch_sync("dev", db.latest_read_version())
        .expect("create");
    let mut dev = db.open_branch_sync("dev").expect("open");
    dev.put_sync("data", b"b", b"2-dev".to_vec())
        .expect("override");
    dev.delete_sync("data", b"c").expect("delete");
    dev.put_sync("data", b"d", b"4".to_vec()).expect("add");

    let rows = dev.range_sync("data", &KeyRange::all()).expect("range");
    let got: Vec<(Vec<u8>, Vec<u8>)> = rows
        .map(|kv| {
            let kv = kv.expect("row");
            (kv.key, kv.value)
        })
        .collect();
    assert_eq!(
        got,
        vec![
            (b"a".to_vec(), b"1".to_vec()),
            (b"b".to_vec(), b"2-dev".to_vec()),
            (b"d".to_vec(), b"4".to_vec()),
        ]
    );
}

#[test]
fn branch_of_branch_reads_through_the_whole_chain() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    let bucket = db.bucket_sync("data").expect("bucket");
    bucket
        .put_sync(b"base".to_vec(), b"root".to_vec())
        .expect("seed");
    bucket
        .put_sync(b"shared".to_vec(), b"root".to_vec())
        .expect("seed");

    // a forks root and overrides `shared`, adds `a-only`.
    db.create_branch_sync("a", db.latest_read_version())
        .expect("create a");
    {
        let mut a = db.open_branch_sync("a").expect("open a");
        a.put_sync("data", b"shared", b"a".to_vec())
            .expect("a override");
        a.put_sync("data", b"a-only", b"a".to_vec()).expect("a add");
    }

    // b forks a, overrides `shared` again, adds `b-only`, deletes `a-only`.
    db.create_branch_from_sync("b", "a")
        .expect("create b from a");
    let mut b = db.open_branch_sync("b").expect("open b");
    b.put_sync("data", b"shared", b"b".to_vec())
        .expect("b override");
    b.put_sync("data", b"b-only", b"b".to_vec()).expect("b add");
    b.delete_sync("data", b"a-only").expect("b delete a-only");

    // b sees: its own writes, then a's, then root's, in that precedence.
    assert_eq!(
        b.get_sync("data", b"shared").expect("get"),
        Some(b"b".to_vec())
    );
    assert_eq!(
        b.get_sync("data", b"b-only").expect("get"),
        Some(b"b".to_vec())
    );
    assert_eq!(
        b.get_sync("data", b"a-only").expect("get"),
        None,
        "b deleted a's key"
    );
    assert_eq!(
        b.get_sync("data", b"base").expect("get"),
        Some(b"root".to_vec()),
        "falls through a (untouched) to the root"
    );

    // The range view merges the whole chain.
    let rows = b.range_sync("data", &KeyRange::all()).expect("range");
    let got: Vec<(Vec<u8>, Vec<u8>)> = rows
        .map(|kv| {
            let kv = kv.expect("row");
            (kv.key, kv.value)
        })
        .collect();
    assert_eq!(
        got,
        vec![
            (b"b-only".to_vec(), b"b".to_vec()),
            (b"base".to_vec(), b"root".to_vec()),
            (b"shared".to_vec(), b"b".to_vec()),
        ]
    );

    // a is unaffected by b.
    let a = db.open_branch_sync("a").expect("reopen a");
    assert_eq!(
        a.get_sync("data", b"shared").expect("get"),
        Some(b"a".to_vec())
    );
    assert_eq!(
        a.get_sync("data", b"a-only").expect("get"),
        Some(b"a".to_vec())
    );
    assert_eq!(a.get_sync("data", b"b-only").expect("get"), None);
}

#[test]
fn branch_of_branch_is_frozen_when_its_parent_advances() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    db.bucket_sync("data").expect("bucket");
    db.create_branch_sync("a", db.latest_read_version())
        .expect("create a");
    {
        let mut a = db.open_branch_sync("a").expect("open a");
        a.put_sync("data", b"k", b"a1".to_vec()).expect("a write");
    }
    // b forks a at this point.
    db.create_branch_from_sync("b", "a").expect("create b");
    // a keeps writing after the fork.
    {
        let mut a = db.open_branch_sync("a").expect("reopen a");
        a.put_sync("data", b"k", b"a2".to_vec())
            .expect("a write later");
    }
    // b sees a's value as of the fork, not a's later write.
    let b = db.open_branch_sync("b").expect("open b");
    assert_eq!(b.get_sync("data", b"k").expect("get"), Some(b"a1".to_vec()));
}

#[test]
fn cannot_delete_branch_with_children() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    db.bucket_sync("data").expect("bucket");
    db.create_branch_sync("a", db.latest_read_version())
        .expect("create a");
    db.create_branch_from_sync("b", "a").expect("create b");

    assert!(
        db.delete_branch_sync("a").is_err(),
        "a still has child b, so it cannot be deleted"
    );
    // Delete the child first, then the parent.
    db.delete_branch_sync("b").expect("delete child");
    db.delete_branch_sync("a")
        .expect("delete parent after child gone");
    assert!(db.list_branches_sync().expect("list").is_empty());
}

#[test]
fn recreated_branch_does_not_inherit_deleted_branch_data() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    let bucket = db.bucket_sync("data").expect("bucket");
    bucket
        .put_sync(b"k".to_vec(), b"parent".to_vec())
        .expect("seed");

    db.create_branch_sync("dev", db.latest_read_version())
        .expect("create");
    {
        let mut dev = db.open_branch_sync("dev").expect("open");
        dev.put_sync("data", b"k", b"old".to_vec()).expect("write");
        dev.put_sync("data", b"only-old", b"x".to_vec())
            .expect("write2");
    }
    db.delete_branch_sync("dev")
        .expect("delete (clears the data bucket)");

    // Recreate the same name and write to the same bucket. The branch must
    // start from the parent, not inherit the deleted branch's rows.
    db.create_branch_sync("dev", db.latest_read_version())
        .expect("recreate");
    let mut dev = db.open_branch_sync("dev").expect("reopen");
    dev.put_sync("data", b"k", b"new".to_vec()).expect("write");
    assert_eq!(
        dev.get_sync("data", b"k").expect("get"),
        Some(b"new".to_vec())
    );
    assert_eq!(
        dev.get_sync("data", b"only-old").expect("get"),
        None,
        "the deleted branch's data was cleared, not inherited"
    );
}

#[test]
fn branch_info_exposes_fork_and_parent_without_opening_data() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    db.bucket_sync("data").expect("bucket");
    assert!(
        db.branch_info_sync("missing").expect("info").is_none(),
        "an unknown branch has no lineage"
    );

    let fork = db.latest_read_version();
    db.create_branch_sync("a", fork).expect("create a");
    let a = db.branch_info_sync("a").expect("info").expect("a present");
    assert_eq!(a.fork(), fork, "exposes the fork version for fall-through");
    assert_eq!(a.parent(), None, "a forked the root lineage");

    db.create_branch_from_sync("b", "a").expect("create b");
    let b = db.branch_info_sync("b").expect("info").expect("b present");
    assert_eq!(b.parent(), Some("a"), "exposes the parent for nesting");
}

#[test]
fn orphan_fork_checkpoint_is_reconciled_to_the_registry_intent() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    let old_fork = db.latest_read_version();
    futures::executor::block_on(
        db.create_internal_checkpoint_at(&fork_checkpoint("dev"), old_fork),
    )
    .expect("orphan checkpoint creates");
    db.delete_branch_sync("dev")
        .expect("absent branch delete is idempotent");
    assert_eq!(
        db.internal_checkpoint_read_version_sync(&fork_checkpoint("dev"))
            .expect("delete must not race a concurrent creator's orphan checkpoint"),
        old_fork
    );
    db.bucket_sync("data")
        .expect("bucket")
        .put_sync(b"advance".to_vec(), b"1".to_vec())
        .expect("version advances");
    let new_fork = db.latest_read_version();
    assert_ne!(old_fork, new_fork);

    db.create_branch_sync("dev", new_fork)
        .expect("stale orphan is removed before activating the new intent");
    assert_eq!(
        db.branch_info_sync("dev")
            .expect("registry reads")
            .expect("branch publishes")
            .fork(),
        new_fork
    );
}

#[test]
fn open_branch_rejects_a_corrupt_lineage_cycle() {
    let db = Db::open_sync(DbOptions::memory().with_keep_last_read_versions(64)).expect("open");
    let fork = db.latest_read_version();
    let registry = db
        .internal_bucket_sync(registry_bucket())
        .expect("open registry");
    for (name, parent, generation) in [("a", "b", [1; 16]), ("b", "a", [2; 16])] {
        let entry = RegistryEntry {
            fork,
            parent: Some(parent.to_owned()),
            written_buckets: std::collections::BTreeSet::default(),
            lifecycle: BranchLifecycle::Active,
            generation,
        };
        registry
            .put_sync(
                name.as_bytes(),
                entry.encode().expect("encode corrupt entry"),
            )
            .expect("install corrupt lineage");
    }

    let Err(error) = db.open_branch_sync("a") else {
        panic!("lineage cycle must fail closed");
    };
    assert!(
        matches!(error, Error::Corruption { ref message } if message.contains("cycle")),
        "unexpected cycle error: {error:?}"
    );
}

#[test]
fn drop_bucket_removes_it_in_memory() {
    let db = memory_db();
    let bucket = db.bucket_sync("scratch").expect("bucket");
    bucket.put_sync(b"k".to_vec(), b"v".to_vec()).expect("put");

    db.drop_bucket_sync("scratch").expect("drop");
    assert!(
        db.drop_bucket_sync("scratch").is_err(),
        "dropping a gone bucket errors"
    );
    assert!(
        db.drop_bucket_sync("default").is_err(),
        "the default bucket cannot be dropped"
    );

    // Recreating the name yields a fresh, empty bucket.
    let fresh = db.bucket_sync("scratch").expect("recreate");
    assert_eq!(fresh.get_sync(b"k").expect("get"), None);
}

#[test]
fn drop_bucket_persists_across_reopen() {
    let dir = std::env::temp_dir().join(format!("trine-drop-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    {
        let db = Db::open_sync(&dir).expect("open");
        db.bucket_sync("scratch")
            .expect("scratch")
            .put_sync(b"k".to_vec(), b"v".to_vec())
            .expect("put");
        db.bucket_sync("keep")
            .expect("keep")
            .put_sync(b"k".to_vec(), b"keep".to_vec())
            .expect("put");
        db.drop_bucket_sync("scratch").expect("drop");
    }
    // Reopen: the dropped bucket is gone (recreates empty); the other survives.
    let db = Db::open_sync(&dir).expect("reopen");
    assert_eq!(
        db.bucket_sync("scratch")
            .expect("scratch")
            .get_sync(b"k")
            .expect("get"),
        None,
        "dropped bucket did not come back with its data"
    );
    assert_eq!(
        db.bucket_sync("keep")
            .expect("keep")
            .get_sync(b"k")
            .expect("get"),
        Some(b"keep".to_vec()),
        "an untouched bucket survives the drop"
    );
    let _ = std::fs::remove_dir_all(&dir);
}