//! Signal retention (P3.5 / SAGA-14): `prune_consumed_signals` deletes
//! consumed `es_saga_signals` rows older than a cutoff, and never touches
//! pending (unconsumed) ones regardless of age.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use std::time::Duration;

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

mod common;

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

#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "ret.review-needed")]
struct ReviewNeeded {
    account_id: String,
}

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

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

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

// A step that awaits a routed signal — so a delivered signal is recorded in
// es_saga_signals and marked consumed once it drains.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "ret.decision")]
struct Decision {
    ok: bool,
}

fn await_decision(State(_): State<St>) -> Emit {
    Emit::none()
}

struct RetSaga;
impl Saga for RetSaga {
    type State = St;
    fn name() -> &'static str {
        "ret-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<Account, _>(|e: &ReviewNeeded| {
            Some(Start {
                instance_id: e.account_id.clone(),
                state: St {
                    account_id: e.account_id.clone(),
                },
            })
        })
        .step(
            "await_decision",
            task(await_decision)
                .on_response(|_: Decision| Out)
                .terminal_success(),
        )
        .signal::<Decision>()
        .routes_to_step("await_decision")
    }
}

#[tokio::test]
async fn prune_removes_old_consumed_signals_but_keeps_pending() {
    let Some(store) = common::store().await else {
        return;
    };
    let pool = store.pool().clone();

    // 1. Drive a saga so a routed signal is delivered + consumed. This
    //    leaves one CONSUMED row in es_saga_signals.
    let runner = PgSagaRunner::new(
        RetSaga::build(SagaBuilder::<RetSaga>::for_saga()),
        store.clone(),
    );
    let id = common::unique("ret");
    runner
        .trigger::<Account>(&ReviewNeeded {
            account_id: id.clone(),
        })
        .await
        .unwrap()
        .unwrap();
    let outcome = runner.signal(&id, Decision { ok: true }).await.unwrap();
    assert_eq!(outcome, mire_sagas::SignalOutcome::Accepted);
    assert!(runner.is_succeeded(&id).await.unwrap());

    // The consumed row exists. Age it well past the cutoff so it's eligible.
    let consumed_now: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_signals \
         WHERE saga_name = 'ret-saga' AND instance_id = $1 AND consumed_at IS NOT NULL",
    )
    .bind(&id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(
        consumed_now, 1,
        "the routed signal should be recorded + consumed"
    );
    sqlx::query(
        "UPDATE es_saga_signals SET consumed_at = now() - interval '2 hours' \
         WHERE saga_name = 'ret-saga' AND instance_id = $1",
    )
    .bind(&id)
    .execute(&pool)
    .await
    .unwrap();

    // 2. Insert a PENDING (unconsumed) signal, also aged old — retention
    //    must NOT touch it (a pending decision can't be lost to age).
    let pending_id = common::unique("ret-pending");
    sqlx::query(
        "INSERT INTO es_saga_signals \
         (saga_name, instance_id, signal_type, payload, shard_id, enqueued_at, consumed_at) \
         VALUES ('ret-saga', $1, 'ret.decision', '{}'::jsonb, 0, now() - interval '2 hours', NULL)",
    )
    .bind(&pending_id)
    .execute(&pool)
    .await
    .unwrap();

    // 3. Prune anything consumed older than 1 hour.
    let pruned = runner
        .prune_consumed_signals(Duration::from_secs(3600))
        .await
        .unwrap();
    assert!(pruned >= 1, "the aged consumed signal must be pruned");

    // The consumed row for our instance is gone…
    let consumed_after: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_signals \
         WHERE saga_name = 'ret-saga' AND instance_id = $1",
    )
    .bind(&id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(consumed_after, 0, "aged consumed signal should be deleted");

    // …but the aged PENDING row survives.
    let pending_after: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_signals \
         WHERE saga_name = 'ret-saga' AND instance_id = $1 AND consumed_at IS NULL",
    )
    .bind(&pending_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(
        pending_after, 1,
        "a pending (unconsumed) signal must never be pruned, regardless of age"
    );
}

#[tokio::test]
async fn prune_keeps_recently_consumed_signals() {
    let Some(store) = common::store().await else {
        return;
    };
    let pool = store.pool().clone();
    let runner = PgSagaRunner::new(
        RetSaga::build(SagaBuilder::<RetSaga>::for_saga()),
        store.clone(),
    );

    // A freshly-consumed signal (consumed_at = now) must survive a prune
    // with a 1-hour cutoff.
    let id = common::unique("ret-recent");
    sqlx::query(
        "INSERT INTO es_saga_signals \
         (saga_name, instance_id, signal_type, payload, shard_id, enqueued_at, consumed_at) \
         VALUES ('ret-saga', $1, 'ret.decision', '{}'::jsonb, 0, now(), now())",
    )
    .bind(&id)
    .execute(&pool)
    .await
    .unwrap();

    runner
        .prune_consumed_signals(Duration::from_secs(3600))
        .await
        .unwrap();

    let still_there: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_signals \
         WHERE saga_name = 'ret-saga' AND instance_id = $1",
    )
    .bind(&id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(
        still_there, 1,
        "a recently-consumed signal must not be pruned"
    );
}
