//! P3.2 — a sharded `SagaWorker` fleet drains `es_saga_outbox` across shards
//! via its `EmitPublisher`, delivering every external emit (no duplicates with
//! publish-then-mark + per-shard ownership), with no host involvement.
//!
//! Complements `outbox.rs`'s single-runner `drain_outbox` test: this exercises
//! the worker's drain tick + shard-filtered `drain_outbox_in_shards` under a
//! multi-worker fleet — the path slice 2 wired but didn't yet test.
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use mire::{Aggregate, EventData};
use mire_sagas::{
    Emit, EmitPublisher, OutboxMessage, 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 = "obf.requested")]
struct Requested {
    id: String,
}
impl Aggregate for Trigger {
    type Event = Requested;
    fn stream_category() -> &'static str {
        "obf-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 = "obf.ext-req")]
struct ExtReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "obf.ext-done")]
struct ExtDone;
#[derive(Debug, Clone, Serialize, Deserialize)]
struct StepDone;

// External step (no handler) → emits ExtReq and suspends. No deadline, so it
// just waits — the test only cares that its outbox emit gets delivered.
async fn ext_work(State(s): State<St>) -> Emit {
    Emit::event(ExtReq { id: s.id.clone() })
}

struct FleetSaga;
impl Saga for FleetSaga {
    type State = St;
    fn name() -> &'static str {
        "outbox-fleet"
    }
    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(ext_work)
                .on_response(|_: ExtDone| StepDone)
                .terminal_success(),
        )
    }
}

const N_SHARDS: u32 = 8;

// Records every published message; tagged with which worker delivered it so we
// can assert the drain was actually distributed.
#[derive(Clone)]
struct FleetPublisher {
    worker: String,
    seen: Arc<Mutex<Vec<(String, String)>>>, // (instance_id, publisher worker tag)
}
impl EmitPublisher for FleetPublisher {
    fn publish(
        &self,
        msg: OutboxMessage<'_>,
    ) -> impl std::future::Future<Output = anyhow::Result<()>> + Send {
        let rec = (msg.instance_id.to_string(), self.worker.clone());
        let seen = self.seen.clone();
        async move {
            seen.lock().unwrap().push(rec);
            Ok(())
        }
    }
}

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

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

    // Trigger M instances; each suspends on the external step, writing one
    // outbox row stamped with its shard.
    const M: usize = 60;
    let trig = runner(&store);
    let mut ids = Vec::new();
    for _ in 0..M {
        let id = common::unique("obf");
        trig.trigger::<Trigger>(&Requested { id: id.clone() })
            .await
            .unwrap()
            .unwrap();
        ids.push(id);
    }

    // Fleet of 3 workers, each with its own publisher tag but a SHARED sink,
    // self-partitioning the 8 shards. Each owns disjoint shards, so each
    // outbox row is published by exactly one worker.
    let seen = Arc::new(Mutex::new(Vec::new()));
    let cancel = CancellationToken::new();
    let mut handles = Vec::new();
    for w in 0..3 {
        let publisher = FleetPublisher {
            worker: format!("w{w}"),
            seen: seen.clone(),
        };
        let worker = SagaWorker::new(runner(&store).with_publisher(publisher))
            .poll_interval(Duration::from_millis(50))
            .shard_lease_ttl(Duration::from_secs(2))
            .drain_interval(Duration::from_millis(50));
        let c = cancel.clone();
        handles.push(tokio::spawn(async move { worker.run(c).await }));
    }

    // Wait until every one of THIS run's outbox rows is published.
    let mut all_published = false;
    let prefix = "obf-%";
    for _ in 0..200 {
        let unpublished: i64 = sqlx::query_scalar(
            "SELECT count(*) FROM es_saga_outbox \
             WHERE saga_name = $1 AND instance_id LIKE $2 AND published_at IS NULL",
        )
        .bind(FleetSaga::name())
        .bind(prefix)
        .fetch_one(store.pool())
        .await
        .unwrap();
        if unpublished == 0 {
            all_published = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }

    // Snapshot deliveries before stopping the fleet.
    let delivered = seen.lock().unwrap().clone();
    cancel.cancel();
    for h in handles {
        let _ = h.await;
    }

    assert!(all_published, "the fleet must publish every outbox row");

    // Every instance delivered exactly once (publish-then-mark + disjoint
    // shard ownership ⇒ no duplicates absent a crash).
    let mut per_instance: HashMap<&str, usize> = HashMap::new();
    for (iid, _) in &delivered {
        *per_instance.entry(iid.as_str()).or_default() += 1;
    }
    for id in &ids {
        assert_eq!(
            per_instance.get(id.as_str()).copied().unwrap_or(0),
            1,
            "instance {id} should be delivered exactly once, got {:?}",
            per_instance.get(id.as_str())
        );
    }
    // (We don't assert the work spread across ≥2 workers: with a small,
    // fast-draining batch a single worker that wins all shards first can
    // legitimately drain everything before peers finish converging. The
    // shard split itself is covered by `saga_worker_sharded`; here the point
    // is exactly-once delivery across the fleet's shared publisher.)
}
