//! Route-layer contract tests for `routes::stripe::webhook::checkout`, the
//! session half of the exactly-once protocol.
//!
//! Stripe redelivers. It redelivers the same event id after a timeout, it
//! delivers a second event for the same checkout session when an asynchronous
//! payment settles, and it delivers events out of the order they happened in.
//! The money contract is that none of that pays anyone twice: one completed
//! transaction per session, one sales-count increment, one license key, one
//! purchase receipt, one Fan+ subscription row.
//!
//! What these tests pin, at the HTTP boundary the real deliveries arrive on:
//!   - a redelivered `checkout.session.completed` (same event id) is a no-op,
//!     down to `completed_at` being the same instant it was;
//!   - the same session arriving under a NEW event id, which defeats the
//!     event-id dedup, is still completed only once, because the
//!     status-guarded UPDATE underneath it refuses the second pass;
//!   - a one-time checkout that reports `payment_status: "unpaid"` delivers
//!     nothing until the settled event arrives (out-of-order settlement);
//!   - a forged signature is refused with 400 and writes nothing at all;
//!   - a redelivered Fan+ checkout creates one subscription.
//!
//! The invoice half is `stripe_webhook_billing_replay`.
//!
//! Delete this file and the suite keeps saying that one delivery of each event
//! works, which was never the risk.

use crate::harness::TestHarness;
use crate::harness::stripe::{TEST_WEBHOOK_SECRET, sign_webhook_payload};
use makenotwork::db::UserId;

/// The item price used throughout. Deliberately not 0, 1, or a round dollar:
/// 1499 tells cents from dollars ($14.99), tells a doubled credit (2998) from
/// a single one, and cannot agree with a sum that dropped a term.
const PRICE_CENTS: i64 = 1499;

// ── posting events ──────────────────────────────────────────────────────────

/// Sign an event envelope with the harness webhook secret and POST it.
async fn post_event(
    h: &mut TestHarness,
    event_id: &str,
    event_type: &str,
    object: serde_json::Value,
) -> crate::harness::client::TestResponse {
    let payload = serde_json::json!({
        "id": event_id,
        "type": event_type,
        "data": {"object": object},
    })
    .to_string();
    let signature = sign_webhook_payload(&payload, TEST_WEBHOOK_SECRET);

    // 503 is the route's documented "redeliver me" answer, not a failure: the
    // handler takes a `pg_try_advisory_xact_lock` on the event id and answers
    // 503 rather than parking a connection when that lock is held. sqlx rolls a
    // dropped transaction back lazily, so the lock from the delivery that just
    // returned can still be held for a moment, and a test that posts the
    // redelivery immediately would be asserting that timing rather than the
    // contract. Stripe's own answer to a 503 is to send the event again, so
    // that is what this does, bounded. The status the caller asserts is the
    // one the route settles on.
    for _ in 0..200 {
        let resp = post_signed(h, &payload, &signature).await;
        if resp.status != 503 {
            return resp;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("{event_id}: the webhook route answered 503 for two seconds");
}

/// POST a payload with a caller-supplied signature header, so a test can hand
/// over one that does not match the body.
async fn post_signed(
    h: &mut TestHarness,
    payload: &str,
    signature: &str,
) -> crate::harness::client::TestResponse {
    h.client
        .request_with_headers(
            "POST",
            "/stripe/webhook",
            Some(payload),
            &[
                ("stripe-signature", signature),
                ("content-type", "application/json"),
            ],
        )
        .await
}

// ── fixtures ────────────────────────────────────────────────────────────────

/// A seller with one license-key-enabled item, plus a buyer, plus a pending
/// transaction for `session_id` waiting on the webhook. Returns
/// `(buyer_id, seller_id, item_id)`.
async fn pending_purchase(
    h: &mut TestHarness,
    tag: &str,
    session_id: &str,
) -> (UserId, UserId, String) {
    let setup = h
        .create_creator_with_item(&format!("seller{tag}"), "audio", PRICE_CENTS)
        .await;

    // License keys are the mint-once oracle for the finalize step: the handler
    // mints at most one per transaction, so a second run of finalize shows up
    // here as a second row.
    sqlx::query("UPDATE items SET enable_license_keys = true WHERE id = $1::uuid")
        .bind(&setup.item_id)
        .execute(&h.db)
        .await
        .expect("enable license keys");

    h.client.post_form("/logout", "").await;
    let buyer_id = h
        .signup(
            &format!("buyer{tag}"),
            &format!("buyer{tag}@test.com"),
            "password123",
        )
        .await;

    sqlx::query(
        r"INSERT INTO transactions
           (buyer_id, seller_id, item_id, amount_cents, status,
            stripe_checkout_session_id, item_title, seller_username)
           VALUES ($1, $2, $3::uuid, $4, 'pending', $5, 'Test Item', 'seller')",
    )
    .bind(buyer_id)
    .bind(setup.user_id)
    .bind(&setup.item_id)
    .bind(PRICE_CENTS)
    .bind(session_id)
    .execute(&h.db)
    .await
    .expect("insert pending transaction");

    (buyer_id, setup.user_id, setup.item_id)
}

/// A `checkout.session.*` object for the one-time purchase path.
fn purchase_session(
    session_id: &str,
    buyer_id: UserId,
    seller_id: UserId,
    item_id: &str,
    payment_status: &str,
) -> serde_json::Value {
    serde_json::json!({
        "id": session_id,
        "object": "checkout.session",
        "payment_intent": "pi_exactly_once",
        "payment_status": payment_status,
        "currency": "usd",
        "amount_subtotal": PRICE_CENTS,
        "metadata": {
            "buyer_id": buyer_id.to_string(),
            "seller_id": seller_id.to_string(),
            "item_id": item_id,
        },
    })
}

// ── observations ────────────────────────────────────────────────────────────

/// Everything one checkout session's completion touches, read together so the
/// state after a redelivery can be compared with the state after the first
/// delivery field by field. A single-field oracle would miss the case where the
/// transaction is guarded but the counters are not.
struct PurchaseState {
    /// Transaction rows carrying this session id. More than one is a second charge.
    rows: i64,
    /// The transaction's status: `pending` until the completion write lands.
    status: String,
    /// When the completion write landed. A redelivery that re-ran it moves this,
    /// which is the difference between a guarded UPDATE and an unguarded one.
    completed_at: Option<chrono::DateTime<chrono::Utc>>,
    /// The item's denormalised sales counter, incremented once per purchase.
    sales_count: i32,
    /// License keys minted for the item. Enabled on the fixture item precisely
    /// so a second run of the finalize step shows up here as a second row
    /// rather than disappearing into an idempotent upsert.
    license_keys: i64,
}

async fn purchase_state(h: &TestHarness, session_id: &str, item_id: &str) -> PurchaseState {
    // MIN over the session's rows rather than `fetch_one`: `rows` is asserted
    // separately, so a duplicate row must be reported as a count rather than
    // blowing up the read that would have shown it.
    let (rows, status, completed_at): (i64, Option<String>, Option<chrono::DateTime<chrono::Utc>>) =
        sqlx::query_as(
            "SELECT COUNT(*), MIN(status), MIN(completed_at) FROM transactions \
         WHERE stripe_checkout_session_id = $1",
        )
        .bind(session_id)
        .fetch_one(&h.db)
        .await
        .expect("read the session's transaction rows");

    let sales_count: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
        .bind(item_id)
        .fetch_one(&h.db)
        .await
        .expect("read item sales count");

    let license_keys: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM license_keys WHERE item_id = $1::uuid")
            .bind(item_id)
            .fetch_one(&h.db)
            .await
            .expect("count license keys");

    PurchaseState {
        rows,
        status: status.unwrap_or_default(),
        completed_at,
        sales_count,
        license_keys,
    }
}

/// How many times the event-id dedup recorded this delivery. Two rows would
/// mean the dedup key stopped being the primary key it is.
async fn processed_event_rows(h: &TestHarness, event_id: &str) -> i64 {
    sqlx::query_scalar("SELECT COUNT(*) FROM processed_webhook_events WHERE event_id = $1")
        .bind(event_id)
        .fetch_one(&h.db)
        .await
        .expect("count processed webhook events")
}

/// Wait for the fire-and-forget mail tasks to settle and return the purchase
/// receipts sent to `address`.
///
/// The receipt is spawned on the background pool, so it lands after the webhook
/// response and cannot be read synchronously. This is a wait on a condition
/// rather than a fixed sleep: it returns as soon as the count has held at
/// `expected` for a stretch, and it fails the instant a second copy appears,
/// which is the failure the redelivery test is hunting.
async fn settled_receipts(h: &TestHarness, address: &str, expected: usize) -> usize {
    let transport = h
        .mock_email
        .as_ref()
        .expect("with_mocks configures a mock email transport");
    let mut stable = 0;
    for _ in 0..400 {
        let n = transport
            .sent_to(address)
            .into_iter()
            .filter(|e| e.subject == "Your purchase is confirmed")
            .count();
        assert!(
            n <= expected,
            "{address} received {n} purchase receipts, contract allows {expected}"
        );
        if n == expected {
            stable += 1;
            if stable == 20 {
                return n;
            }
        } else {
            stable = 0;
        }
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    }
    panic!("purchase receipts to {address} never settled at {expected}");
}

// ── checkout: redelivery ────────────────────────────────────────────────────

#[tokio::test]
async fn redelivered_checkout_session_credits_the_purchase_exactly_once() {
    let mut h = TestHarness::with_mocks().await;
    let session_id = "cs_exactly_once_replay";
    let (buyer_id, seller_id, item_id) = pending_purchase(&mut h, "replay", session_id).await;
    let event_id = "evt_exactly_once_replay";
    let session = purchase_session(session_id, buyer_id, seller_id, &item_id, "paid");

    let first = post_event(
        &mut h,
        event_id,
        "checkout.session.completed",
        session.clone(),
    )
    .await;
    assert_eq!(first.status, 200, "first delivery rejected: {}", first.text);

    let after_first = purchase_state(&h, session_id, &item_id).await;
    assert_eq!(
        after_first.status, "completed",
        "the pending transaction must be completed by the first delivery"
    );
    let n = after_first.sales_count;
    assert_eq!(n, 1, "one purchase is one sale, got {n}");
    let n = after_first.license_keys;
    assert_eq!(n, 1, "one purchase mints one license key, got {n}");
    assert_eq!(
        processed_event_rows(&h, event_id).await,
        1,
        "the event must be marked processed once the work committed"
    );

    // The redelivery Stripe actually sends: byte-identical event, same id.
    let second = post_event(&mut h, event_id, "checkout.session.completed", session).await;
    assert_eq!(second.status, 200, "redelivery rejected: {}", second.text);

    let after_second = purchase_state(&h, session_id, &item_id).await;
    let n = after_second.rows;
    assert_eq!(
        n, 1,
        "the session must hold exactly one transaction row, got {n}"
    );
    assert_eq!(
        after_second.completed_at, after_first.completed_at,
        "completed_at moved: the redelivery re-ran the completion write"
    );
    let n = after_second.sales_count;
    assert_eq!(n, 1, "the redelivery moved sales_count to {n}");
    let n = after_second.license_keys;
    assert_eq!(n, 1, "the redelivery left {n} license keys");
    assert_eq!(
        processed_event_rows(&h, event_id).await,
        1,
        "the processed-event mark must stay a single row"
    );

    // One receipt, and it quotes the credited amount in dollars-and-cents. A
    // handler that read 1499 as dollars, or credited twice, says something else.
    let receipts = settled_receipts(&h, "buyerreplay@test.com", 1).await;
    assert_eq!(receipts, 1, "exactly one purchase receipt for one purchase");
    let receipt = h
        .mock_email
        .as_ref()
        .unwrap()
        .sent_to("buyerreplay@test.com")
        .into_iter()
        .find(|e| e.subject == "Your purchase is confirmed")
        .expect("receipt captured");
    assert!(
        receipt.body.contains("$14.99"),
        "receipt must quote the 1499-cent price as $14.99, got: {}",
        receipt.body
    );
}

#[tokio::test]
async fn a_new_event_id_for_the_same_session_does_not_complete_it_twice() {
    let mut h = TestHarness::with_mocks().await;
    let session_id = "cs_exactly_once_newid";
    let (buyer_id, seller_id, item_id) = pending_purchase(&mut h, "newid", session_id).await;
    let session = purchase_session(session_id, buyer_id, seller_id, &item_id, "paid");

    let first = post_event(
        &mut h,
        "evt_exactly_once_newid_a",
        "checkout.session.completed",
        session.clone(),
    )
    .await;
    assert_eq!(first.status, 200, "first delivery failed: {}", first.text);
    let after_first = purchase_state(&h, session_id, &item_id).await;

    // A different event id defeats the event-id dedup on purpose: this is what
    // is left when that layer does not fire, and the status-guarded UPDATE
    // under it is the thing being pinned.
    let second = post_event(
        &mut h,
        "evt_exactly_once_newid_b",
        "checkout.session.completed",
        session,
    )
    .await;
    assert_eq!(second.status, 200, "second event rejected: {}", second.text);

    let after_second = purchase_state(&h, session_id, &item_id).await;
    assert_eq!(
        after_second.completed_at, after_first.completed_at,
        "the second event re-ran the completion write on an already-completed session"
    );
    let n = after_second.sales_count;
    assert_eq!(
        n, 1,
        "sales_count must stay 1 across two event ids, got {n}"
    );
    let n = after_second.license_keys;
    assert_eq!(
        n, 1,
        "crash-recovery finalize must mint no second key, got {n}"
    );
}

// ── checkout: out-of-order settlement ───────────────────────────────────────

#[tokio::test]
async fn an_unpaid_checkout_delivers_nothing_until_the_settled_event_arrives() {
    let mut h = TestHarness::with_mocks().await;
    let session_id = "cs_exactly_once_async";
    let (buyer_id, seller_id, item_id) = pending_purchase(&mut h, "async", session_id).await;

    // An asynchronous method (ACH, SEPA) reports `unpaid` on
    // checkout.session.completed. Funds are not captured, so nothing may ship.
    let unpaid = purchase_session(session_id, buyer_id, seller_id, &item_id, "unpaid");
    let deferred = post_event(
        &mut h,
        "evt_exactly_once_async_a",
        "checkout.session.completed",
        unpaid,
    )
    .await;
    assert_eq!(deferred.status, 200, "unsettled session: {}", deferred.text);

    let waiting = purchase_state(&h, session_id, &item_id).await;
    assert_eq!(
        waiting.status, "pending",
        "an unpaid session must leave the transaction pending"
    );
    let n = waiting.sales_count;
    assert_eq!(n, 0, "an unpaid session must not count a sale, got {n}");
    let n = waiting.license_keys;
    assert_eq!(
        n, 0,
        "an unpaid session must not mint a license key, got {n}"
    );

    // Stripe settles it later, under its own event id.
    let paid = purchase_session(session_id, buyer_id, seller_id, &item_id, "paid");
    let settled = post_event(
        &mut h,
        "evt_exactly_once_async_b",
        "checkout.session.async_payment_succeeded",
        paid,
    )
    .await;
    assert_eq!(
        settled.status, 200,
        "settled event rejected: {}",
        settled.text
    );

    let delivered = purchase_state(&h, session_id, &item_id).await;
    assert_eq!(
        delivered.status, "completed",
        "the settled event must complete the transaction"
    );
    let n = delivered.sales_count;
    assert_eq!(n, 1, "settlement counts exactly one sale, got {n}");
    let n = delivered.license_keys;
    assert_eq!(n, 1, "settlement mints exactly one license key, got {n}");
}

// ── checkout: signature ─────────────────────────────────────────────────────

#[tokio::test]
async fn a_forged_signature_is_refused_and_writes_nothing() {
    let mut h = TestHarness::with_mocks().await;
    let session_id = "cs_exactly_once_forged";
    let (buyer_id, seller_id, item_id) = pending_purchase(&mut h, "forged", session_id).await;

    let event_id = "evt_exactly_once_forged";
    let payload = serde_json::json!({
        "id": event_id,
        "type": "checkout.session.completed",
        "data": {"object": purchase_session(session_id, buyer_id, seller_id, &item_id, "paid")},
    })
    .to_string();

    // Correctly shaped header, signed with a secret we do not hold. The payload
    // is a real, fully valid completion, so only the signature separates this
    // from the accepted delivery above.
    let forged = sign_webhook_payload(&payload, "whsec_not_our_secret");
    let resp = post_signed(&mut h, &payload, &forged).await;
    assert_eq!(
        resp.status, 400,
        "a forged signature must be refused with 400: {}",
        resp.text
    );

    let state = purchase_state(&h, session_id, &item_id).await;
    assert_eq!(
        state.status, "pending",
        "a refused event must leave the transaction pending"
    );
    let n = state.sales_count;
    assert_eq!(n, 0, "a refused event must not count a sale, got {n}");
    let n = state.license_keys;
    assert_eq!(n, 0, "a refused event must not mint a license key, got {n}");
    assert_eq!(
        processed_event_rows(&h, event_id).await,
        0,
        "a refused event must not be marked processed"
    );
    let queued: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events")
        .fetch_one(&h.db)
        .await
        .expect("count retry queue");
    assert_eq!(
        queued, 0,
        "a signature failure is not a handler failure and must not enter the retry queue"
    );
}

// ── checkout: Fan+ subscription creation ────────────────────────────────────

#[tokio::test]
async fn redelivered_fan_plus_checkout_creates_one_subscription() {
    let mut h = TestHarness::with_mocks().await;
    let user_id = h
        .signup("fanplusonce", "fanplusonce@test.com", "password123")
        .await;

    let session = serde_json::json!({
        "id": "cs_exactly_once_fanplus",
        "object": "checkout.session",
        "subscription": "sub_exactly_once_fanplus",
        "customer": "cus_exactly_once_fanplus",
        "payment_status": "paid",
        "currency": "usd",
        "metadata": {"checkout_type": "fan_plus", "user_id": user_id.to_string()},
    });

    let first = post_event(
        &mut h,
        "evt_exactly_once_fanplus_a",
        "checkout.session.completed",
        session.clone(),
    )
    .await;
    assert_eq!(first.status, 200, "Fan+ checkout failed: {}", first.text);

    // Redelivery under the same id (dedup) and under a fresh id (the ON CONFLICT
    // guard beneath it): both must leave one active subscription, not two.
    let replay = post_event(
        &mut h,
        "evt_exactly_once_fanplus_a",
        "checkout.session.completed",
        session.clone(),
    )
    .await;
    assert_eq!(replay.status, 200, "Fan+ replay failed: {}", replay.text);
    let reissued = post_event(
        &mut h,
        "evt_exactly_once_fanplus_b",
        "checkout.session.completed",
        session,
    )
    .await;
    assert_eq!(reissued.status, 200, "Fan+ second event: {}", reissued.text);

    let (rows, status): (i64, Option<String>) = sqlx::query_as(
        "SELECT COUNT(*), MIN(status::text) FROM fan_plus_subscriptions WHERE user_id = $1",
    )
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .expect("read fan+ subscriptions");
    assert_eq!(
        rows, 1,
        "three deliveries of one Fan+ checkout must leave one subscription, got {rows}"
    );
    assert_eq!(status.as_deref(), Some("active"), "must still be active");
}