nordisk/monetize

crates/monetize-server/src/estate_order/tests/estate_webhook.rs

//! ★ **A32-26: ONE completion path for an estate's payment.** A sign-up's and a purchase's money
//! is settled from the webhook inbox; an estate's was settled only by the tick polling the vendor
//! — its verified webhook was answered "ignored" (no ledger attempt holds `estate/<co>`). Now the
//! estate's ONE path stays the driver's (the vendor asked, the money captured, `paid`), and its
//! webhook is a thin trigger of it: kept in the same inbox (a redelivery answered from the kept
//! row), the run driven at once. The tick still drives it when no webhook comes.

use super::*;

fn deliver_paid(
    r: &Rig,
    reference: &str,
    handle: &str,
    event_id: &str,
) -> (u16, serde_json::Value) {
    let body = serde_json::to_vec(&serde_json::json!({ "id": event_id, "event": "paid", "reference": reference, "handle": handle })).unwrap();
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()
        .unwrap();
    let state = r.state.clone();
    let company = reference.trim_start_matches("estate/").to_string();
    rt.block_on(async move {
        let resp = crate::callback::handle(
            axum::extract::State(state.clone()),
            axum::extract::Path("holdpay".to_string()),
            axum::http::HeaderMap::new(),
            axum::body::Bytes::from(body),
        )
        .await;
        let status = resp.status().as_u16();
        let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20)
            .await
            .unwrap();
        // The drive it queued runs on this runtime: let it finish before the runtime goes.
        for _ in 0..200 {
            if state
                .estates
                .get(&company)
                .is_some_and(|r| r.captured_at_ms > 0)
                && !state.estates.driving.lock().unwrap().contains(&company)
            {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
        }
        (
            status,
            serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null),
        )
    })
}

#[test]
fn an_estates_paid_webhook_is_kept_and_drives_its_one_completion_path_at_once() {
    let r = rig_with("estate-webhook", cfg(), "PASS");
    start_for(
        &r.state,
        "acme",
        "buyer@example.com",
        "",
        "alice",
        &EstateSizes::default(),
    )
    .unwrap();
    let run = r.state.estates.get("acme").unwrap();
    assert_eq!(run.phase, phase::ORDERED);
    let handle = handle_of(&r, "acme");
    r.pay.authorize(&handle);

    // No tick: the webhook alone sets the one path going.
    let (status, said) = deliver_paid(&r, &run.reference, &handle, "evt_acme_1");
    assert_eq!(status, 200, "{said}");
    assert!(
        said.get("ignored").is_none(),
        "an estate's payment is ours, never ignored: {said}"
    );
    assert_eq!(said["reference"], run.reference.as_str(), "{said}");
    let after = r.state.estates.get("acme").unwrap();
    assert!(
        after.captured_at_ms > 0,
        "captured on the driver's own path, at once: {:#?}",
        after.history
    );
    assert_ne!(after.phase, phase::ORDERED);
    let rows = r.state.callbacks.rows().unwrap();
    assert!(
        rows.iter().any(|row| row.reference == run.reference),
        "the delivery is kept in the one inbox: {rows:?}"
    );

    // Delivered again: answered from the kept row; nothing captured twice.
    let captures = r.pay.captures();
    let (status, again) = deliver_paid(&r, &run.reference, &handle, "evt_acme_1");
    assert_eq!(
        (status, again["duplicate"].clone()),
        (200, serde_json::Value::Bool(true)),
        "{again}"
    );
    assert_eq!(r.pay.captures(), captures, "one capture, whoever asks");
}

/// ★ **RA5-43: the webhook's trigger says QUEUED only when the drive was queued.** `drive_now`
/// threw away its submit's answer (`let _ = lane.submit(spec)`), so a refused submit was kept in
/// the inbox as Done "QUEUED … driven on its one completion path" and a redelivery was answered
/// from that false row. A refused submit (here a newer drive already holds the estate's lane) is
/// kept as KEPT: the tick drives it.
#[test]
fn a_webhook_whose_drive_is_not_queued_is_kept_not_queued() {
    let r = rig_with("estate-webhook-kept", cfg(), "PASS");
    start_for(
        &r.state,
        "acme",
        "buyer@example.com",
        "",
        "alice",
        &EstateSizes::default(),
    )
    .unwrap();
    let run = r.state.estates.get("acme").unwrap();
    let handle = handle_of(&r, "acme");
    r.pay.authorize(&handle);
    // The estate's lane, on a runtime nobody drives: a newer drive waits in it and never runs.
    let idle = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    {
        let _in = idle.enter();
        let lane = crate::estate_order::drive_lane(&r.state, "acme").unwrap();
        let newer = jera::jobs::chain::ChainSpec::new(
            crate::estate_order::DRIVE_JOB,
            "drive=acme-newer",
            "drive-acme",
            5,
            "monetize",
            serde_json::json!({ "company": "acme" }),
        );
        assert!(lane.submit(newer).is_new());
    }
    let (status, said) = deliver_paid(&r, &run.reference, &handle, "evt_acme_kept");
    assert_eq!(status, 200, "{said}");
    let rows = r.state.callbacks.rows().unwrap();
    let row = rows
        .iter()
        .find(|row| row.reference == run.reference)
        .expect("kept in the inbox");
    match &row.state {
        crate::inbox::InboxState::Done { state, .. } => assert_eq!(state, "KEPT", "{row:?}"),
        other => panic!("not closed: {other:?}"),
    }
}