//! P3.3 — lazy recovery re-drives instances stranded by a crash
//! (non-terminal, no pending await) back to quiescence.
//!
//! `commit` writes the envelope + await rows in one transaction, so the only
//! genuinely stranded states are the documented crash gaps: (1) after the
//! `Started` commit, before the first drive commit, and (2) after `tick_one`'s
//! `StepFailed` claim commit, before the compensation-cascade commit. Both are
//! reproduced here by writing the exact envelope event prefix with
//! `store.save` (which, unlike the runner's `commit`, does NOT create the
//! await/projection rows — mimicking a crash right after the events committed),
//! then seeding an aged `es_saga_instances` row and calling `recover`.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use mire::{Aggregate, AggregateRoot, EventData};
use mire_sagas::envelope::{SagaEnvelope, SagaEnvelopeEvent};
use mire_sagas::result::StepError;
use mire_sagas::{
    Emit, PgSagaRunner, Req, Saga, SagaBuilder, SagaInstanceStatus, SagaStepStatus, Start, State,
    comp, task,
};
use serde::{Deserialize, Serialize};

mod common;

#[derive(Default, Clone)]
struct Trigger;
#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "rec.requested")]
struct Requested {
    id: String,
}
impl Aggregate for Trigger {
    type Event = Requested;
    fn stream_category() -> &'static str {
        "rec-trigger"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "rec.work-req")]
struct WorkReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "rec.work-done")]
struct WorkDone;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct WorkOut;

fn work(State(s): State<St>) -> Emit {
    Emit::event(WorkReq { id: s.id.clone() })
}
async fn handle_work(Req(_): Req<WorkReq>) -> WorkDone {
    WorkDone
}

/// Single in-process step → driving from `Started` runs straight to terminal.
struct FwdSaga;
impl Saga for FwdSaga {
    type State = St;
    fn name() -> &'static str {
        "rec-fwd"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Trigger, _>(|e: &Requested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "work",
            task(work)
                .on_response(|_: WorkDone| WorkOut)
                .terminal_success(),
        )
        .effect(handle_work)
    }
}

async fn seed_instance_row(store: &mire::EventStore, saga: &str, id: &str, aged_secs: i64) {
    sqlx::query(
        "INSERT INTO es_saga_instances (saga_name, instance_id, status, shard_id, updated_at) \
         VALUES ($1, $2, 'running', 0, now() - ($3 || ' seconds')::interval) \
         ON CONFLICT (saga_name, instance_id) DO UPDATE \
         SET status = 'running', updated_at = now() - ($3 || ' seconds')::interval",
    )
    .bind(saga)
    .bind(id)
    .bind(aged_secs.to_string())
    .execute(store.pool())
    .await
    .unwrap();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn recovers_instance_stranded_after_started() {
    let Some(store) = common::store().await else {
        return;
    };
    let id = common::unique("recf");

    // Crash after `Started`: write only the Started event (status → Running),
    // no drive, no await row. Seed an aged projection row so it's a candidate.
    let mut root = AggregateRoot::<SagaEnvelope<FwdSaga>>::new(&id);
    root.record(SagaEnvelopeEvent::Started {
        state: serde_json::to_value(St { id: id.clone() }).unwrap(),
        trigger: Some(serde_json::to_value(Requested { id: id.clone() }).unwrap()),
    });
    store.save(&mut root).await.unwrap();
    seed_instance_row(&store, FwdSaga::name(), &id, 3600).await;

    let runner = PgSagaRunner::new(
        FwdSaga::build(SagaBuilder::<FwdSaga>::for_saga()),
        store.clone(),
    );

    // Precondition: stranded — Running, no await.
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::Running)
    );

    // `n >= 1` (not `== 1`): the shared test DB may hold other stranded
    // FwdSaga rows from parallel tests/reruns; assert on this instance.
    let n = runner.recover().await.unwrap();
    assert!(
        n >= 1,
        "recovery should re-drive at least the stranded instance"
    );
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::SucceededTerminal),
        "forward-recovery drove the in-process step to terminal"
    );
}

/// Distinct saga (own name) so this test's recovery scan can't be perturbed
/// by — or perturb — the FwdSaga instances other tests drive in parallel.
struct FreshSaga;
impl Saga for FreshSaga {
    type State = St;
    fn name() -> &'static str {
        "rec-fresh"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Trigger, _>(|e: &Requested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "work",
            task(work)
                .on_response(|_: WorkDone| WorkOut)
                .terminal_success(),
        )
        .effect(handle_work)
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn recovery_skips_a_live_leased_instance() {
    let Some(store) = common::store().await else {
        return;
    };
    let runner = PgSagaRunner::new(
        FreshSaga::build(SagaBuilder::<FreshSaga>::for_saga()),
        store.clone(),
    );

    // A stranded instance (Started, no steps yet, no await) — but with a LIVE
    // lease, i.e. a worker IS driving it. Phase 7: recovery is lease-expiry-
    // driven, so a live lease (not a staleness clock) is what shields an
    // in-flight instance from being re-driven out from under its worker.
    let leased = common::unique("recleased");
    let mut root = AggregateRoot::<SagaEnvelope<FreshSaga>>::new(&leased);
    root.record(SagaEnvelopeEvent::Started {
        state: serde_json::to_value(St { id: leased.clone() }).unwrap(),
        trigger: Some(serde_json::to_value(Requested { id: leased.clone() }).unwrap()),
    });
    store.save(&mut root).await.unwrap();
    seed_instance_row(&store, FreshSaga::name(), &leased, 0).await;
    // A live lease held by some other worker (valid 60s).
    sqlx::query(
        "INSERT INTO es_saga_leases (saga_name, instance_id, worker_id, leased_until, fence) \
         VALUES ($1, $2, 'other-worker', now() + interval '60 seconds', 1)",
    )
    .bind(FreshSaga::name())
    .bind(&leased)
    .execute(store.pool())
    .await
    .unwrap();

    runner.recover().await.unwrap();
    assert_eq!(
        runner.status(&leased).await.unwrap(),
        Some(SagaInstanceStatus::Running),
        "a live-leased instance must not be re-driven out from under its worker"
    );

    // Once the lease lapses, the same instance becomes fair game and recovery
    // drives it to terminal (its `work` step has an in-process handler).
    sqlx::query(
        "UPDATE es_saga_leases SET leased_until = now() - interval '1 second' \
         WHERE saga_name = $1 AND instance_id = $2",
    )
    .bind(FreshSaga::name())
    .bind(&leased)
    .execute(store.pool())
    .await
    .unwrap();
    runner.recover().await.unwrap();
    assert_eq!(
        runner.status(&leased).await.unwrap(),
        Some(SagaInstanceStatus::SucceededTerminal),
        "an expired-lease instance is fair game; recovery re-drives it to terminal"
    );
}

// ---- gap 2: StepFailed committed, cascade not (tick_one crash) ----

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "rec.hold-req")]
struct HoldReq;
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "rec.held")]
struct Held;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct HoldOut;
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "rec.rel-req")]
struct RelReq;
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "rec.rel")]
struct Rel;

fn hold(State(_): State<St>) -> Emit {
    Emit::event(HoldReq)
}
async fn handle_hold(Req(_): Req<HoldReq>) -> Held {
    Held
}
fn pay(State(_): State<St>) -> Emit {
    Emit::event(WorkReq { id: "x".into() })
}
fn release(State(_): State<St>) -> Emit {
    Emit::event(RelReq)
}
async fn handle_release(Req(_): Req<RelReq>) -> Rel {
    Rel
}

struct CompRecSaga;
impl Saga for CompRecSaga {
    type State = St;
    fn name() -> &'static str {
        "rec-comp"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Trigger, _>(|e: &Requested| {
            Some(Start {
                instance_id: e.id.clone(),
                state: St { id: e.id.clone() },
            })
        })
        .step(
            "hold",
            task(hold)
                .on_response(|_: Held| HoldOut)
                .compensate(comp(release).on_response::<Rel>()),
        )
        .step(
            "pay",
            task(pay)
                .after("hold")
                .on_response(|_: WorkDone| WorkOut)
                .terminal_success(),
        )
        .effect(handle_hold)
        .effect(handle_release)
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn recovers_cascade_after_stepfailed_claim() {
    let Some(store) = common::store().await else {
        return;
    };
    let id = common::unique("recc");

    // Crash between tick_one's claim commit and the cascade commit: hold is
    // Completed, pay is Failed (originating_error set), instance non-terminal,
    // no await rows. Recovery must re-run the compensation cascade → terminal.
    let mut root = AggregateRoot::<SagaEnvelope<CompRecSaga>>::new(&id);
    root.record(SagaEnvelopeEvent::Started {
        state: serde_json::to_value(St { id: id.clone() }).unwrap(),
        trigger: Some(serde_json::to_value(Requested { id: id.clone() }).unwrap()),
    });
    root.record(SagaEnvelopeEvent::StepActivated {
        step: "hold".into(),
    });
    root.record(SagaEnvelopeEvent::StepCompleted {
        step: "hold".into(),
        output: serde_json::Value::Null,
    });
    root.record(SagaEnvelopeEvent::StepActivated { step: "pay".into() });
    root.record(SagaEnvelopeEvent::StepFailed {
        step: "pay".into(),
        error: StepError::Timeout {
            step: "pay".into(),
            timeout_ms: 0,
        },
    });
    store.save(&mut root).await.unwrap();
    seed_instance_row(&store, CompRecSaga::name(), &id, 3600).await;

    let runner = PgSagaRunner::new(
        CompRecSaga::build(SagaBuilder::<CompRecSaga>::for_saga()),
        store.clone(),
    );
    assert_eq!(
        runner.status(&id).await.unwrap(),
        Some(SagaInstanceStatus::Running)
    );

    let n = runner.recover().await.unwrap();
    assert_eq!(n, 1, "recovery re-drove the stranded cascade");

    let env = runner.load(&id).await.unwrap().unwrap();
    assert_eq!(
        env.status,
        SagaInstanceStatus::FailedTerminal,
        "cascade resumed to terminal failure"
    );
    assert_eq!(
        env.steps.get("hold"),
        Some(&SagaStepStatus::Compensated),
        "the completed upstream was compensated by the resumed cascade"
    );
}
