//! The ledger's promises (spec §9.5, goal G6): an untouched ledger verifies,
//! with checkpoints signed by the instance key; an edited, removed or
//! re-signed event is caught; projections always equal a rebuild.

use rand::{Rng, SeedableRng};
use spectra_core::event::NewEvent;
use spectra_core::keys::KeyPair;
use spectra_core::ledger::{Ledger, NoKeys, Problem, CHECKPOINT_EVERY};
use std::path::Path;

async fn open(dir: &Path) -> Ledger {
    let instance = KeyPair::instance(dir, "test-instance").unwrap();
    let ledger = Ledger::open(&dir.join("spectra.db"), Some(instance.clone())).await.unwrap();
    if ledger.head().await.0 == 0 {
        ledger
            .append_one(NewEvent::system(
                "identity.instance",
                serde_json::json!({ "key": instance.key.to_hex(), "identity": hex::encode(instance.identity.encode()) }),
            ))
            .await
            .unwrap();
    }
    ledger
}

fn msg(i: usize) -> NewEvent {
    NewEvent::system("test.message", serde_json::json!({ "n": i, "body": format!("message {i}") })).subject(format!("m{i}"))
}

async fn raw(dir: &Path, sql: &str) {
    let pool = sqlx::SqlitePool::connect(&format!("sqlite://{}", dir.join("spectra.db").display())).await.unwrap();
    sqlx::query(sql).execute(&pool).await.unwrap();
    pool.close().await;
}

#[tokio::test]
async fn an_untouched_ledger_verifies_with_signed_checkpoints() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    for i in 0..(CHECKPOINT_EVERY as usize + 10) {
        ledger.append_one(msg(i)).await.unwrap();
    }
    let cp = ledger.checkpoint().await.unwrap().expect("events since the last checkpoint");
    assert_eq!(cp.kind, "ledger.checkpoint");
    let v = ledger.verify(&mut NoKeys).await.unwrap();
    assert!(v.clean(), "{:?}", v.findings);
    assert_eq!(v.checkpoints, 2, "one at 256 events, one on demand");
    assert_eq!(v.uncovered, 0);
    assert_eq!(v.head, ledger.head().await.1);
    assert!(ledger.checkpoint().await.unwrap().is_none(), "nothing new to cover");
}

#[tokio::test]
async fn an_edited_event_is_caught() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    for i in 0..20 {
        ledger.append_one(msg(i)).await.unwrap();
    }
    ledger.checkpoint().await.unwrap();
    drop(ledger);
    raw(dir.path(), r#"UPDATE events SET payload = CAST('{"body":"message 5, edited","n":5}' AS BLOB) WHERE subject = 'm5'"#).await;
    let v = open(dir.path()).await.verify(&mut NoKeys).await.unwrap();
    assert_eq!(v.findings.len(), 1, "{:?}", v.findings);
    assert_eq!(v.findings[0].problem, Problem::HashMismatch);
}

#[tokio::test]
async fn an_edit_with_its_hash_recomputed_breaks_the_chain_or_the_checkpoint() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    for i in 0..20 {
        ledger.append_one(msg(i)).await.unwrap();
    }
    ledger.checkpoint().await.unwrap();
    // A careful forger recomputes the edited event's hash: the next event's
    // prev no longer matches, and the checkpoint names the old head.
    let mut e = ledger.events(1, 100).await.unwrap().into_iter().find(|e| e.subject.as_deref() == Some("m19")).unwrap();
    e.payload = Some(br#"{"body":"forged","n":19}"#.to_vec());
    let forged = hex::encode(e.computed_hash());
    drop(ledger);
    raw(
        dir.path(),
        &format!(r#"UPDATE events SET payload = CAST('{{"body":"forged","n":19}}' AS BLOB), hash = X'{forged}' WHERE seq = {}"#, e.seq),
    )
    .await;
    let v = open(dir.path()).await.verify(&mut NoKeys).await.unwrap();
    assert!(v.findings.iter().any(|f| f.problem == Problem::ChainBreak || f.problem == Problem::BadCheckpoint), "{:?}", v.findings);
}

#[tokio::test]
async fn a_removed_event_is_caught() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    for i in 0..10 {
        ledger.append_one(msg(i)).await.unwrap();
    }
    drop(ledger);
    raw(dir.path(), "DELETE FROM events WHERE subject = 'm4'").await;
    let v = open(dir.path()).await.verify(&mut NoKeys).await.unwrap();
    assert!(v.findings.iter().any(|f| f.problem == Problem::Gap), "{:?}", v.findings);
    assert!(v.findings.iter().any(|f| f.problem == Problem::ChainBreak), "{:?}", v.findings);
}

#[tokio::test]
async fn a_checkpoint_signed_by_another_key_is_caught() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    ledger.append_one(msg(1)).await.unwrap();
    let cp = ledger.checkpoint().await.unwrap().unwrap();
    drop(ledger);
    let mallory = KeyPair::generate("mallory", "");
    let sig = mallory.sign(&spectra_core::ledger::checkpoint_message(cp.seq - 1, &cp.prev));
    let proof = hex::encode(spectra_core::cbor::encode(&sig.to_value()));
    raw(dir.path(), &format!("UPDATE events SET auth_proof = X'{proof}' WHERE seq = {}", cp.seq)).await;
    let v = open(dir.path()).await.verify(&mut NoKeys).await.unwrap();
    assert_eq!(v.findings.len(), 1, "{:?}", v.findings);
    assert_eq!(v.findings[0].problem, Problem::BadCheckpoint);
}

#[tokio::test]
async fn projections_equal_a_rebuild_over_ten_thousand_random_events() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    let mut rng = rand::rngs::StdRng::seed_from_u64(7);
    // Kinds that projections read, with payloads that are often malformed:
    // a projection must skip what it cannot use, never refuse the append.
    let kinds = [
        "message.posted",
        "agent.presence",
        "agent.status",
        "agent.session_started",
        "agent.session_ended",
        "agent.added",
        "agent.disabled",
        "identity.owner",
        "identity.rotated",
        "project.added",
        "task.created",
        "tool.called",
    ];
    let started = std::time::Instant::now();
    let mut appended = 0;
    while appended < 10_000 {
        let n = rng.gen_range(1..50);
        let batch = (0..n)
            .map(|i| {
                NewEvent::system(kinds[rng.gen_range(0..kinds.len())], serde_json::json!({ "i": appended + i, "r": rng.gen::<u32>() }))
            })
            .collect();
        ledger.append(batch).await.unwrap();
        appended += n;
    }
    let per_batch_ms = started.elapsed().as_millis() as f64 / (10_000.0 / 25.0);
    assert!(ledger.check().await.unwrap().is_empty(), "projections drifted from the ledger");
    ledger.rebuild().await.unwrap();
    assert!(ledger.check().await.unwrap().is_empty());
    let v = ledger.verify(&mut NoKeys).await.unwrap();
    assert!(v.clean(), "{:?}", &v.findings[..v.findings.len().min(3)]);
    assert!(v.checkpoints >= 10_000 / CHECKPOINT_EVERY);
    // Generous in a debug build; the release number is in docs/plan.md.
    assert!(per_batch_ms < 200.0, "{per_batch_ms} ms a batch");
}

#[tokio::test]
async fn a_ledger_from_a_newer_spectra_is_refused() {
    let dir = tempfile::tempdir().unwrap();
    drop(open(dir.path()).await);
    raw(dir.path(), "UPDATE meta SET value = '99' WHERE key = 'ledger_format'").await;
    let err = Ledger::open(&dir.path().join("spectra.db"), None).await.err().expect("refused");
    assert!(err.to_string().contains("newer Spectra"), "{err}");
}

#[tokio::test]
async fn the_bus_delivers_what_was_appended_in_order() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    let mut rx = ledger.subscribe();
    ledger.append(vec![msg(1), msg(2)]).await.unwrap();
    assert_eq!(rx.recv().await.unwrap().subject.as_deref(), Some("m1"));
    assert_eq!(rx.recv().await.unwrap().subject.as_deref(), Some("m2"));
}

#[tokio::test]
async fn reopening_continues_the_chain() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    ledger.append_one(msg(1)).await.unwrap();
    let (seq, hash) = ledger.head().await;
    drop(ledger);
    let ledger = open(dir.path()).await;
    let e = ledger.append_one(msg(2)).await.unwrap();
    assert_eq!((e.seq, e.prev), (seq + 1, hash));
    assert!(ledger.verify(&mut NoKeys).await.unwrap().clean());
}

#[tokio::test]
async fn a_single_append_is_fast() {
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    let started = std::time::Instant::now();
    for i in 0..1_000 {
        ledger.append_one(msg(i)).await.unwrap();
    }
    let each_us = started.elapsed().as_micros() / 1_000;
    eprintln!("append: {each_us} µs an event over 1,000 single appends");
    // The target is 10 ms with fsync (spec §24); debug builds get headroom.
    assert!(each_us < 50_000, "{each_us} µs");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_append_abandoned_mid_flight_still_commits_and_never_blocks_later_ones() {
    // A client that hangs up mid-request drops the future doing its append.
    // Before the writer ran in a task of its own, that could leave a pooled
    // connection inside an open transaction, and every later "commit" on it
    // went nowhere (found by hand in the Radar demo, 2026-11-02).
    let dir = tempfile::tempdir().unwrap();
    let ledger = open(dir.path()).await;
    for i in 0..300 {
        let f = ledger.append_one(msg(i));
        tokio::select! {
            biased;
            _ = f => {}
            _ = tokio::task::yield_now() => {}
        }
    }
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    let last = ledger.append_one(msg(10_000)).await.unwrap();
    let (head, _) = ledger.head().await;
    assert_eq!(last.seq, head);
    let other = sqlx::SqlitePool::connect(&format!("sqlite://{}", dir.path().join("spectra.db").display())).await.unwrap();
    let seen: i64 = sqlx::query_scalar("SELECT MAX(seq) FROM events").fetch_one(&other).await.unwrap();
    assert_eq!(seen as u64, head, "another connection sees every committed event");
    let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE kind = 'test.message'").fetch_one(&other).await.unwrap();
    assert_eq!(n, 301, "every abandoned append still committed");
    assert!(ledger.verify(&mut NoKeys).await.unwrap().clean());
}