pagedb 0.1.0-beta.6

use std::time::Duration;

use pagedb::vfs::memory::MemVfs;
use pagedb::{CommitId, Db, OpenOptions, PagedbError, RealmId, RealmQuotas, RetainPolicy};

const PAGE: usize = 4096;
const KEK: [u8; 32] = [7u8; 32];
const REALM: RealmId = RealmId::new([1u8; 16]);

async fn open_with_policy(policy: RetainPolicy) -> Db<MemVfs> {
    let opts = OpenOptions::default().with_commit_history_retain(policy);
    Db::open(MemVfs::new(), KEK, PAGE, REALM, opts)
        .await
        .unwrap()
}

/// Write `n` transactions, each putting key `b"k"` with value = commit number.
/// Returns the list of `CommitId`s in order.
async fn write_n(db: &Db<MemVfs>, n: usize) -> Vec<CommitId> {
    let mut ids = Vec::with_capacity(n);
    for i in 0..n {
        let mut w = db.begin_write().await.unwrap();
        let val = (i as u64).to_le_bytes().to_vec();
        w.put(b"k", &val).await.unwrap();
        let cid = w.commit().await.unwrap();
        ids.push(cid);
    }
    ids
}

#[tokio::test(flavor = "current_thread")]
async fn begin_read_at_recent_commit() {
    let db = open_with_policy(RetainPolicy::Unbounded).await;
    let ids = write_n(&db, 5).await;

    for (i, &cid) in ids.iter().enumerate() {
        let rtxn = db.begin_read_at(cid).await.unwrap();
        let val = rtxn.get(b"k").await.unwrap().expect("key must exist");
        let stored = u64::from_le_bytes(val.as_ref().try_into().unwrap());
        assert_eq!(stored, i as u64, "commit {cid:?} should see value {i}");
        assert_eq!(rtxn.commit_id(), cid);
    }
}

#[tokio::test(flavor = "current_thread")]
async fn begin_read_at_pruned_returns_commit_gone() {
    let db = open_with_policy(RetainPolicy::Count(3)).await;
    let ids = write_n(&db, 10).await;

    // Commit 1 (ids[0]) should be pruned; Count(3) keeps newest 3.
    let result = db.begin_read_at(ids[0]).await;
    match result {
        Err(PagedbError::CommitGone {
            commit,
            oldest_available,
        }) => {
            assert_eq!(commit, ids[0]);
            // oldest_available must be >= 7th commit (index 6).
            assert!(
                oldest_available.value() >= ids[6].value(),
                "oldest_available={oldest_available:?} expected >= {:?}",
                ids[6]
            );
        }
        Err(other) => panic!("expected CommitGone, got {other:?}"),
        Ok(_) => panic!("expected CommitGone for pruned commit, but got Ok"),
    }
}

#[tokio::test(flavor = "current_thread")]
async fn retain_policy_count_prunes_to_n() {
    let db = open_with_policy(RetainPolicy::Count(3)).await;
    let ids = write_n(&db, 10).await;

    // Exactly 3 entries should be reachable.
    let mut reachable = Vec::new();
    for &cid in &ids {
        if db.begin_read_at(cid).await.is_ok() {
            reachable.push(cid);
        }
    }

    assert_eq!(
        reachable.len(),
        3,
        "expected exactly 3 reachable commits, got {reachable:?}"
    );
    // The reachable ones should be the newest 3.
    assert_eq!(reachable, &ids[7..]);
}

#[tokio::test(flavor = "current_thread")]
async fn retain_policy_unbounded_keeps_all() {
    let db = open_with_policy(RetainPolicy::Unbounded).await;
    let ids = write_n(&db, 20).await;

    for &cid in &ids {
        db.begin_read_at(cid)
            .await
            .unwrap_or_else(|e| panic!("expected commit {cid:?} to be reachable, got {e:?}"));
    }
}

#[tokio::test(flavor = "current_thread")]
async fn reader_pin_blocks_pruning() {
    let db = open_with_policy(RetainPolicy::Count(2)).await;
    let ids = write_n(&db, 1).await;
    let first = ids[0];

    // Pin a reader at commit 1.
    let _pinned = db.begin_read_at(first).await.unwrap();

    // Write 10 more txns — normally this would prune commit 1.
    write_n(&db, 10).await;

    // Pinned commit must still be reachable.
    db.begin_read_at(first)
        .await
        .unwrap_or_else(|e| panic!("pinned commit should still be reachable, got {e:?}"));
}

#[tokio::test(flavor = "current_thread")]
async fn history_persists_across_reopen() {
    let vfs = MemVfs::new();
    let opts = OpenOptions::default().with_commit_history_retain(RetainPolicy::Unbounded);

    let ids = {
        let db = Db::open(vfs.clone(), KEK, PAGE, REALM, opts.clone())
            .await
            .unwrap();
        write_n(&db, 5).await
    };

    // Reopen.
    let db2 = Db::open(vfs, KEK, PAGE, REALM, opts).await.unwrap();

    let cid3 = ids[2]; // commit 3
    db2.begin_read_at(cid3)
        .await
        .unwrap_or_else(|e| panic!("expected commit {cid3:?} to survive reopen, got {e:?}"));
}

#[tokio::test(flavor = "current_thread")]
async fn quota_update_preserves_history_across_reopen() {
    let vfs = MemVfs::new();
    let opts = OpenOptions::default().with_commit_history_retain(RetainPolicy::Unbounded);

    let first_commit = {
        let db = Db::open(vfs.clone(), KEK, PAGE, REALM, opts.clone())
            .await
            .unwrap();
        let ids = write_n(&db, 2).await;
        db.set_realm_quotas(REALM, RealmQuotas::default())
            .await
            .unwrap();
        db.begin_read_at(ids[0])
            .await
            .expect("quota publication must preserve live retained history");
        ids[0]
    };

    let reopened = Db::open(vfs, KEK, PAGE, REALM, opts).await.unwrap();
    let historical = reopened
        .begin_read_at(first_commit)
        .await
        .expect("quota publication must preserve retained history after reopen");
    assert_eq!(
        historical.get(b"k").await.unwrap().as_deref(),
        Some(0_u64.to_le_bytes().as_slice())
    );
}

#[tokio::test(flavor = "current_thread")]
async fn retain_policy_age_prunes_old() {
    // Age(0) means threshold = now - 0 = now, so every entry with
    // unix_seconds < now is eligible. Because commits happen nearly
    // instantaneously (< 1 second apart), the pruner sees all earlier entries
    // as potentially expired. The latest commit is always exempt (we skip
    // deleting the entry we just inserted).
    let db = open_with_policy(RetainPolicy::Age(Duration::from_secs(0))).await;
    let ids = write_n(&db, 3).await;

    // Only the latest commit must be reachable; earlier ones may be pruned.
    db.begin_read_at(ids[2])
        .await
        .unwrap_or_else(|e| panic!("latest commit should be reachable, got {e:?}"));

    // Commits 1 and 2 should be pruned (Age 0).
    for &cid in &ids[..2] {
        let result = db.begin_read_at(cid).await;
        // They may or may not be pruned depending on timing; we accept either
        // NotFound/CommitGone or a valid txn (sub-second window). This test
        // is mainly verifying no panic and that the policy runs at all.
        let _ = result;
    }

    // Verify at least one entry was pruned OR the latest still reads correctly.
    let latest_txn = db.begin_read_at(ids[2]).await.unwrap();
    assert_eq!(latest_txn.commit_id(), ids[2]);
}