//! Regression for review findings C1 + C2: when two steps emit in
//! parallel and ONE handler returns Err, the runner must fail the
//! source step that emitted that request — not the alphabetically-first
//! step in the awaiting map.
//!
//! Pre-fix, `invoke_and_deliver` used
//! `instance.awaiting.keys().next().copied()` to pick which step to
//! fail. BTreeMap keys are sorted, so under parallel emits with
//! step names "a" and "b", an error from b's handler would
//! deterministically fail "a".
//!
//! This test sets up:
//!   - step `alpha` and step `bravo`, both activate in parallel.
//!   - `bravo`'s handler returns Err; `alpha`'s handler succeeds.
//!   - Expectation: `bravo` is the failed step. `alpha` is Completed.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use mire::{Aggregate, EventData};
use mire_sagas::{
    Emit, PgSagaRunner, Req, Saga, SagaBuilder, SagaInstanceStatus, SagaStepStatus, Start, State,
    task,
};
use serde::{Deserialize, Serialize};

mod common;

#[derive(Default, Clone)]
struct Source;

#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "source.event")]
struct SourceEvent {
    id: String,
}

impl Aggregate for Source {
    type Event = SourceEvent;
    fn stream_category() -> &'static str {
        "source"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

#[derive(Default, Clone, Serialize, Deserialize)]
struct St {
    id: String,
}

// =====================================================================
// Step alpha: emits AlphaRequest, awaits AlphaOk
// =====================================================================

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "alpha.req")]
struct AlphaRequest {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "alpha.ok")]
struct AlphaOk;

#[derive(Debug, Clone, Serialize, Deserialize)]
struct AlphaDone;

fn alpha_step(State(s): State<St>) -> Emit {
    Emit::event(AlphaRequest { id: s.id.clone() })
}

async fn handle_alpha(Req(_): Req<AlphaRequest>) -> AlphaOk {
    AlphaOk
}

// =====================================================================
// Step bravo: emits BravoRequest, awaits BravoOk. Handler ERRORS.
// =====================================================================

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "bravo.req")]
struct BravoRequest {
    id: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "bravo.ok")]
struct BravoOk;

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "bravo.err")]
struct BravoErr {
    reason: String,
}

impl std::fmt::Display for BravoErr {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.reason)
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BravoDone;

fn bravo_step(State(s): State<St>) -> Emit {
    Emit::event(BravoRequest { id: s.id.clone() })
}

async fn handle_bravo(Req(_): Req<BravoRequest>) -> Result<BravoOk, BravoErr> {
    Err(BravoErr {
        reason: "bravo always errs".into(),
    })
}

// =====================================================================
// Saga: alpha and bravo run in parallel (no .after between them).
// =====================================================================

struct ParallelSaga;
impl Saga for ParallelSaga {
    type State = St;
    fn name() -> &'static str {
        "parallel-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Source, _>(|e: &SourceEvent| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "alpha",
            task(alpha_step).on_response(|_: AlphaOk| AlphaDone),
        )
        .step(
            "bravo",
            task(bravo_step)
                .terminal_success()
                .on_response(|_: BravoOk| BravoDone),
        )
        .effect(handle_alpha)
        .effect(handle_bravo)
    }
}

// Mirror saga to verify naming order doesn't matter — if bravo's
// handler is the one that errors, ALPHA shouldn't get blamed
// (which is exactly the pre-fix bug since "alpha" < "bravo").
struct ParallelSagaReversed;
impl Saga for ParallelSagaReversed {
    type State = St;
    fn name() -> &'static str {
        "parallel-saga-reversed"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Source, _>(|e: &SourceEvent| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        // Now the FAILING step is alphabetically first. If routing
        // was correct, "alpha" should still survive and the (alphabetically
        // second) "bravo" should be the one that fails. But in this
        // saga the FAILING handler is on alpha; the SUCCESSFUL handler
        // is on bravo. So this also proves nothing's relying on order.
        //
        // Concretely: this saga has `failing` step name (alphabetically
        // FIRST) erring, and `surviving` (second) succeeding.
        .step(
            "failing",
            task(bravo_step).on_response(|_: BravoOk| BravoDone),
        )
        .step(
            "surviving",
            task(alpha_step)
                .terminal_success()
                .on_response(|_: AlphaOk| AlphaDone),
        )
        .effect(handle_alpha)
        .effect(handle_bravo)
    }
}

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

#[tokio::test]
async fn err_response_fails_the_source_step_not_the_first_in_btreemap() {
    let Some(store) = common::store().await else {
        return;
    };
    let topology = ParallelSaga::build(SagaBuilder::<ParallelSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = runner
        .trigger::<Source>(&SourceEvent {
            id: common::unique("parallel-1"),
        })
        .await
        .unwrap()
        .unwrap();

    // bravo's handler errored — bravo should be Failed, NOT alpha.
    assert_eq!(
        runner.step_status(&id, "bravo").await.unwrap(),
        SagaStepStatus::Failed,
        "the step whose handler erred should be Failed"
    );

    // alpha's handler succeeded — alpha should be Completed.
    assert_eq!(
        runner.step_status(&id, "alpha").await.unwrap(),
        SagaStepStatus::Completed,
        "the step whose handler succeeded should be Completed — NOT mis-blamed for bravo's failure"
    );

    // Saga overall reached FailedTerminal because bravo was terminal_success
    // and it failed; compensation cascade runs (nothing to compensate since
    // alpha had no compensation registered).
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::FailedTerminal)
    );
}

#[tokio::test]
async fn err_routes_correctly_when_alphabetically_first_step_is_the_failing_one() {
    // In this saga, the alphabetically-first step has the erring
    // handler. Pre-fix this happened to "work" coincidentally (the
    // mis-routing pointed at the right step by accident). This test
    // makes sure neither direction relies on alphabetical ordering.
    let Some(store) = common::store().await else {
        return;
    };
    let topology = ParallelSagaReversed::build(SagaBuilder::<ParallelSagaReversed>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = runner
        .trigger::<Source>(&SourceEvent {
            id: common::unique("parallel-2"),
        })
        .await
        .unwrap()
        .unwrap();

    // `failing` step uses bravo_step which emits BravoRequest; bravo's
    // handler errors. So `failing` should be Failed — never the sibling.
    assert_eq!(
        runner.step_status(&id, "failing").await.unwrap(),
        SagaStepStatus::Failed
    );
    // The durable runner drives one step at a time and stops the drive
    // when a failure concludes the saga, so the not-yet-activated sibling
    // stays Pending (it is NOT mis-blamed as Failed). The wave-based
    // drive loop (tasks/effects-vs-commit.md §3) will activate ready
    // siblings before any effect runs; until then Pending is the
    // documented outcome here.
    assert_eq!(
        runner.step_status(&id, "surviving").await.unwrap(),
        SagaStepStatus::Pending
    );
}
