nodedb-lite 0.1.0

//! End-to-end integration tests for NodeDB-Lite.
//!
//! CRDT convergence works across instances, and the compensation flow is correct.

use nodedb_client::NodeDb;
use nodedb_lite::{NodeDbLite, PagedbStorageMem};
use nodedb_types::document::Document;
use nodedb_types::id::NodeId;
use nodedb_types::value::Value;

async fn open_db() -> std::sync::Arc<NodeDbLite<PagedbStorageMem>> {
    let storage = PagedbStorageMem::open_in_memory().await.unwrap();
    NodeDbLite::open(storage).await.unwrap()
}

// ═══════════════════════════════════════════════════════════════════════
// 6.1 Standalone Lite (No Origin)
// ═══════════════════════════════════════════════════════════════════════
//
// Vector search, graph BFS, document CRUD, and flush/reopen correctness
// are covered by the equivalent tests in `integration.rs`
// (`vector_batch_insert_and_search_correctness`,
// `graph_batch_and_traverse_correctness`, `document_crud_100`,
// `flush_and_reopen_persists_all`). Only Lite-specific concerns
// (memory budget, CRDT convergence, compensation, delta ack, native
// smoke) live in this file.

#[tokio::test]
async fn e2e_memory_stays_within_budget() {
    let db = open_db().await;

    // Small workload — 50 × 32d. Enough to exercise the budget tracker
    // without making the test a benchmark.
    let vectors: Vec<(String, Vec<f32>)> = (0..50)
        .map(|i| {
            let emb: Vec<f32> = (0..32).map(|d| ((i * 32 + d) as f32) * 0.001).collect();
            (format!("v{i}"), emb)
        })
        .collect();
    let refs: Vec<(&str, &[f32])> = vectors
        .iter()
        .map(|(id, e)| (id.as_str(), e.as_slice()))
        .collect();
    db.batch_vector_insert("vecs", &refs).await.unwrap();

    let used = db.governor().total_allocated();
    let budget = db.governor().global_ceiling();

    assert!(used <= budget, "memory {used} exceeds budget {budget}");
}

// ═══════════════════════════════════════════════════════════════════════
// 6.2 / 6.3 CRDT Convergence & Compensation (simulated, no Origin)
// ═══════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn e2e_two_lite_instances_crdt_convergence() {
    // Simulate two Lite devices making independent edits, then merging.
    let db1 = {
        let s = PagedbStorageMem::open_in_memory().await.unwrap();
        NodeDbLite::open(s).await.unwrap()
    };
    let db2 = {
        let s = PagedbStorageMem::open_in_memory().await.unwrap();
        NodeDbLite::open(s).await.unwrap()
    };

    // Device 1 writes.
    let mut doc1 = Document::new("shared-doc");
    doc1.set("author", Value::String("alice".into()));
    db1.document_put("notes", doc1).await.unwrap();

    // Device 2 writes different field.
    let mut doc2 = Document::new("shared-doc");
    doc2.set("reviewer", Value::String("bob".into()));
    db2.document_put("notes", doc2).await.unwrap();

    // Export deltas from both.
    let deltas1 = db1.pending_crdt_deltas().unwrap();
    let deltas2 = db2.pending_crdt_deltas().unwrap();

    assert!(!deltas1.is_empty());
    assert!(!deltas2.is_empty());

    // Cross-import: db1 gets db2's deltas and vice versa.
    for d in &deltas2 {
        db1.import_remote_deltas(&d.collection, &d.delta_bytes)
            .unwrap();
    }
    for d in &deltas1 {
        db2.import_remote_deltas(&d.collection, &d.delta_bytes)
            .unwrap();
    }

    // Both should now see both fields.
    let doc_from_1 = db1
        .document_get("notes", "shared-doc")
        .await
        .unwrap()
        .unwrap();
    let doc_from_2 = db2
        .document_get("notes", "shared-doc")
        .await
        .unwrap()
        .unwrap();

    // CRDT merge: both fields should be present on both devices.
    assert!(doc_from_1.get_str("author").is_some() || doc_from_1.get_str("reviewer").is_some());
    assert!(doc_from_2.get_str("author").is_some() || doc_from_2.get_str("reviewer").is_some());
}

#[tokio::test]
async fn e2e_compensation_reject_rollback() {
    let db = open_db().await;

    // Write a document.
    let mut doc = Document::new("user-1");
    doc.set("username", Value::String("alice".into()));
    db.document_put("users", doc).await.unwrap();

    // Verify it exists.
    assert!(db.document_get("users", "user-1").await.unwrap().is_some());

    // Get the pending delta.
    let deltas = db.pending_crdt_deltas().unwrap();
    assert!(!deltas.is_empty());
    let mutation_id = deltas[0].mutation_id;

    // Simulate Origin rejection (UNIQUE violation).
    db.reject_delta(mutation_id).unwrap();

    // After rejection, the document should be rolled back.
    let doc = db.document_get("users", "user-1").await.unwrap();
    assert!(doc.is_none(), "rejected document should be rolled back");
}

#[tokio::test]
async fn e2e_delta_acknowledge_clears_pending() {
    let db = open_db().await;

    db.document_put("a", Document::new("d1")).await.unwrap();
    db.document_put("a", Document::new("d2")).await.unwrap();
    db.document_put("a", Document::new("d3")).await.unwrap();

    let deltas = db.pending_crdt_deltas().unwrap();
    assert_eq!(deltas.len(), 3);

    // ACK only the second delta. Acks are per-mutation, so the first stays
    // queued — retiring it here would drop a write Origin never acknowledged.
    db.acknowledge_deltas(deltas[1].mutation_id).unwrap();

    let remaining: Vec<u64> = db
        .pending_crdt_deltas()
        .unwrap()
        .iter()
        .map(|d| d.mutation_id)
        .collect();
    assert_eq!(
        remaining,
        vec![deltas[0].mutation_id, deltas[2].mutation_id]
    );
}

// ═══════════════════════════════════════════════════════════════════════
// 6.5 Platform-Specific
// ═══════════════════════════════════════════════════════════════════════

#[tokio::test]
async fn e2e_native_full_suite_passes() {
    // This test IS the proof that native works — it runs alongside all others.
    let db = open_db().await;
    db.vector_insert("test", "v1", &[1.0, 0.0], None)
        .await
        .unwrap();
    let r = db
        .vector_search("test", &[1.0, 0.0], 1, None, None)
        .await
        .unwrap();
    assert_eq!(r.len(), 1);

    db.graph_insert_edge(
        "test",
        &NodeId::from_validated("a".to_string()),
        &NodeId::from_validated("b".to_string()),
        "L",
        None,
    )
    .await
    .unwrap();
    let sg = db
        .graph_traverse(
            "test",
            &NodeId::from_validated("a".to_string()),
            1,
            nodedb_types::graph::Direction::Out,
            None,
        )
        .await
        .unwrap();
    assert!(sg.node_count() >= 2);

    db.document_put("d", Document::new("d1")).await.unwrap();
    assert!(db.document_get("d", "d1").await.unwrap().is_some());
}