//! Postgres-runner Chunk 2: `SagaEnvelope<S>` is a mire aggregate.
//!
//! This test locks the contract: a SagaEnvelope built by appending
//! envelope events through the existing `EventStore::save` pipeline
//! and reloaded via `EventStore::load` reconstructs to the same
//! state.
//!
//! Gated on `DATABASE_URL`, like the rest of the mire integration
//! suite. Without it, the test is a no-op (eprintln + return) so it
//! doesn't block CI on machines without Postgres.

#![allow(dead_code)]

use mire::{Aggregate, EventStore};
use mire_sagas::envelope::SagaEnvelopeEvent;
use mire_sagas::{Saga, SagaBuilder, SagaEnvelope, SagaInstanceStatus, SagaStepStatus};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;

// =====================================================================
// Minimal Saga type — only the trait impl matters; we drive the envelope
// directly via SagaEnvelopeEvent::* and never touch the runner.
// =====================================================================

#[derive(Default, Clone, Serialize, Deserialize, Debug, PartialEq)]
struct DemoState {
    counter: i64,
}

struct DemoSaga;
impl Saga for DemoSaga {
    type State = DemoState;
    fn name() -> &'static str {
        "agg-roundtrip-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b
    }
}

// =====================================================================
// Helpers
// =====================================================================

async fn maybe_store() -> Option<EventStore> {
    let url = std::env::var("DATABASE_URL")
        .ok()
        .filter(|s| !s.is_empty())?;
    let pool = PgPool::connect(&url).await.expect("connect to postgres");
    let store = EventStore::new(pool);
    store.migrate().await.expect("migrate");
    Some(store)
}

fn stream_id(instance: &str) -> String {
    format!(
        "{}-{}",
        <SagaEnvelope<DemoSaga> as Aggregate>::stream_category(),
        instance
    )
}

// =====================================================================
// Tests
// =====================================================================

#[test]
fn stream_category_matches_saga_name() {
    // Pure assertion — no DB. Confirms the Aggregate impl wires
    // category to S::name() so each saga lives in its own category
    // space.
    let cat = <SagaEnvelope<DemoSaga> as Aggregate>::stream_category();
    assert_eq!(cat, "agg-roundtrip-saga");
}

#[test]
fn replay_via_apply_matches_in_memory_state() {
    // Round-trip in memory: start with default envelope, apply a sequence
    // of events, verify the resulting state. Mirrors what EventStore::load
    // does internally via AggregateRoot::hydrate.
    let mut env: SagaEnvelope<DemoSaga> = SagaEnvelope::default();
    let initial_state = DemoState { counter: 7 };

    let events = vec![
        SagaEnvelopeEvent::Started {
            state: serde_json::to_value(&initial_state).unwrap(),
            trigger: None,
        },
        SagaEnvelopeEvent::StepActivated {
            step: "step1".into(),
        },
        SagaEnvelopeEvent::StepCompleted {
            step: "step1".into(),
            output: serde_json::json!({"ok": true}),
        },
        SagaEnvelopeEvent::Confirmed,
    ];

    for ev in &events {
        env.apply(ev);
    }

    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
    assert_eq!(env.state, initial_state, "state must replay from Started");
    assert_eq!(
        env.step_status("step1"),
        SagaStepStatus::Completed,
        "step1 must be Completed after StepCompleted"
    );
    assert_eq!(
        env.outputs.get("step1"),
        Some(&serde_json::json!({"ok": true}))
    );
}

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

    let instance = uuid::Uuid::new_v4().to_string();
    let initial_state = DemoState { counter: 42 };

    // Build a fresh envelope aggregate root and record events on it,
    // exactly as the durable runner will do.
    let mut root: mire::AggregateRoot<SagaEnvelope<DemoSaga>> = mire::AggregateRoot::new(&instance);
    root.record(SagaEnvelopeEvent::Started {
        state: serde_json::to_value(&initial_state).unwrap(),
        trigger: None,
    });
    root.record(SagaEnvelopeEvent::StepActivated {
        step: "step1".into(),
    });
    root.record(SagaEnvelopeEvent::StepCompleted {
        step: "step1".into(),
        output: serde_json::json!({"ok": true}),
    });
    root.record(SagaEnvelopeEvent::Confirmed);

    store.save(&mut root).await.expect("save");

    // Reload via EventStore::load — the durable runner's load path.
    let reloaded = store
        .load::<SagaEnvelope<DemoSaga>>(&instance)
        .await
        .expect("load")
        .expect("instance must exist after save");

    // Same shape as the in-memory replay test: state, status, step
    // statuses, outputs all reconstructed.
    assert_eq!(reloaded.version, 4, "four events committed");
    assert_eq!(reloaded.state.status, SagaInstanceStatus::SucceededTerminal);
    assert_eq!(reloaded.state.state, initial_state);
    assert_eq!(
        reloaded.state.step_status("step1"),
        SagaStepStatus::Completed
    );
    assert_eq!(
        reloaded.state.outputs.get("step1"),
        Some(&serde_json::json!({"ok": true}))
    );
}

#[tokio::test]
async fn envelope_two_appends_advance_version_with_occ() {
    // Locks the OCC contract: appending two batches of events to the
    // same envelope stream advances stream_version atomically. The
    // durable runner's per-decision commit relies on this exact
    // behaviour for replica safety (one writer per instance per
    // shard; OCC catches lease-handoff races).
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    let instance = uuid::Uuid::new_v4().to_string();

    // First batch — Started + StepActivated.
    let mut root: mire::AggregateRoot<SagaEnvelope<DemoSaga>> = mire::AggregateRoot::new(&instance);
    root.record(SagaEnvelopeEvent::Started {
        state: serde_json::to_value(DemoState::default()).unwrap(),
        trigger: None,
    });
    root.record(SagaEnvelopeEvent::StepActivated { step: "a".into() });
    store.save(&mut root).await.unwrap();
    assert_eq!(root.version, 2);

    // Second batch — re-load, record more, save.
    let mut root = store
        .load::<SagaEnvelope<DemoSaga>>(&instance)
        .await
        .unwrap()
        .unwrap();
    root.record(SagaEnvelopeEvent::StepCompleted {
        step: "a".into(),
        output: serde_json::json!(null),
    });
    store.save(&mut root).await.unwrap();
    assert_eq!(root.version, 3);

    // Now simulate a concurrent racing writer: stale root from the
    // first load, tries to record more and save. OCC must reject.
    let mut stale_root = mire::AggregateRoot::<SagaEnvelope<DemoSaga>>::new(&instance);
    stale_root.version = 2; // pretend we're still at version 2
    stale_root.record(SagaEnvelopeEvent::Cancelled {
        reason: "stale write".into(),
    });
    let res = store.save(&mut stale_root).await;
    assert!(
        res.is_err(),
        "concurrent stale write should fail OCC, got: {res:?}"
    );
}
