//! Chunk 4b: PgSagaRunner handles the same pure-decide cases the
//! in-memory runner does — gating (output + state), MergeKind::Any
//! merges, skip-cascade — through real Postgres.
//!
//! All tests gated on `DATABASE_URL`. The negative paths use
//! immediate-complete handlers so we stay in 4a/b scope.

#![allow(dead_code)]

use mire::{Aggregate, EventStore};
use mire_sagas::{
    Output, PgSagaRunner, Saga, SagaBuilder, SagaInstanceStatus, SagaStepStatus, Start, State, task,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;

// =====================================================================
// Trigger aggregate
// =====================================================================

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

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "order.placed")]
struct OrderPlaced {
    id: String,
    amount: i64,
}

impl Aggregate for Order {
    type Event = OrderPlaced;
    fn stream_category() -> &'static str {
        "pg-decide-order"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

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");
    let store = EventStore::new(pool);
    store.migrate().await.expect("mire migrate");
    mire_sagas::migrate(&store).await.expect("saga migrate");
    Some(store)
}

// =====================================================================
// 1. gated_by_state — skip when state predicate is false.
// =====================================================================

#[derive(Debug, Clone, Serialize, Deserialize)]
struct AutoOut;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ManualOut;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct DoneOut;

async fn auto_approve(State(_): State<OrderState>) -> AutoOut {
    AutoOut
}
async fn manual_review(State(_): State<OrderState>) -> ManualOut {
    ManualOut
}
async fn finalise(State(_): State<OrderState>) -> DoneOut {
    DoneOut
}

struct StateGateSaga;
impl Saga for StateGateSaga {
    type State = OrderState;
    fn name() -> &'static str {
        "pg-state-gate-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Order, _>(|e: &OrderPlaced| {
            Some(Start {
                instance_id: e.id.clone(),
                state: OrderState {
                    id: e.id.clone(),
                    amount: e.amount,
                },
            })
        })
        .step("auto", task(auto_approve))
        .step(
            "review",
            task(manual_review)
                .after("auto")
                .gated_by_state(|s: &OrderState| s.amount > 10_000),
        )
        .step(
            "finalise",
            task(finalise)
                .after_any_of(["review", "auto"])
                .terminal_success(),
        )
    }
}

#[tokio::test]
async fn state_gate_false_skips_step_and_any_of_proceeds() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = StateGateSaga::build(SagaBuilder::<StateGateSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("small-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Order>(&OrderPlaced {
            id: id.clone(),
            amount: 100,
        })
        .await
        .unwrap();
    let env = runner.load(&id).await.unwrap().unwrap();

    assert_eq!(
        env.step_status("review"),
        SagaStepStatus::Skipped,
        "amount=100 → review should skip"
    );
    assert_eq!(env.step_status("auto"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("finalise"), SagaStepStatus::Completed);
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
}

#[tokio::test]
async fn state_gate_true_runs_step() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = StateGateSaga::build(SagaBuilder::<StateGateSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("big-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Order>(&OrderPlaced {
            id: id.clone(),
            amount: 50_000,
        })
        .await
        .unwrap();
    let env = runner.load(&id).await.unwrap().unwrap();

    assert_eq!(env.step_status("review"), SagaStepStatus::Completed);
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
}

// =====================================================================
// 2. gated_by (output predicate) — skip when upstream output doesn't match.
// =====================================================================

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Decision {
    approve: bool,
}

async fn classify(State(_): State<OrderState>) -> Decision {
    Decision { approve: false }
}

async fn approve_action(Output(_): Output<Decision>) -> AutoOut {
    AutoOut
}

async fn reject_action(Output(_): Output<Decision>) -> ManualOut {
    ManualOut
}

async fn done(State(_): State<OrderState>) -> DoneOut {
    DoneOut
}

struct OutputGateSaga;
impl Saga for OutputGateSaga {
    type State = OrderState;
    fn name() -> &'static str {
        "pg-output-gate-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Order, _>(|e: &OrderPlaced| {
            Some(Start {
                instance_id: e.id.clone(),
                state: OrderState {
                    id: e.id.clone(),
                    amount: e.amount,
                },
            })
        })
        .step("classify", task(classify))
        .step(
            "approve",
            task(approve_action)
                .after("classify")
                .gated_by("classify", |d: &Decision| d.approve),
        )
        .step(
            "reject",
            task(reject_action)
                .after("classify")
                .gated_by("classify", |d: &Decision| !d.approve),
        )
        .step(
            "done",
            task(done)
                .after_any_of(["approve", "reject"])
                .terminal_success(),
        )
    }
}

#[tokio::test]
async fn output_gate_routes_to_the_matching_branch() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = OutputGateSaga::build(SagaBuilder::<OutputGateSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("branch-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Order>(&OrderPlaced {
            id: id.clone(),
            amount: 100,
        })
        .await
        .unwrap();
    let env = runner.load(&id).await.unwrap().unwrap();

    // classify returned approve=false → reject runs, approve skips.
    assert_eq!(env.step_status("classify"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("approve"), SagaStepStatus::Skipped);
    assert_eq!(env.step_status("reject"), SagaStepStatus::Completed);
    assert_eq!(env.step_status("done"), SagaStepStatus::Completed);
    assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
}

// =====================================================================
// 3. Skip-cascade — every after-dep was skipped → cascade.
// =====================================================================

struct CascadeSaga;
impl Saga for CascadeSaga {
    type State = OrderState;
    fn name() -> &'static str {
        "pg-cascade-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Order, _>(|e: &OrderPlaced| {
            Some(Start {
                instance_id: e.id.clone(),
                state: OrderState {
                    id: e.id.clone(),
                    amount: e.amount,
                },
            })
        })
        .step(
            "a",
            task(auto_approve).gated_by_state(|s: &OrderState| s.amount > 10_000),
        )
        // b after a — a is skipped, b cascades.
        .step("b", task(finalise).after("a").terminal_success())
    }
}

#[tokio::test]
async fn skip_cascade_propagates_through_after() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let topology = CascadeSaga::build(SagaBuilder::<CascadeSaga>::for_saga());
    let runner = PgSagaRunner::new(topology, store);
    let id = format!("cascade-{}", uuid::Uuid::new_v4());
    runner
        .trigger::<Order>(&OrderPlaced {
            id: id.clone(),
            amount: 1,
        })
        .await
        .unwrap();
    let env = runner.load(&id).await.unwrap().unwrap();

    assert_eq!(env.step_status("a"), SagaStepStatus::Skipped);
    assert_eq!(
        env.step_status("b"),
        SagaStepStatus::Skipped,
        "b cascades because all of its after-deps were skipped"
    );
    // No terminal-success was reached — saga remains Running. This
    // mirrors the in-memory behaviour: skipping a terminal step
    // doesn't itself confirm the saga.
    assert_eq!(env.status, SagaInstanceStatus::Running);
}
