//! 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());

}