//! P4.1 — shard configuration + startup guard.
//!
//! `with_num_shards` makes the runner stamp real `shard_id`s on the rows it
//! writes (awaits, signals), and `verify_shard_config` / `SagaWorker::run`
//! refuse to operate when the fleet disagrees on `num_shards` — i.e. when a
//! persisted row was stamped under more shards than this process is
//! configured for, which would otherwise silently strand the out-of-range
//! shards' work (plan §P4.1, principle 9).
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

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

mod common;

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

#[derive(Serialize, Deserialize, Clone, EventData)]
#[mire(event_type = "shard.requested")]
struct Requested {
    id: String,
}

impl Aggregate for Trigger {
    type Event = Requested;
    fn stream_category() -> &'static str {
        "shard-trigger"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

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

// One step that emits to an EXTERNAL worker (no handler) → the instance
// suspends awaiting the response, which writes an `es_saga_awaits` row
// stamped with `shard_of(instance_id, num_shards)`.
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "shard.work-req")]
struct WorkReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "shard.work-done")]
struct WorkDone;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct StepDone;

async fn do_work(State(s): State<St>) -> Emit {
    Emit::event(WorkReq { id: s.id.clone() })
}

struct ShardSaga;
impl Saga for ShardSaga {
    type State = St;
    fn name() -> &'static str {
        "shard-config-test"
    }
    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(do_work)
                .on_response(|_: WorkDone| StepDone)
                .terminal_success(),
        )
    }
}

fn runner(store: &mire::EventStore, num_shards: u32) -> PgSagaRunner<ShardSaga> {
    PgSagaRunner::new(
        ShardSaga::build(SagaBuilder::<ShardSaga>::for_saga()),
        store.clone(),
    )
    .with_num_shards(num_shards)
}

async fn max_await_shard(store: &mire::EventStore) -> i32 {
    sqlx::query_scalar::<_, Option<i32>>(
        "SELECT max(shard_id) FROM es_saga_awaits WHERE saga_name = $1",
    )
    .bind(ShardSaga::name())
    .fetch_one(store.pool())
    .await
    .unwrap()
    .unwrap_or(-1)
}

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

    // Stamp a spread of awaits across 64 shards. 40 instances over 64 shards
    // makes max(shard_id) >= 1 a statistical certainty (P(all land in shard
    // 0) = 64^-40), so the boundary assertions below are deterministic in
    // practice; the explicit `m >= 1` precondition fails loudly if that ever
    // breaks rather than silently mis-testing.
    const STAMP_SHARDS: u32 = 64;
    let big = runner(&store, STAMP_SHARDS);
    for _ in 0..40 {
        let id = common::unique("shard");
        big.trigger::<Trigger>(&Requested { id: id.clone() })
            .await
            .unwrap()
            .unwrap();
    }

    // A correctly-configured runner (matching or larger shard count) is fine.
    big.verify_shard_config().await.unwrap();
    runner(&store, STAMP_SHARDS + 1)
        .verify_shard_config()
        .await
        .unwrap();

    // Observe the actual max shard the rows landed in (independent of the
    // FNV-1a distribution), then probe the boundary exactly.
    let m = max_await_shard(&store).await;
    assert!(
        m >= 1,
        "expected at least one await beyond shard 0 across 40 instances / 64 shards, got max {m}"
    );
    let m = m as u32;

    // num_shards == m: row with shard_id m is `>= num_shards` → mismatch.
    match runner(&store, m).verify_shard_config().await {
        Err(PgRunnerError::ShardConfigMismatch {
            max_shard_id,
            num_shards,
            ..
        }) => {
            assert_eq!(max_shard_id as u32, m);
            assert_eq!(num_shards, m);
        }
        other => panic!("expected ShardConfigMismatch at num_shards={m}, got {other:?}"),
    }

    // num_shards == m + 1: every row's shard_id is `< num_shards` → ok.
    runner(&store, m + 1).verify_shard_config().await.unwrap();

    // A fleet that mistakenly dropped back to a single shard is rejected.
    assert!(matches!(
        runner(&store, 1).verify_shard_config().await,
        Err(PgRunnerError::ShardConfigMismatch { .. })
    ));
}

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

    // Stamp awaits under a wide shard count.
    let big = runner(&store, 64);
    for _ in 0..40 {
        let id = common::unique("shardw");
        big.trigger::<Trigger>(&Requested { id: id.clone() })
            .await
            .unwrap()
            .unwrap();
    }

    // A worker configured with a single shard must refuse to start rather
    // than silently never scan shards 1..64.
    let worker = SagaWorker::new(runner(&store, 1));
    let err = worker
        .run(CancellationToken::new())
        .await
        .expect_err("worker should refuse to start on shard config mismatch");
    assert!(
        matches!(err, PgRunnerError::ShardConfigMismatch { .. }),
        "got {err:?}"
    );
}

#[tokio::test]
#[should_panic(expected = "num_shards must be >= 1")]
async fn zero_shards_is_a_programming_error() {
    // No DB connection needed — builder-time validation fires before any IO.
    // (`connect_lazy` doesn't dial; it just needs a Tokio context to exist.)
    let _ = PgSagaRunner::new(
        ShardSaga::build(SagaBuilder::<ShardSaga>::for_saga()),
        // a store we never touch; new() only validates topology + this assert
        // fires before any IO.
        mire::EventStore::new(
            sqlx::postgres::PgPoolOptions::new()
                .connect_lazy("postgres://unused")
                .unwrap(),
        ),
    )
    .with_num_shards(0);
}
