mire 0.2.4

//! Integration tests against a real Postgres.
//!
//! These need a database; set `DATABASE_URL` to run them. When it is absent
//! (e.g. a plain `cargo test` with no database) each test prints a skip notice
//! and returns, so the suite stays green without external services.

use std::time::Duration;

use mire::{
    Aggregate, EventData, EventHandler, EventStore, ExpectedVersion, HandledEvent,
    ProjectionRunner, Snapshot, Subscription,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

mod common;

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
enum CounterEvent {
    Incremented { by: i64 },
}

impl EventData for CounterEvent {
    fn event_type(&self) -> &'static str {
        match self {
            CounterEvent::Incremented { .. } => "counter.incremented",
        }
    }
}

#[derive(Debug, Default, Serialize, Deserialize)]
struct Counter {
    total: i64,
}

impl Aggregate for Counter {
    type Event = CounterEvent;

    fn stream_category() -> &'static str {
        "counter"
    }

    fn apply(&mut self, event: &CounterEvent) {
        match event {
            CounterEvent::Incremented { by } => self.total += by,
        }
    }
}

impl Snapshot for Counter {
    const SNAPSHOT_VERSION: i32 = 1;
    // Snapshot on every other event so the test exercises it quickly.
    const SNAPSHOT_FREQUENCY: i64 = 2;
}

async fn store() -> Option<EventStore> {
    common::store().await
}

#[tokio::test]
async fn records_and_rebuilds_from_the_log() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();

    let mut counter = store.load_or_default::<Counter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 3 });
    counter.record(CounterEvent::Incremented { by: 7 });
    store.save(&mut counter).await.unwrap();
    assert_eq!(counter.version, 2);

    let reloaded = store.load::<Counter>(&id).await.unwrap().unwrap();
    assert_eq!(reloaded.state.total, 10);
    assert_eq!(reloaded.version, 2);
}

#[tokio::test]
async fn rejects_stale_writes() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();

    let mut counter = store.load_or_default::<Counter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 1 });
    store.save(&mut counter).await.unwrap();

    let result = store
        .append::<CounterEvent>(
            &counter.stream_id,
            Counter::stream_category(),
            ExpectedVersion::Exact(0), // real version is 1
            &[mire::Event::new(CounterEvent::Incremented { by: 1 })],
        )
        .await;

    assert!(matches!(
        result,
        Err(mire::EventStoreError::ConcurrencyConflict { .. })
    ));
}

// ============================================================
// Lock-free append safety properties (tasks/lock-free-append.md §6)
// ============================================================

/// P2 — `NoStream` on an existing stream returns a `ConcurrencyConflict`
/// with `actual = current_version`.
#[tokio::test]
async fn no_stream_on_existing_returns_actual_version() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();

    let mut counter = store.load_or_default::<Counter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 1 });
    counter.record(CounterEvent::Incremented { by: 1 });
    counter.record(CounterEvent::Incremented { by: 1 });
    counter.record(CounterEvent::Incremented { by: 1 });
    counter.record(CounterEvent::Incremented { by: 1 });
    store.save(&mut counter).await.unwrap();
    assert_eq!(counter.version, 5);

    let result = store
        .append::<CounterEvent>(
            &counter.stream_id,
            Counter::stream_category(),
            ExpectedVersion::NoStream,
            &[mire::Event::new(CounterEvent::Incremented { by: 1 })],
        )
        .await;

    match result {
        Err(mire::EventStoreError::ConcurrencyConflict {
            expected, actual, ..
        }) => {
            assert_eq!(expected, 0);
            assert_eq!(actual, 5, "NoStream on existing must surface actual=5");
        }
        other => panic!("expected ConcurrencyConflict, got: {other:?}"),
    }
}

/// P3 — N parallel `Any` writers against the same stream all succeed
/// and produce a contiguous version sequence with no gaps and no
/// duplicates.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn parallel_any_appends_produce_contiguous_versions() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();
    let stream_id = format!("counter-{id}");
    let n_writers: i64 = 100;

    let mut handles = Vec::new();
    for _ in 0..n_writers {
        let store = store.clone();
        let stream_id = stream_id.clone();
        handles.push(tokio::spawn(async move {
            store
                .append::<CounterEvent>(
                    &stream_id,
                    Counter::stream_category(),
                    ExpectedVersion::Any,
                    &[mire::Event::new(CounterEvent::Incremented { by: 1 })],
                )
                .await
                .expect("Any must never produce ConcurrencyConflict")
        }));
    }
    for h in handles {
        h.await.unwrap();
    }

    // The N writers should have allocated exactly versions 1..=N with no
    // gaps and no duplicates. Read the stream back and check.
    let events = store
        .read_stream(&stream_id, mire::StreamQuery::default())
        .await
        .unwrap();
    let mut versions: Vec<i64> = events.iter().map(|e| e.stream_version).collect();
    versions.sort();
    let expected: Vec<i64> = (1..=n_writers).collect();
    assert_eq!(versions, expected, "version range must be contiguous 1..=N");
}

/// P4 — Backup invariant: heavy `Exact(v)` contention must never trigger
/// the `UNIQUE (stream_id, stream_version)` constraint. If it does, the
/// CAS is broken and the constraint had to catch the corruption.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn exact_contention_never_violates_unique_constraint() {
    // Dedicated, large pool so this test doesn't starve peer tests that
    // share the binary's default pool (sqlx's default max_connections=10
    // would be saturated by the writer fleet below).
    let Some(url) = std::env::var("DATABASE_URL").ok() else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = sqlx::postgres::PgPoolOptions::new()
        .max_connections(40)
        .connect(&url)
        .await
        .expect("connect");
    // Schema is migrated once per binary behind the shared guard, on its own
    // connection — never concurrently with the writer fleet below.
    let store = common::store_with_pool(pool).await;
    let id = Uuid::new_v4().to_string();
    let stream_id = format!("counter-{id}");
    let writers: i64 = 16;
    let events_per_writer: i64 = 4;

    let mut handles = Vec::new();
    for _ in 0..writers {
        let store = store.clone();
        let stream_id = stream_id.clone();
        handles.push(tokio::spawn(async move {
            let mut wins: i64 = 0;
            while wins < events_per_writer {
                // Read current version, then attempt to append at exactly
                // that version. Most attempts conflict; we retry until
                // we win `events_per_writer` times.
                let row: Option<i64> = sqlx::query_scalar(
                    "SELECT stream_version FROM es_streams WHERE stream_id = $1",
                )
                .bind(&stream_id)
                .fetch_optional(store.pool())
                .await
                .unwrap();
                let expected_version = row.unwrap_or(0);
                let expected = if expected_version == 0 {
                    ExpectedVersion::NoStream
                } else {
                    ExpectedVersion::Exact(expected_version)
                };
                match store
                    .append::<CounterEvent>(
                        &stream_id,
                        Counter::stream_category(),
                        expected,
                        &[mire::Event::new(CounterEvent::Incremented { by: 1 })],
                    )
                    .await
                {
                    Ok(_) => wins += 1,
                    Err(mire::EventStoreError::ConcurrencyConflict { .. }) => {}
                    Err(mire::EventStoreError::Database(e)) => {
                        // The UNIQUE constraint would fire as SQLSTATE
                        // 23505. If we see this, the CAS leaked a
                        // duplicate version slot — a real bug.
                        if e.as_database_error().and_then(|d| d.code()).as_deref() == Some("23505")
                        {
                            panic!(
                                "UNIQUE(stream_id, stream_version) violation under \
                                 contended Exact(v) — CAS allowed a duplicate slot: {e}"
                            );
                        }
                        panic!("unexpected DB error: {e}");
                    }
                    Err(e) => panic!("unexpected error: {e}"),
                }
            }
        }));
    }
    for h in handles {
        h.await.unwrap();
    }

    let row: i64 = sqlx::query_scalar("SELECT stream_version FROM es_streams WHERE stream_id = $1")
        .bind(&stream_id)
        .fetch_one(store.pool())
        .await
        .unwrap();
    assert_eq!(row, writers * events_per_writer);
}

/// P6 — A long-running concurrent reader is never blocked by an append
/// in progress. Under FOR UPDATE the reader would have to wait for the
/// writer's commit; under CAS the reader proceeds immediately.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn long_append_does_not_block_concurrent_reader() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();
    let stream_id = format!("counter-{id}");

    // Seed the stream so the row exists.
    let mut counter = store.load_or_default::<Counter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 1 });
    store.save(&mut counter).await.unwrap();

    // Writer task: open a tx, do an append, then sleep for 1s before
    // commit. The append should not hold a row lock past its UPDATE.
    let writer_store = store.clone();
    let writer_stream_id = stream_id.clone();
    let writer = tokio::spawn(async move {
        let mut tx = writer_store.pool().begin().await.unwrap();
        mire::EventStore::append_in_tx::<CounterEvent>(
            &mut tx,
            &writer_stream_id,
            Counter::stream_category(),
            ExpectedVersion::Exact(1),
            &[mire::Event::new(CounterEvent::Incremented { by: 1 })],
        )
        .await
        .unwrap();
        // Hold the tx open for a second to simulate a slow commit.
        tokio::time::sleep(Duration::from_secs(1)).await;
        tx.commit().await.unwrap();
    });

    // Give the writer a head start to enter its sleep.
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Reader task: should complete quickly, NOT wait for the writer.
    let reader_start = tokio::time::Instant::now();
    let _row: i64 =
        sqlx::query_scalar("SELECT stream_version FROM es_streams WHERE stream_id = $1")
            .bind(&stream_id)
            .fetch_one(store.pool())
            .await
            .unwrap();
    let reader_elapsed = reader_start.elapsed();

    assert!(
        reader_elapsed < Duration::from_millis(500),
        "reader was blocked by an in-flight append for {reader_elapsed:?} — \
         lock contract is broken",
    );

    writer.await.unwrap();
}

#[tokio::test]
async fn snapshot_seeds_then_replays_tail() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();

    let mut counter = store.load_or_default::<Counter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 1 });
    counter.record(CounterEvent::Incremented { by: 2 });
    // version 2 crosses the freq-2 boundary -> a snapshot is written.
    store.save_snapshotting(&mut counter).await.unwrap();

    // More events land after the snapshot; these must be replayed on top of it.
    counter.record(CounterEvent::Incremented { by: 3 });
    store.save_snapshotting(&mut counter).await.unwrap();

    // The snapshot stream holds a snapshot reflecting version 2.
    let snapshot_events = store
        .read_stream(
            &format!("{}-snapshot", counter.stream_id),
            mire::StreamQuery::default(),
        )
        .await
        .unwrap();
    assert!(!snapshot_events.is_empty(), "a snapshot should be written");
    let envelope = &snapshot_events.last().unwrap().data;
    assert_eq!(envelope["version"], 2);

    // Loading via the snapshot path yields the full, current state.
    let loaded = store
        .load_snapshotted::<Counter>(&id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(loaded.state.total, 6);
    assert_eq!(loaded.version, 3);
}

// Two aggregates over the SAME stream category + event type, differing only in
// SNAPSHOT_VERSION and `apply`. Used to prove the snapshot-version guard:
// loading a v1-snapshotted stream as v2 must IGNORE the stale snapshot and
// replay (so the ×10 apply is what produces the state), never deserialize the
// old snapshot bytes into the new shape — that would be silent corruption on a
// schema/version bump.
#[derive(Debug, Default, Serialize, Deserialize)]
struct SnapV1 {
    total: i64,
}
impl Aggregate for SnapV1 {
    type Event = CounterEvent;
    fn stream_category() -> &'static str {
        "snapver-test"
    }
    fn apply(&mut self, e: &CounterEvent) {
        let CounterEvent::Incremented { by } = e;
        self.total += by;
    }
}
impl Snapshot for SnapV1 {
    const SNAPSHOT_VERSION: i32 = 1;
    const SNAPSHOT_FREQUENCY: i64 = 2;
}

#[derive(Debug, Default, Serialize, Deserialize)]
struct SnapV2 {
    total: i64,
}
impl Aggregate for SnapV2 {
    type Event = CounterEvent;
    fn stream_category() -> &'static str {
        "snapver-test"
    }
    fn apply(&mut self, e: &CounterEvent) {
        let CounterEvent::Incremented { by } = e;
        self.total += by * 10; // distinguishable from V1's apply
    }
}
impl Snapshot for SnapV2 {
    const SNAPSHOT_VERSION: i32 = 2;
    const SNAPSHOT_FREQUENCY: i64 = 2;
}

#[tokio::test]
async fn snapshot_from_an_older_version_is_ignored_not_deserialized() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();

    // Write the stream + a v1 snapshot (state {total: 3} via V1's ×1 apply).
    let mut v1 = store.load_or_default::<SnapV1>(&id).await.unwrap();
    v1.record(CounterEvent::Incremented { by: 1 });
    v1.record(CounterEvent::Incremented { by: 2 });
    store.save_snapshotting(&mut v1).await.unwrap();
    assert_eq!(v1.state.total, 3);

    // Load as V2 (a bumped SNAPSHOT_VERSION). The stored snapshot is v1, so the
    // version guard must reject it and replay the log with V2's ×10 apply:
    //   ignored + replay → 1*10 + 2*10 = 30  (correct)
    //   stale snapshot used → 3              (the corruption we must prevent)
    let loaded = store
        .load_snapshotted::<SnapV2>(&id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        loaded.state.total, 30,
        "a snapshot from an older SNAPSHOT_VERSION must be ignored and the log replayed"
    );
    assert_eq!(loaded.version, 2);
}

// Companion to the version-guard test: same SNAPSHOT_VERSION, but the loader's
// state shape can't deserialize the stored snapshot (a required new field). The
// load path's `if let Ok(state)` guard must then fall back to a full replay —
// gracefully, never an error and never a corrupt half-deserialized state.
#[derive(Debug, Default, Serialize, Deserialize)]
struct SnapWriter {
    total: i64,
}
impl Aggregate for SnapWriter {
    type Event = CounterEvent;
    fn stream_category() -> &'static str {
        "snapcorrupt-test"
    }
    fn apply(&mut self, e: &CounterEvent) {
        let CounterEvent::Incremented { by } = e;
        self.total += by;
    }
}
impl Snapshot for SnapWriter {
    const SNAPSHOT_VERSION: i32 = 1;
    const SNAPSHOT_FREQUENCY: i64 = 2;
}

// Same category + version, but a required field absent from SnapWriter's
// snapshot → `serde_json::from_value::<SnapReader>` fails on it. Distinct
// `apply` (×7) so a replay is detectable.
#[derive(Debug, Default, Serialize, Deserialize)]
struct SnapReader {
    total: i64,
    required_new_field: String,
}
impl Aggregate for SnapReader {
    type Event = CounterEvent;
    fn stream_category() -> &'static str {
        "snapcorrupt-test"
    }
    fn apply(&mut self, e: &CounterEvent) {
        let CounterEvent::Incremented { by } = e;
        self.total += by * 7;
    }
}
impl Snapshot for SnapReader {
    const SNAPSHOT_VERSION: i32 = 1;
    const SNAPSHOT_FREQUENCY: i64 = 2;
}

#[tokio::test]
async fn snapshot_that_fails_to_deserialize_falls_back_to_replay() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();

    // Write the stream + a snapshot whose state is {total: 3} (no required_new_field).
    let mut w = store.load_or_default::<SnapWriter>(&id).await.unwrap();
    w.record(CounterEvent::Incremented { by: 1 });
    w.record(CounterEvent::Incremented { by: 2 });
    store.save_snapshotting(&mut w).await.unwrap();

    // Load as SnapReader: the version matches (1), so the version guard passes,
    // but the snapshot state can't deserialize into SnapReader (missing field).
    // The loader must fall back to a full replay (×7), not error or corrupt:
    //   fallback replay → 1*7 + 2*7 = 21   (correct)
    let loaded = store
        .load_snapshotted::<SnapReader>(&id)
        .await
        .expect("an undeserializable snapshot must not error the load")
        .unwrap();
    assert_eq!(
        loaded.state.total, 21,
        "an incompatible snapshot must fall back to a full replay"
    );
    assert_eq!(loaded.version, 2);
}

#[tokio::test]
async fn watermark_does_not_skip_in_flight_lower_transactions() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = store.pool().clone();

    // Open a transaction that appends an event but does NOT commit yet. It holds
    // a low transaction id, pinning the snapshot xmin horizon.
    let mut tx = store.begin_transaction().await.unwrap();
    let stream_a = format!("counter-{}", Uuid::new_v4());
    tx.append::<CounterEvent>(
        &stream_a,
        "counter",
        ExpectedVersion::Any,
        &[mire::Event::new(CounterEvent::Incremented { by: 1 })],
    )
    .await
    .unwrap();

    // Meanwhile a *later* transaction commits a higher-position event.
    let id_b = Uuid::new_v4().to_string();
    let mut b = store.load_or_default::<Counter>(&id_b).await.unwrap();
    b.record(CounterEvent::Incremented { by: 2 });
    store.save(&mut b).await.unwrap();

    // A subscriber must not consume the committed higher event while a lower
    // transaction is still open — otherwise it would skip the lower event once
    // it commits. (A bare global_position cursor would fail here.)
    let mut sub = Subscription::create(
        store.clone(),
        pool.clone(),
        format!("watermark-{}", Uuid::new_v4()),
        500,
    )
    .await
    .unwrap();
    let before = sub.poll_category("counter").await.unwrap();
    assert!(
        before.iter().all(|e| e.stream_id != b.stream_id),
        "must not consume past an in-flight lower transaction"
    );

    // After the open transaction commits, both events become consumable in
    // order. Poll until the xmin horizon advances past every concurrent test.
    tx.commit().await.unwrap();
    let mut sub2 = Subscription::create(
        store.clone(),
        pool.clone(),
        format!("watermark2-{}", Uuid::new_v4()),
        500,
    )
    .await
    .unwrap();
    let (mut seen_a, mut seen_b) = (false, false);
    for _ in 0..400 {
        for e in sub2.poll_category("counter").await.unwrap() {
            seen_a |= e.stream_id == stream_a;
            seen_b |= e.stream_id == b.stream_id;
        }
        if seen_a && seen_b {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(seen_a && seen_b, "both events must eventually be delivered");
}

/// A counter living in its **own stream category**, used only by the
/// projection test. A `ProjectionRunner` subscription replays its whole
/// category from position 0 by design; if this test reused the shared
/// `counter` category, its fresh subscription would have to catch up
/// through every counter event every other test ever wrote (tens of
/// thousands on a long-lived dev database) before reaching its own two —
/// blowing the poll budget. An isolated category keeps catch-up O(this
/// test's own events).
#[derive(Debug, Default, Serialize, Deserialize)]
struct ProjCounter {
    total: i64,
}

impl Aggregate for ProjCounter {
    type Event = CounterEvent;
    fn stream_category() -> &'static str {
        "counter-proj"
    }
    fn apply(&mut self, event: &CounterEvent) {
        match event {
            CounterEvent::Incremented { by } => self.total += by,
        }
    }
}

struct CounterTotals {
    db: PgPool,
    table: String,
}

impl EventHandler for CounterTotals {
    type Aggregate = ProjCounter;

    async fn handle(&self, event: HandledEvent<CounterEvent>) -> anyhow::Result<()> {
        let CounterEvent::Incremented { by } = event.event;
        sqlx::query(sqlx::AssertSqlSafe(format!(
            "INSERT INTO {} (stream_id, total) VALUES ($1, $2)
             ON CONFLICT (stream_id) DO UPDATE SET total = {}.total + EXCLUDED.total",
            self.table, self.table
        )))
        .bind(event.stream_id())
        .bind(by)
        .execute(&self.db)
        .await?;
        Ok(())
    }
}

#[tokio::test]
async fn projection_runner_builds_read_model() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = store.pool().clone();
    // Unique table + subscription per run so concurrent tests don't collide.
    let suffix = Uuid::new_v4().simple().to_string();
    let table = format!("counter_totals_{suffix}");
    sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
        "CREATE TABLE IF NOT EXISTS {table} (stream_id TEXT PRIMARY KEY, total BIGINT NOT NULL)"
    )))
    .execute(&pool)
    .await
    .unwrap();

    let id = Uuid::new_v4().to_string();
    let mut counter = store.load_or_default::<ProjCounter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 4 });
    counter.record(CounterEvent::Incremented { by: 6 });
    store.save(&mut counter).await.unwrap();

    let runner = ProjectionRunner::builder(store.clone())
        .poll_interval(Duration::from_millis(20))
        .subscribe(
            format!("counter-totals-{suffix}"),
            CounterTotals {
                db: pool.clone(),
                table: table.clone(),
            },
        )
        .build();

    let token = CancellationToken::new();
    let handle = {
        let token = token.clone();
        tokio::spawn(async move { runner.run(token).await })
    };

    // Wait for our stream to land in the read model, then shut down.
    let mut total = None;
    // Generous budget: a concurrent test may hold a transaction open, pinning
    // the xmin horizon and delaying delivery (by design).
    for _ in 0..400 {
        total = sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(format!(
            "SELECT total FROM {table} WHERE stream_id = $1"
        )))
        .bind(&counter.stream_id)
        .fetch_optional(&pool)
        .await
        .unwrap();
        if total.is_some() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    token.cancel();
    handle.await.unwrap().unwrap();

    assert_eq!(total, Some(10));
}

/// A handler that always fails — a "poison" event. The subscription must NOT
/// silently skip it: it retries up to `max_attempts`, then stops the runner
/// loudly (`run()` returns `Err`) WITHOUT advancing the checkpoint past the
/// failed event. That preserves at-least-once — a restart re-attempts the
/// event rather than losing it — and a poison can never corrupt the read model
/// by being skipped. (The alternative — advancing past a failed event — would
/// be a silent data-loss bug.)
struct AlwaysFails;
impl EventHandler for AlwaysFails {
    type Aggregate = ProjCounter;
    async fn handle(&self, _event: HandledEvent<CounterEvent>) -> anyhow::Result<()> {
        anyhow::bail!("poison: this handler always fails")
    }
}

#[tokio::test]
async fn poison_handler_stops_loudly_without_advancing_checkpoint() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = store.pool().clone();

    // Ensure there is at least one event in the category to poison on.
    let id = Uuid::new_v4().to_string();
    let mut counter = store.load_or_default::<ProjCounter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 1 });
    store.save(&mut counter).await.unwrap();

    let sub_id = format!("poison-sub-{}", Uuid::new_v4().simple());
    let runner = ProjectionRunner::builder(store.clone())
        .poll_interval(Duration::from_millis(20))
        .max_attempts(2)
        .subscribe(sub_id.clone(), AlwaysFails)
        .build();

    // run() must RETURN (with an error) on poison — never hang, never skip.
    let token = CancellationToken::new();
    let outcome = tokio::time::timeout(Duration::from_secs(30), runner.run(token))
        .await
        .expect("a poison handler must stop the runner, not hang forever");
    assert!(
        outcome.is_err(),
        "a handler failing past max_attempts must stop the subscription loudly, got: {outcome:?}"
    );

    // The checkpoint must NOT have advanced past the poison event: a fresh
    // subscription that never succeeded stays at its initial cursor (0), so a
    // restart re-attempts the event — it is never silently skipped.
    let last_position: Option<i64> =
        sqlx::query_scalar("SELECT last_position FROM es_subscriptions WHERE subscription_id = $1")
            .bind(&sub_id)
            .fetch_optional(&pool)
            .await
            .unwrap();
    assert_eq!(
        last_position.unwrap_or(0),
        0,
        "a poison event must not advance the checkpoint (no silent skip)"
    );
}

/// A handler that PANICS (not a graceful `Err`) — a buggy handler must not be
/// able to kill the host. The runner runs subscription loops as joined tasks,
/// so a panic is contained at the `JoinSet` boundary and surfaced as a `run()`
/// error; the host process survives. (If the panic escaped, this test process
/// would abort and the run would fail.) Like a poison error, it must not
/// advance the checkpoint.
struct PanickingHandler;
impl EventHandler for PanickingHandler {
    type Aggregate = ProjCounter;
    async fn handle(&self, _event: HandledEvent<CounterEvent>) -> anyhow::Result<()> {
        panic!("boom: a handler panic, not a graceful Err");
    }
}

#[tokio::test]
async fn panicking_handler_is_contained_and_stops_the_runner() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = store.pool().clone();

    let id = Uuid::new_v4().to_string();
    let mut counter = store.load_or_default::<ProjCounter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 1 });
    store.save(&mut counter).await.unwrap();

    let sub_id = format!("panic-sub-{}", Uuid::new_v4().simple());
    let runner = ProjectionRunner::builder(store.clone())
        .poll_interval(Duration::from_millis(20))
        .max_attempts(2)
        .subscribe(sub_id.clone(), PanickingHandler)
        .build();

    // The panic must be contained: run() returns an error, the host lives on.
    let token = CancellationToken::new();
    let outcome = tokio::time::timeout(Duration::from_secs(30), runner.run(token))
        .await
        .expect("a panicking handler must stop the runner, not hang");
    assert!(
        outcome.is_err(),
        "a panicking handler must surface as a run error, got: {outcome:?}"
    );

    // Reaching here at all proves the host survived the handler panic. And the
    // checkpoint must not have advanced past the event that panicked.
    let last_position: Option<i64> =
        sqlx::query_scalar("SELECT last_position FROM es_subscriptions WHERE subscription_id = $1")
            .bind(&sub_id)
            .fetch_optional(&pool)
            .await
            .unwrap();
    assert_eq!(
        last_position.unwrap_or(0),
        0,
        "a panicking handler must not advance the checkpoint"
    );
}

#[tokio::test]
async fn subscription_polls_and_checkpoints() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let pool = store.pool().clone();
    let id = Uuid::new_v4().to_string();

    let mut counter = store.load_or_default::<Counter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 5 });
    store.save(&mut counter).await.unwrap();

    let sub_id = format!("test-sub-{}", Uuid::new_v4());
    let mut subscription = Subscription::create(store.clone(), pool, sub_id, 100)
        .await
        .unwrap();

    // Poll until our event surfaces: a concurrently-open transaction elsewhere
    // can pin the xmin horizon and briefly withhold it (by design).
    let mut seen = false;
    for _ in 0..400 {
        if subscription
            .poll_category("counter")
            .await
            .unwrap()
            .iter()
            .any(|e| e.stream_id == counter.stream_id)
        {
            seen = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
    assert!(seen, "our event should eventually be delivered");
    subscription.checkpoint().await.unwrap();
    assert!(subscription.position() > 0);
}

// --- Phase 0 regression tests ---------------------------------------------

/// C1/SNAP-1: a stream longer than the hydration page limit (1000) must
/// rebuild from its *full* history, not a truncated first page. Before the
/// fix, `load` folded only the first 1000 events while stamping the root with
/// the true version, so a command validated against the truncated state would
/// pass the OCC check and commit a wrong decision.
#[tokio::test]
async fn load_reads_streams_longer_than_the_page_limit() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();

    // 1001 events in a single append (> the 1000-row page size).
    const N: i64 = 1001;
    let mut counter = store.load_or_default::<Counter>(&id).await.unwrap();
    counter.record_many((0..N).map(|_| CounterEvent::Incremented { by: 1 }));
    store.save(&mut counter).await.unwrap();
    assert_eq!(counter.version, N);

    let reloaded = store.load::<Counter>(&id).await.unwrap().unwrap();
    assert_eq!(reloaded.version, N);
    assert_eq!(
        reloaded.state.total, N,
        "every event must be folded, not just the first page"
    );
}

/// C1 backstop (no DB): `hydrate` must reject a history with a gap rather than
/// silently fold partial state. Catches a truncated read or a partial-append
/// hole before it becomes a corrupt aggregate.
#[test]
fn hydrate_rejects_non_contiguous_history() {
    use mire::{AggregateRoot, RecordedEvent};

    let mk = |version: i64| RecordedEvent {
        global_position: version,
        stream_id: "counter-x".into(),
        stream_version: version,
        event_type: "counter.incremented".into(),
        data: serde_json::json!({ "type": "Incremented", "by": 1 }),
        metadata: serde_json::json!({}),
        transaction_id: 1,
        created_at: chrono::Utc::now(),
    };

    // Versions 1, 2, 4 — a gap at 3, recorded version 4.
    let events = vec![mk(1), mk(2), mk(4)];
    let res = AggregateRoot::<Counter>::hydrate("counter-x".into(), &events, 4);
    assert!(
        matches!(res, Err(mire::EventStoreError::StreamCorruption { .. })),
        "expected StreamCorruption for a version gap"
    );

    // Recorded version 5 but only 3 events read (truncation).
    let truncated = vec![mk(1), mk(2), mk(3)];
    let res = AggregateRoot::<Counter>::hydrate("counter-x".into(), &truncated, 5);
    assert!(
        matches!(res, Err(mire::EventStoreError::StreamCorruption { .. })),
        "expected StreamCorruption for a truncated read"
    );

    // A clean contiguous history hydrates fine.
    let good = vec![mk(1), mk(2), mk(3)];
    let root = AggregateRoot::<Counter>::hydrate("counter-x".into(), &good, 3).unwrap();
    assert_eq!(root.version, 3);
    assert_eq!(root.state.total, 3);
}

/// C8/CORE-3: a failed `save` must leave the pending events on the root so a
/// retry re-attempts the write. Before the fix, `take_pending` ran before the
/// append, so a conflict (or any error) dropped the events and a retry was a
/// silent no-op reporting success.
#[tokio::test]
async fn failed_save_keeps_pending_events_for_retry() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();

    // Establish the stream at version 1.
    let mut writer = store.load_or_default::<Counter>(&id).await.unwrap();
    writer.record(CounterEvent::Incremented { by: 1 });
    store.save(&mut writer).await.unwrap();

    // A stale root that still thinks the stream is empty (version 0).
    let mut stale = store.load_or_default::<Counter>(&id).await.unwrap();
    stale.version = 0; // force the NoStream/Exact(0) precondition to conflict
    stale.record(CounterEvent::Incremented { by: 5 });
    assert_eq!(stale.pending_count(), 1);

    let err = store.save(&mut stale).await.unwrap_err();
    assert!(matches!(
        err,
        mire::EventStoreError::ConcurrencyConflict { .. }
    ));
    assert_eq!(
        stale.pending_count(),
        1,
        "pending event must survive a failed save so a retry is not a no-op"
    );
    assert!(stale.has_pending());
}

/// CORE-14: an empty append still asserts the expected version (it is used as
/// a fence/precondition), instead of blindly returning the current version.
#[tokio::test]
async fn empty_append_enforces_expected_version() {
    let Some(store) = store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let id = Uuid::new_v4().to_string();
    let stream_id = format!("counter-{id}");

    let mut counter = store.load_or_default::<Counter>(&id).await.unwrap();
    counter.record(CounterEvent::Incremented { by: 1 });
    store.save(&mut counter).await.unwrap(); // stream now at version 1

    // Wrong expected version on an empty append must conflict.
    let err = store
        .append::<CounterEvent>(&stream_id, "counter", ExpectedVersion::Exact(0), &[])
        .await
        .unwrap_err();
    assert!(matches!(
        err,
        mire::EventStoreError::ConcurrencyConflict { .. }
    ));

    // Correct expected version on an empty append is a no-op returning current.
    let v = store
        .append::<CounterEvent>(&stream_id, "counter", ExpectedVersion::Exact(1), &[])
        .await
        .unwrap();
    assert_eq!(v, 1);
}