//! P3.2 (slice 1) — external (Regime-B) emits are written to `es_saga_outbox`
//! in the same transaction as the `StepEmitted` event + await row. In-process
//! (Regime-A) emits are NOT (they're delivered inline, nothing to publish).
//!
//! Gated on `DATABASE_URL`; skips silently when unset.

#![allow(dead_code)]

use std::sync::{Arc, Mutex};

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

mod common;

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

// External step: emits ExtReq with NO registered handler → suspends (Regime B).
fn ext_work(State(s): State<St>) -> Emit {
    Emit::event(ExtReq { id: s.id.clone() })
}

struct ExtSaga;
impl Saga for ExtSaga {
    type State = St;
    fn name() -> &'static str {
        "outbox-ext"
    }
    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(),
        )
    }
}

// In-process step: same emit, but WITH a handler → Regime A, no outbox row.
fn inproc_work(State(s): State<St>) -> Emit {
    Emit::event(ExtReq { id: s.id.clone() })
}
async fn handle_ext(Req(_): Req<ExtReq>) -> ExtDone {
    ExtDone
}

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

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn external_emit_writes_an_unpublished_outbox_row() {
    let Some(store) = common::store().await else {
        return;
    };
    let runner = PgSagaRunner::new(
        ExtSaga::build(SagaBuilder::<ExtSaga>::for_saga()),
        store.clone(),
    );
    let id = common::unique("ob-ext");
    runner
        .trigger::<Trigger>(&Requested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();

    let rows: Vec<(
        String,
        String,
        serde_json::Value,
        Option<chrono::DateTime<chrono::Utc>>,
    )> = sqlx::query_as(
        "SELECT step, event_type, payload, published_at FROM es_saga_outbox \
             WHERE saga_name = $1 AND instance_id = $2",
    )
    .bind(ExtSaga::name())
    .bind(&id)
    .fetch_all(store.pool())
    .await
    .unwrap();

    assert_eq!(rows.len(), 1, "one external emit → one outbox row");
    let (step, event_type, payload, published_at) = &rows[0];
    assert_eq!(step, "work");
    assert_eq!(event_type, "ob.ext-req");
    assert_eq!(
        payload["id"],
        serde_json::json!(id),
        "the emitted request payload is captured"
    );
    assert!(
        published_at.is_none(),
        "unpublished until a publisher drains it"
    );
}

// A publisher saga gets its OWN name: `drain_outbox` is saga-wide, so a
// distinct name keeps this test's drain from marking the no-publisher test's
// row published (the saga-test-isolation rule).
struct PubSaga;
impl Saga for PubSaga {
    type State = St;
    fn name() -> &'static str {
        "outbox-pub"
    }
    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(),
        )
    }
}

#[derive(Clone)]
struct CapturePublisher {
    seen: Arc<Mutex<Vec<(String, String, serde_json::Value)>>>,
}
impl EmitPublisher for CapturePublisher {
    fn publish(
        &self,
        msg: OutboxMessage<'_>,
    ) -> impl std::future::Future<Output = anyhow::Result<()>> + Send {
        let rec = (
            msg.instance_id.to_string(),
            msg.event_type.to_string(),
            msg.payload.clone(),
        );
        let seen = self.seen.clone();
        async move {
            seen.lock().unwrap().push(rec);
            Ok(())
        }
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn publisher_drains_external_emits_then_marks_published() {
    let Some(store) = common::store().await else {
        return;
    };
    let seen = Arc::new(Mutex::new(Vec::new()));
    let runner = PgSagaRunner::new(
        PubSaga::build(SagaBuilder::<PubSaga>::for_saga()),
        store.clone(),
    )
    .with_publisher(CapturePublisher { seen: seen.clone() });
    assert!(runner.has_publisher());

    let id = common::unique("ob-pub");
    runner
        .trigger::<Trigger>(&Requested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();

    // Drain: the publisher receives our external emit, then the row is marked.
    let n = runner.drain_outbox().await.unwrap();
    assert!(n >= 1, "drain should publish at least our row");

    let got = seen
        .lock()
        .unwrap()
        .iter()
        .find(|(iid, _, _)| iid == &id)
        .cloned()
        .expect("publisher received our external emit");
    assert_eq!(got.1, "ob.ext-req");
    assert_eq!(got.2["id"], serde_json::json!(id));

    // Our row is now published.
    let unpublished: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_outbox \
         WHERE saga_name = $1 AND instance_id = $2 AND published_at IS NULL",
    )
    .bind(PubSaga::name())
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(unpublished, 0, "publish-then-mark stamped published_at");

    // Idempotent: a second drain does not re-publish our (now-published) row.
    runner.drain_outbox().await.unwrap();
    let mine = seen
        .lock()
        .unwrap()
        .iter()
        .filter(|(iid, _, _)| iid == &id)
        .count();
    assert_eq!(mine, 1, "a published row is not re-delivered");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn in_process_emit_writes_no_outbox_row() {
    let Some(store) = common::store().await else {
        return;
    };
    let runner = PgSagaRunner::new(
        InProcSaga::build(SagaBuilder::<InProcSaga>::for_saga()),
        store.clone(),
    );
    let id = common::unique("ob-inproc");
    runner
        .trigger::<Trigger>(&Requested { id: id.clone() })
        .await
        .unwrap()
        .unwrap();

    let count: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_outbox WHERE saga_name = $1 AND instance_id = $2",
    )
    .bind(InProcSaga::name())
    .bind(&id)
    .fetch_one(store.pool())
    .await
    .unwrap();
    assert_eq!(
        count, 0,
        "in-process-handled emit is delivered inline — nothing to publish"
    );
}
