//! Chunk 4d first slice: PgSagaRunner signals.
//!
//! Covers both signal policies through real Postgres:
//! - `TriggersCompensation` — an admin-cancel signal cancels the
//!   in-flight saga and runs the compensation cascade.
//! - `RoutesToStep` — a decision signal IS the awaited event for a
//!   suspended step; it resumes the saga.
//!
//! Plus the rejection cases (terminal, unregistered).
//! Gated on `DATABASE_URL`.

#![allow(dead_code)]

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

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

#[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
#[mire(event_type = "pg-sig.account-review")]
struct ReviewNeeded {
    id: String,
}

impl Aggregate for Account {
    type Event = ReviewNeeded;
    fn stream_category() -> &'static str {
        "pg-sig-account"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

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. RoutesToStep — the decision signal resumes a suspended step.
// =====================================================================

mod routes_to_step {
    use super::*;

    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-sig.decision")]
    struct Decision {
        approved: bool,
    }

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

    // Step emits nothing — just awaits the operator's Decision signal.
    async fn await_decision(State(_): State<ReviewState>) -> Emit {
        Emit::none()
    }

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

    async fn finalise(Output(o): Output<Outcome>) -> Done {
        let _ = o.approved;
        Done
    }

    struct ReviewSaga;
    impl Saga for ReviewSaga {
        type State = ReviewState;
        fn name() -> &'static str {
            "pg-sig-review-saga"
        }
        fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
            b.start_on::<Account, _>(|e: &ReviewNeeded| {
                Some(Start {
                    instance_id: e.id.clone(),
                    state: ReviewState { id: e.id.clone() },
                })
            })
            .step(
                "decision",
                task(await_decision).on_response(|d: Decision| Outcome {
                    approved: d.approved,
                }),
            )
            .step(
                "finalise",
                task(finalise).after("decision").terminal_success(),
            )
            .signal::<Decision>()
            .routes_to_step("decision")
        }
    }

    #[tokio::test]
    async fn decision_signal_resumes_suspended_step() {
        let Some(store) = maybe_store().await else {
            eprintln!("skipping: DATABASE_URL not set");
            return;
        };
        let topology = ReviewSaga::build(SagaBuilder::<ReviewSaga>::for_saga());
        let runner = PgSagaRunner::new(topology, store);
        let id = format!("review-{}", uuid::Uuid::new_v4());

        // Trigger → suspends awaiting Decision.
        runner
            .trigger::<Account>(&ReviewNeeded { id: id.clone() })
            .await
            .unwrap();
        let env = runner.load(&id).await.unwrap().unwrap();
        assert_eq!(env.status, SagaInstanceStatus::Running);
        assert_eq!(env.step_status("decision"), SagaStepStatus::Active);

        // Operator sends the decision signal.
        runner
            .signal(&id, Decision { approved: true })
            .await
            .expect("signal ok");

        // Saga resumed to terminal-success.
        let env = runner.load(&id).await.unwrap().unwrap();
        assert_eq!(env.status, SagaInstanceStatus::SucceededTerminal);
        assert_eq!(env.step_status("decision"), SagaStepStatus::Completed);
        assert_eq!(env.step_status("finalise"), SagaStepStatus::Completed);

        let outcome: Outcome =
            serde_json::from_value(env.outputs.get("decision").unwrap().clone()).unwrap();
        assert!(outcome.approved);
    }
}

// =====================================================================
// 2. TriggersCompensation — admin-cancel cancels + compensates.
// =====================================================================

mod triggers_compensation {
    use super::*;
    use mire_sagas::{Req, comp};

    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-sig.hold-req")]
    struct HoldRequest {
        id: String,
    }
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-sig.held")]
    struct Held {
        hold_id: String,
    }
    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct Reserved {
        hold_id: String,
    }

    async fn reserve(State(s): State<ReviewState>) -> Emit {
        Emit::event(HoldRequest { id: s.id.clone() })
    }
    async fn handle_hold(Req(r): Req<HoldRequest>) -> Held {
        Held {
            hold_id: format!("h-{}", r.id),
        }
    }
    async fn release(Output(_): Output<Reserved>) -> Emit {
        Emit::none()
    }

    // A second step that suspends awaiting an external event, so the
    // saga sits Running (not terminal) when the cancel arrives.
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-sig.confirm-req")]
    struct ConfirmRequest;
    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-sig.confirmed")]
    struct ConfirmedEv;
    #[derive(Debug, Clone, Serialize, Deserialize)]
    struct ConfirmOut;

    async fn confirm(Output(_): Output<Reserved>) -> Emit {
        Emit::event(ConfirmRequest)
    }

    #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
    #[mire(event_type = "pg-sig.admin-force-cancel")]
    struct AdminForceCancel {
        reason: String,
    }

    struct BookingSaga;
    impl Saga for BookingSaga {
        type State = ReviewState;
        fn name() -> &'static str {
            "pg-sig-booking-saga"
        }
        fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
            b.start_on::<Account, _>(|e: &ReviewNeeded| {
                Some(Start {
                    instance_id: e.id.clone(),
                    state: ReviewState { id: e.id.clone() },
                })
            })
            .step(
                "flight",
                task(reserve)
                    .on_response(|h: Held| Reserved { hold_id: h.hold_id })
                    .compensate(comp(release)),
            )
            // confirm has NO handler for ConfirmRequest → saga suspends.
            .step(
                "confirm",
                task(confirm)
                    .after("flight")
                    .terminal_success()
                    .on_response(|_: ConfirmedEv| ConfirmOut),
            )
            .handle(handle_hold)
            .signal::<AdminForceCancel>()
            .triggers_compensation()
        }
    }

    #[tokio::test]
    async fn admin_cancel_compensates_and_fails() {
        let Some(store) = maybe_store().await else {
            eprintln!("skipping: DATABASE_URL not set");
            return;
        };
        let topology = BookingSaga::build(SagaBuilder::<BookingSaga>::for_saga());
        let runner = PgSagaRunner::new(topology, store);
        let id = format!("booking-{}", uuid::Uuid::new_v4());

        // Trigger: flight completes inline (handler registered), confirm
        // suspends awaiting the external ConfirmedEv.
        runner
            .trigger::<Account>(&ReviewNeeded { id: id.clone() })
            .await
            .unwrap();
        let env = runner.load(&id).await.unwrap().unwrap();
        assert_eq!(env.step_status("flight"), SagaStepStatus::Completed);
        assert_eq!(env.step_status("confirm"), SagaStepStatus::Active);
        assert_eq!(env.status, SagaInstanceStatus::Running);

        // Admin cancels.
        runner
            .signal(
                &id,
                AdminForceCancel {
                    reason: "fraud".into(),
                },
            )
            .await
            .expect("signal ok");

        // flight compensated, saga failed-terminal.
        let env = runner.load(&id).await.unwrap().unwrap();
        assert_eq!(env.status, SagaInstanceStatus::FailedTerminal);
        assert_eq!(env.step_status("flight"), SagaStepStatus::Compensated);
    }

    #[tokio::test]
    async fn signal_to_terminal_saga_is_rejected() {
        let Some(store) = maybe_store().await else {
            eprintln!("skipping: DATABASE_URL not set");
            return;
        };
        let topology = BookingSaga::build(SagaBuilder::<BookingSaga>::for_saga());
        let runner = PgSagaRunner::new(topology, store);
        let id = format!("booking-term-{}", uuid::Uuid::new_v4());
        runner
            .trigger::<Account>(&ReviewNeeded { id: id.clone() })
            .await
            .unwrap();
        // Cancel once → FailedTerminal.
        runner
            .signal(
                &id,
                AdminForceCancel {
                    reason: "first".into(),
                },
            )
            .await
            .unwrap();
        // Second cancel → rejected (terminal).
        let res = runner
            .signal(
                &id,
                AdminForceCancel {
                    reason: "second".into(),
                },
            )
            .await;
        assert!(
            matches!(res, Err(PgRunnerError::SagaTerminal { .. })),
            "got {res:?}"
        );
    }

    #[tokio::test]
    async fn unregistered_signal_is_rejected() {
        let Some(store) = maybe_store().await else {
            eprintln!("skipping: DATABASE_URL not set");
            return;
        };
        let topology = BookingSaga::build(SagaBuilder::<BookingSaga>::for_saga());
        let runner = PgSagaRunner::new(topology, store);
        let id = format!("booking-unreg-{}", uuid::Uuid::new_v4());
        runner
            .trigger::<Account>(&ReviewNeeded { id: id.clone() })
            .await
            .unwrap();

        #[derive(Debug, Clone, Serialize, Deserialize, mire::EventData)]
        #[mire(event_type = "pg-sig.unregistered")]
        struct Unregistered;

        let res = runner.signal(&id, Unregistered).await;
        assert!(
            matches!(res, Err(PgRunnerError::UnregisteredSignal(_))),
            "got {res:?}"
        );
    }
}
