//! Stripe webhook workflow tests, purchase, refund, account update,
//! invalid signature, subscription lifecycle.

use crate::harness::TestHarness;
use crate::harness::stripe::{TEST_WEBHOOK_SECRET, TEST_WEBHOOK_SECRET_V2, sign_webhook_payload};
use makenotwork::db::UserId;
use serde_json::Value;
use std::collections::HashMap;

/// Build a JSON event with the given type and object, sign it, and POST to /stripe/webhook.
async fn post_event_json(
    h: &mut TestHarness,
    event_type: &str,
    object: serde_json::Value,
) -> crate::harness::client::TestResponse {
    post_event_json_with_id(h, "evt_test_000", event_type, object).await
}

async fn post_event_json_with_id(
    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);

    h.client
        .request_with_headers(
            "POST",
            "/stripe/webhook",
            Some(&payload),
            &[
                ("stripe-signature", &signature),
                ("content-type", "application/json"),
            ],
        )
        .await
}

// Tests

#[tokio::test]
async fn webhook_invalid_signature() {
    let mut h = TestHarness::with_stripe().await;

    let payload = r#"{"id":"evt_bad","type":"account.updated","data":{"object":{}}}"#;
    let bad_sig = "t=0,v1=00000000000000000000000000000000";

    let resp = h
        .client
        .request_with_headers(
            "POST",
            "/stripe/webhook",
            Some(payload),
            &[
                ("stripe-signature", bad_sig),
                ("content-type", "application/json"),
            ],
        )
        .await;
    assert_eq!(
        resp.status.as_u16(),
        400,
        "Expected 400 for bad signature, got: {}",
        resp.status
    );
}

#[tokio::test]
async fn webhook_account_updated() {
    let mut h = TestHarness::with_stripe().await;

    // Create a user with a known stripe_account_id
    let user_id = h
        .signup("stripecreator", "sc@test.com", "password123")
        .await;
    let acct_id = "acct_test_wh_123";
    sqlx::query("UPDATE users SET stripe_account_id = $1 WHERE id = $2")
        .bind(acct_id)
        .bind(user_id)
        .execute(&h.db)
        .await
        .unwrap();

    // Build account object with valid id prefix
    let account = serde_json::json!({
        "id": acct_id,
        "object": "account",
        "charges_enabled": true,
        "payouts_enabled": true,
        "details_submitted": true,
    });

    let resp = post_event_json(&mut h, "account.updated", account).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Verify DB was updated
    let (charges, payouts, onboarding): (bool, bool, bool) = sqlx::query_as(
        "SELECT stripe_charges_enabled, stripe_payouts_enabled, stripe_onboarding_complete FROM users WHERE id = $1",
    )
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .unwrap();

    assert!(charges, "charges_enabled should be true");
    assert!(payouts, "payouts_enabled should be true");
    assert!(onboarding, "onboarding_complete should be true");
}

#[tokio::test]
async fn webhook_purchase_completed() {
    let mut h = TestHarness::with_stripe().await;

    // Create buyer + seller
    let buyer_id = h.signup("buyer", "buyer@test.com", "password123").await;
    h.client.post_form("/logout", "").await;
    let seller_id = h.signup("seller", "seller@test.com", "password123").await;
    h.grant_creator(seller_id).await;
    h.client.post_form("/logout", "").await;
    h.login("seller", "password123").await;

    // Create project + item
    let resp = h
        .client
        .post_form("/api/projects", "slug=stripeproj&title=Stripe+Project")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();
    let resp = h
        .client
        .post_form(
            &format!("/api/projects/{project_id}/items"),
            "title=Paid+Track&price_cents=999&item_type=audio",
        )
        .await;
    let item: Value = resp.json();
    let item_id = item["id"].as_str().unwrap().to_string();

    // Insert a pending transaction via direct SQL
    let session_id = "cs_test_purchase_123";
    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, 999, 'pending', $4, 'Paid Track', 'seller')",
    )
    .bind(buyer_id)
    .bind(seller_id)
    .bind(&item_id)
    .bind(session_id)
    .execute(&h.db)
    .await
    .unwrap();

    // Build checkout session with valid IDs
    let mut meta = HashMap::new();
    meta.insert("buyer_id".to_string(), buyer_id.to_string());
    meta.insert("seller_id".to_string(), seller_id.to_string());
    meta.insert("item_id".to_string(), item_id.clone());
    let session = serde_json::json!({
        "id": session_id,
        "object": "checkout_session",
        "mode": "payment",
        "metadata": meta,
        "payment_intent": "pi_test_purchase_123",
    });

    let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Verify transaction was completed
    let status: String =
        sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_checkout_session_id = $1")
            .bind(session_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(status, "completed");

    // Verify sales_count was incremented
    let sales: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
        .bind(&item_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(sales, 1);
}

#[tokio::test]
async fn webhook_charge_refunded() {
    let mut h = TestHarness::with_stripe().await;

    // Create buyer + seller + item
    let buyer_id = h.signup("rbuyer", "rb@test.com", "password123").await;
    h.client.post_form("/logout", "").await;
    let seller_id = h.signup("rseller", "rs@test.com", "password123").await;
    h.grant_creator(seller_id).await;
    h.client.post_form("/logout", "").await;
    h.login("rseller", "password123").await;

    let resp = h
        .client
        .post_form("/api/projects", "slug=refundproj&title=Refund+Project")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();
    let resp = h
        .client
        .post_form(
            &format!("/api/projects/{project_id}/items"),
            "title=Refund+Track&price_cents=500&item_type=audio",
        )
        .await;
    let item: Value = resp.json();
    let item_id = item["id"].as_str().unwrap().to_string();

    // Set sales_count to 1 and insert a completed transaction
    let pi_id = "pi_test_refund_123";
    sqlx::query("UPDATE items SET sales_count = 1 WHERE id = $1::uuid")
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();
    sqlx::query(
        r"INSERT INTO transactions
           (buyer_id, seller_id, item_id, amount_cents, status,
            stripe_payment_intent_id, stripe_checkout_session_id,
            item_title, seller_username, completed_at)
           VALUES ($1, $2, $3::uuid, 500, 'completed', $4, 'cs_refund', 'Refund Track', 'rseller', NOW())",
    )
    .bind(buyer_id)
    .bind(seller_id)
    .bind(&item_id)
    .bind(pi_id)
    .execute(&h.db)
    .await
    .unwrap();

    // Build charge with valid id and payment_intent
    let charge = serde_json::json!({
        "id": "ch_test_refund",
        "object": "charge",
        "amount": 500,
        "amount_refunded": 500,
        "payment_intent": "pi_test_refund_123",
    });

    let resp = post_event_json(&mut h, "charge.refunded", charge).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Verify transaction was refunded
    let status: String =
        sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_payment_intent_id = $1")
            .bind(pi_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(status, "refunded");

    // Verify sales_count was decremented
    let sales: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
        .bind(&item_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(sales, 0);
}

#[tokio::test]
async fn webhook_subscription_deleted() {
    let mut h = TestHarness::with_stripe().await;

    // Create subscriber and creator with project + tier
    let sub_user_id = h.signup("subscriber", "sub@test.com", "password123").await;
    h.client.post_form("/logout", "").await;
    let creator_id = h.signup("tiercreator", "tc@test.com", "password123").await;
    h.grant_creator(creator_id).await;
    h.client.post_form("/logout", "").await;
    h.login("tiercreator", "password123").await;

    let resp = h
        .client
        .post_form("/api/projects", "slug=subproj&title=Sub+Project")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();

    // Create subscription tier via direct SQL
    let tier_id = uuid::Uuid::new_v4();
    sqlx::query(
        r"INSERT INTO subscription_tiers
           (id, project_id, name, price_cents)
           VALUES ($1, $2::uuid, 'Basic', 500)",
    )
    .bind(tier_id)
    .bind(&project_id)
    .execute(&h.db)
    .await
    .unwrap();

    // Create subscription via direct SQL
    let stripe_sub_id = "sub_test_delete_123";
    sqlx::query(
        r"INSERT INTO subscriptions
           (subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id, status)
           VALUES ($1, $2, $3::uuid, $4, 'cus_test', 'active')",
    )
    .bind(sub_user_id)
    .bind(tier_id)
    .bind(&project_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    let sub = serde_json::json!({
        "id": stripe_sub_id,
        "object": "subscription",
        "status": "canceled",
        "cancel_at_period_end": false,
        "items": {
            "object": "list",
            "data": [{
                "id": "si_test_del",
                "object": "subscription_item",
                "subscription": stripe_sub_id,
                "current_period_start": 1_700_000_000,
                "current_period_end": 1_702_592_000,
                "metadata": {},
            }],
        },
    });

    let resp = post_event_json(&mut h, "customer.subscription.deleted", sub).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Verify subscription was canceled
    let status: String =
        sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
            .bind(stripe_sub_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(status, "canceled");
}

// Shared fixture for subscription webhook tests

#[allow(
    clippy::struct_field_names,
    reason = "every field names a distinct entity id; the shared _id suffix is the meaningful convention, not noise"
)]
struct SubscriptionFixture {
    #[allow(dead_code)]
    creator_id: UserId,
    subscriber_id: UserId,
    project_id: String,
    tier_id: uuid::Uuid,
}

/// Creates a creator (with project + tier) and a subscriber user.
/// Leaves the harness logged out.
async fn setup_subscription_fixture(h: &mut TestHarness) -> SubscriptionFixture {
    let subscriber_id = h.signup("subuser", "subuser@test.com", "password123").await;
    h.client.post_form("/logout", "").await;
    let creator_id = h.signup("creator", "creator@test.com", "password123").await;
    h.grant_creator(creator_id).await;
    h.client.post_form("/logout", "").await;
    h.login("creator", "password123").await;

    let resp = h
        .client
        .post_form("/api/projects", "slug=subfix&title=Sub+Fixture")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();

    let tier_id = uuid::Uuid::new_v4();
    sqlx::query(
        r"INSERT INTO subscription_tiers (id, project_id, name, price_cents)
           VALUES ($1, $2::uuid, 'Pro', 1000)",
    )
    .bind(tier_id)
    .bind(&project_id)
    .execute(&h.db)
    .await
    .unwrap();

    h.client.post_form("/logout", "").await;

    SubscriptionFixture {
        creator_id,
        subscriber_id,
        project_id,
        tier_id,
    }
}

/// Insert an active subscription row into the DB. Returns the stripe subscription ID.
async fn insert_active_subscription(
    h: &TestHarness,
    fix: &SubscriptionFixture,
    stripe_sub_id: &str,
) {
    sqlx::query(
        r"INSERT INTO subscriptions
           (subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id, status)
           VALUES ($1, $2, $3::uuid, $4, 'cus_test_fixture', 'active')",
    )
    .bind(fix.subscriber_id)
    .bind(fix.tier_id)
    .bind(&fix.project_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();
}

/// Insert an active creator-tier subscription row and sync the denormalized
/// `users.creator_tier` column to match. Returns nothing; the caller owns the ids.
async fn insert_active_creator_sub(
    h: &TestHarness,
    user_id: UserId,
    stripe_sub_id: &str,
    tier: &str,
) {
    sqlx::query(
        r"INSERT INTO creator_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, tier, status)
           VALUES ($1, $2, 'cus_ct_fixture', $3, 'active')",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .bind(tier)
    .execute(&h.db)
    .await
    .unwrap();
    sqlx::query("UPDATE users SET creator_tier = $2 WHERE id = $1")
        .bind(user_id)
        .bind(tier)
        .execute(&h.db)
        .await
        .unwrap();
}

// New tests

#[tokio::test]
async fn webhook_subscription_checkout_completed() {
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;

    let stripe_sub_id = "sub_test_checkout_001";
    let stripe_customer_id = "cus_test_checkout_001";

    // Build checkout session with subscription metadata
    let mut meta = HashMap::new();
    meta.insert("checkout_type".to_string(), "subscription".to_string());
    meta.insert("subscriber_id".to_string(), fix.subscriber_id.to_string());
    meta.insert("project_id".to_string(), fix.project_id.clone());
    meta.insert("tier_id".to_string(), fix.tier_id.to_string());
    let session = serde_json::json!({
        "id": "cs_test_sub_checkout_001",
        "object": "checkout_session",
        "mode": "subscription",
        "metadata": meta,
        "subscription": stripe_sub_id,
        "customer": stripe_customer_id,
    });

    let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Verify subscription row was created
    let (status, sub_stripe_id, sub_customer_id): (String, String, String) = sqlx::query_as(
        "SELECT status, stripe_subscription_id, stripe_customer_id FROM subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();

    assert_eq!(status, "active");
    assert_eq!(sub_stripe_id, stripe_sub_id);
    assert_eq!(sub_customer_id, stripe_customer_id);
}

#[tokio::test]
async fn webhook_subscription_checkout_completed_idempotent() {
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;

    let stripe_sub_id = "sub_test_checkout_idem";
    let stripe_customer_id = "cus_test_checkout_idem";

    let build_session = || {
        let mut meta = HashMap::new();
        meta.insert("checkout_type".to_string(), "subscription".to_string());
        meta.insert("subscriber_id".to_string(), fix.subscriber_id.to_string());
        meta.insert("project_id".to_string(), fix.project_id.clone());
        meta.insert("tier_id".to_string(), fix.tier_id.to_string());
        serde_json::json!({
            "id": "cs_test_sub_idem",
            "object": "checkout_session",
            "mode": "subscription",
            "metadata": meta,
            "subscription": stripe_sub_id,
            "customer": stripe_customer_id,
        })
    };

    // First event
    let resp = post_event_json(&mut h, "checkout.session.completed", build_session()).await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "First webhook failed: {}",
        resp.text
    );

    // Second event (duplicate), use a different event ID
    let resp = post_event_json_with_id(
        &mut h,
        "evt_test_001",
        "checkout.session.completed",
        build_session(),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "Duplicate webhook should succeed: {}",
        resp.text
    );

    // Verify still only one subscription row
    let count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM subscriptions WHERE stripe_subscription_id = $1")
            .bind(stripe_sub_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(count, 1, "Should have exactly one subscription row");
}

#[tokio::test]
async fn webhook_subscription_updated() {
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;

    let stripe_sub_id = "sub_test_updated_001";
    insert_active_subscription(&h, &fix, stripe_sub_id).await;

    let sub = serde_json::json!({
        "id": stripe_sub_id,
        "object": "subscription",
        "status": "past_due",
        "cancel_at_period_end": false,
        "items": {
            "object": "list",
            "data": [{
                "id": "si_test_upd",
                "object": "subscription_item",
                "subscription": stripe_sub_id,
                "current_period_start": 1_702_592_000,
                "current_period_end": 1_705_184_000,
                "metadata": {},
            }],
        },
    });

    let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Verify status changed
    let status: String =
        sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
            .bind(stripe_sub_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(status, "past_due");

    // Verify period was updated
    let (period_start, period_end): (Option<chrono::DateTime<chrono::Utc>>, Option<chrono::DateTime<chrono::Utc>>) = sqlx::query_as(
        "SELECT current_period_start, current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert!(period_start.is_some(), "period_start should be set");
    assert!(period_end.is_some(), "period_end should be set");
    assert_eq!(period_start.unwrap().timestamp(), 1_702_592_000);
    assert_eq!(period_end.unwrap().timestamp(), 1_705_184_000);
}

/// CHRONIC C regression: a thin `customer.subscription.updated` carrying no
/// period must NOT overwrite an active row's period with an epoch (1970) value.
/// The access gate requires `current_period_end > NOW()`, so an epoch write
/// would cut off a paying subscriber until the next invoice repaired the row.
/// The sealed writer drops a missing/zero period (COALESCE keeps the existing).
#[tokio::test]
async fn webhook_subscription_updated_thin_event_preserves_period() {
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;

    let stripe_sub_id = "sub_test_thin_period";
    let original_end: i64 = 1_893_456_000; // 2030-01-01, comfortably in the future
    sqlx::query(
        r"INSERT INTO subscriptions
           (subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id,
            status, current_period_start, current_period_end)
           VALUES ($1, $2, $3::uuid, $4, 'cus_test_fixture', 'active',
                   to_timestamp($5), to_timestamp($6))",
    )
    .bind(fix.subscriber_id)
    .bind(fix.tier_id)
    .bind(&fix.project_id)
    .bind(stripe_sub_id)
    .bind(1_700_000_000_i64)
    .bind(original_end)
    .execute(&h.db)
    .await
    .unwrap();

    // Thin update: a status the row already has, and NO items/period at all.
    let sub = serde_json::json!({
        "id": stripe_sub_id,
        "object": "subscription",
        "status": "active",
        "cancel_at_period_end": false,
        "items": { "object": "list", "data": [] },
    });
    let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    let period_end: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        period_end.expect("period_end must remain set").timestamp(),
        original_end,
        "thin webhook must leave the period untouched, not stamp 1970"
    );
}

#[tokio::test]
async fn webhook_invoice_payment_succeeded() {
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;

    let stripe_sub_id = "sub_test_inv_success";
    insert_active_subscription(&h, &fix, stripe_sub_id).await;

    let invoice = serde_json::json!({
        "id": "in_test_success_001",
        "object": "invoice",
        "subscription": stripe_sub_id,
        "period_start": 1_702_592_000,
        "period_end": 1_705_184_000,
        "billing_reason": "subscription_cycle",
        "livemode": false,
    });

    let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Verify subscription period was updated
    let (period_start, period_end): (Option<chrono::DateTime<chrono::Utc>>, Option<chrono::DateTime<chrono::Utc>>) = sqlx::query_as(
        "SELECT current_period_start, current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert!(period_start.is_some(), "period_start should be set");
    assert!(period_end.is_some(), "period_end should be set");
    assert_eq!(period_start.unwrap().timestamp(), 1_702_592_000);
    assert_eq!(period_end.unwrap().timestamp(), 1_705_184_000);
}

#[tokio::test]
async fn webhook_invoice_payment_failed() {
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;

    let stripe_sub_id = "sub_test_inv_failed";
    insert_active_subscription(&h, &fix, stripe_sub_id).await;

    let invoice = serde_json::json!({
        "id": "in_test_failed_001",
        "object": "invoice",
        "subscription": stripe_sub_id,
        "period_start": 1_700_000_000,
        "period_end": 1_702_592_000,
        "livemode": false,
    });

    let resp = post_event_json(&mut h, "invoice.payment_failed", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Verify status changed to past_due
    let status: String =
        sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
            .bind(stripe_sub_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(status, "past_due");
}

#[tokio::test]
async fn webhook_account_updated_partial() {
    let mut h = TestHarness::with_stripe().await;

    let user_id = h
        .signup("partialcreator", "pc@test.com", "password123")
        .await;
    let acct_id = "acct_test_partial_123";
    sqlx::query("UPDATE users SET stripe_account_id = $1 WHERE id = $2")
        .bind(acct_id)
        .bind(user_id)
        .execute(&h.db)
        .await
        .unwrap();

    // Only details_submitted is true; charges and payouts still false
    let account = serde_json::json!({
        "id": acct_id,
        "object": "account",
        "charges_enabled": false,
        "payouts_enabled": false,
        "details_submitted": true,
    });

    let resp = post_event_json(&mut h, "account.updated", account).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    let (charges, payouts, onboarding): (bool, bool, bool) = sqlx::query_as(
        "SELECT stripe_charges_enabled, stripe_payouts_enabled, stripe_onboarding_complete FROM users WHERE id = $1",
    )
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .unwrap();

    assert!(!charges, "charges_enabled should be false");
    assert!(!payouts, "payouts_enabled should be false");
    assert!(onboarding, "onboarding_complete should be true");
}

#[tokio::test]
async fn webhook_account_updated_unknown_account() {
    let mut h = TestHarness::with_stripe().await;

    // No user has this stripe_account_id
    let account = serde_json::json!({
        "id": "acct_nonexistent",
        "object": "account",
        "charges_enabled": true,
        "payouts_enabled": true,
        "details_submitted": true,
    });

    let resp = post_event_json(&mut h, "account.updated", account).await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "Unknown account should still return 200: {}",
        resp.text
    );

    // Verify no users were affected
    let count: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM users WHERE stripe_account_id = 'acct_nonexistent'",
    )
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(count, 0);
}

#[tokio::test]
async fn webhook_purchase_completed_idempotent() {
    let mut h = TestHarness::with_stripe().await;

    // Create buyer + seller
    let buyer_id = h.signup("idembuyer", "ib@test.com", "password123").await;
    h.client.post_form("/logout", "").await;
    let seller_id = h.signup("idemseller", "is@test.com", "password123").await;
    h.grant_creator(seller_id).await;
    h.client.post_form("/logout", "").await;
    h.login("idemseller", "password123").await;

    // Create project + item
    let resp = h
        .client
        .post_form("/api/projects", "slug=idemproj&title=Idem+Project")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();
    let resp = h
        .client
        .post_form(
            &format!("/api/projects/{project_id}/items"),
            "title=Idem+Track&price_cents=500&item_type=audio",
        )
        .await;
    let item: Value = resp.json();
    let item_id = item["id"].as_str().unwrap().to_string();

    // Insert a pending transaction
    let session_id = "cs_test_idem_001";
    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, 500, 'pending', $4, 'Idem Track', 'idemseller')",
    )
    .bind(buyer_id)
    .bind(seller_id)
    .bind(&item_id)
    .bind(session_id)
    .execute(&h.db)
    .await
    .unwrap();

    let build_session = || {
        let mut meta = HashMap::new();
        meta.insert("buyer_id".to_string(), buyer_id.to_string());
        meta.insert("seller_id".to_string(), seller_id.to_string());
        meta.insert("item_id".to_string(), item_id.clone());
        serde_json::json!({
            "id": session_id,
            "object": "checkout_session",
            "mode": "payment",
            "metadata": meta,
            "payment_intent": "pi_test_idem_001",
        })
    };

    // First event
    let resp = post_event_json(&mut h, "checkout.session.completed", build_session()).await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "First webhook failed: {}",
        resp.text
    );

    // Second event (duplicate)
    let resp = post_event_json_with_id(
        &mut h,
        "evt_test_002",
        "checkout.session.completed",
        build_session(),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "Duplicate webhook should succeed: {}",
        resp.text
    );

    // Verify still one completed transaction
    let count: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM transactions WHERE stripe_checkout_session_id = $1 AND status = 'completed'",
    )
    .bind(session_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(count, 1, "Should have exactly one completed transaction");

    // Verify sales_count is still 1 (not incremented twice)
    let sales: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
        .bind(&item_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(sales, 1, "sales_count should be 1, not 2");
}

/// Dedup primitive regression: the "processed" row is written only AFTER a
/// handler succeeds, so the read/mark pair must compose correctly, an unseen
/// event reads false, a marked event reads true, and the mark is idempotent
/// (a redelivery or concurrent duplicate must not error). This pins the layer
/// the live handlers rely on (the `_idempotent` tests above use distinct event
/// IDs, so they only exercise the handler's SQL idempotency, not this one).
#[tokio::test]
async fn webhook_event_dedup_layering() {
    use makenotwork::db::webhook_events;

    let h = TestHarness::new().await;
    let id = "evt_dedup_layer_001";

    // Never seen → not processed.
    assert!(
        !webhook_events::is_event_processed(&h.db, id).await.unwrap(),
        "an unseen event must read as not-processed"
    );
    // Mark after a (hypothetical) successful handler → now processed, so a
    // redelivery short-circuits.
    webhook_events::mark_event_processed(&h.db, id)
        .await
        .unwrap();
    assert!(
        webhook_events::is_event_processed(&h.db, id).await.unwrap(),
        "a marked event must read as processed"
    );
    // Marking again (redelivery / concurrent duplicate) is idempotent, not an error.
    webhook_events::mark_event_processed(&h.db, id)
        .await
        .unwrap();
    assert!(
        webhook_events::is_event_processed(&h.db, id).await.unwrap(),
        "re-marking must stay processed and must not error"
    );
}

// v2 thin event tests

/// Helper to POST a raw JSON payload to /stripe/webhook/v2 with a given signature.
async fn post_v2_raw(
    h: &mut TestHarness,
    payload: &str,
    signature: &str,
) -> crate::harness::client::TestResponse {
    h.client
        .request_with_headers(
            "POST",
            "/stripe/webhook/v2",
            Some(payload),
            &[
                ("stripe-signature", signature),
                ("content-type", "application/json"),
            ],
        )
        .await
}

#[tokio::test]
async fn webhook_v2_invalid_signature() {
    let mut h = TestHarness::with_stripe().await;

    let payload = r#"{"id":"evt_v2_bad","type":"v2.core.account.updated","related_object":{"id":"acct_123","type":"account"}}"#;
    let bad_sig = "t=0,v1=00000000000000000000000000000000";

    let resp = post_v2_raw(&mut h, payload, bad_sig).await;
    assert_eq!(
        resp.status.as_u16(),
        400,
        "Expected 400 for bad v2 signature, got: {}",
        resp.status
    );
}

#[tokio::test]
async fn webhook_v2_account_event_accepted() {
    let mut h = TestHarness::with_mocks().await;

    // Uses mock Stripe so fetch_account returns success
    let payload = serde_json::json!({
        "id": "evt_v2_acct_001",
        "type": "v2.core.account.updated",
        "related_object": {
            "id": "acct_test_v2_123",
            "type": "account"
        }
    })
    .to_string();

    let signature = sign_webhook_payload(&payload, TEST_WEBHOOK_SECRET_V2);
    let resp = post_v2_raw(&mut h, &payload, &signature).await;

    assert_eq!(
        resp.status.as_u16(),
        200,
        "v2 account event should return 200 even if API fetch fails: {}",
        resp.text
    );
}

#[tokio::test]
async fn webhook_v2_unknown_event_type_returns_200() {
    let mut h = TestHarness::with_stripe().await;

    let payload = serde_json::json!({
        "id": "evt_v2_unknown_001",
        "type": "v2.billing.meter.no_meter_found",
        "related_object": {
            "id": "mtr_123",
            "type": "billing.meter"
        }
    })
    .to_string();

    let signature = sign_webhook_payload(&payload, TEST_WEBHOOK_SECRET_V2);
    let resp = post_v2_raw(&mut h, &payload, &signature).await;

    assert_eq!(
        resp.status.as_u16(),
        200,
        "Unknown v2 event type should return 200: {}",
        resp.text
    );
}

// Fan+ subscription webhook lifecycle
//
// Pins the earlier cascade branches in handle_subscription_updated /
// handle_subscription_deleted / handle_invoice_payment_succeeded /
// handle_invoice_payment_failed, previously only the generic creator-sub
// fallback path was exercised.

fn make_subscription(stripe_sub_id: &str, status: &str) -> serde_json::Value {
    serde_json::json!({
        "id": stripe_sub_id,
        "object": "subscription",
        "status": status,
        "cancel_at_period_end": false,
        "items": {
            "object": "list",
            "data": [{
                "id": "si_fan_plus_test",
                "object": "subscription_item",
                "subscription": stripe_sub_id,
                "current_period_start": 1_700_000_000_i64,
                "current_period_end": 1_702_592_000_i64,
                "metadata": {},
            }],
        },
    })
}

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

    let stripe_sub_id = "sub_fp_update_1";
    sqlx::query(
        r"INSERT INTO fan_plus_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
           VALUES ($1, $2, 'cus_fp_update', 'active', NOW() + interval '30 days')",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    let sub = make_subscription(stripe_sub_id, "past_due");
    let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // fan_plus row reflects the new status, pins that the fan_plus branch
    // ran and reached `update_fan_plus_status`, not the generic fallback.
    let status: String = sqlx::query_scalar(
        "SELECT status::text FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(status, "past_due");
}

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

    let stripe_sub_id = "sub_fp_delete_1";
    sqlx::query(
        r"INSERT INTO fan_plus_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
           VALUES ($1, $2, 'cus_fp_delete', 'active', NOW() + interval '30 days')",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    let sub = make_subscription(stripe_sub_id, "canceled");
    let resp = post_event_json(&mut h, "customer.subscription.deleted", sub).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Pins that `cancel_fan_plus` ran. We don't pin the exact column the
    // cancellation writes to (status vs separate canceled_at); just that
    // a downstream lookup classifies this user as NOT active.
    let active: bool = sqlx::query_scalar(
        "SELECT EXISTS(SELECT 1 FROM fan_plus_subscriptions \
         WHERE user_id = $1 AND status = 'active' AND canceled_at IS NULL)",
    )
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert!(
        !active,
        "Fan+ subscription must not be active after cancellation"
    );
}

fn make_invoice(stripe_sub_id: &str, billing_reason: &str) -> serde_json::Value {
    serde_json::json!({
        "id": "in_test_fp",
        "object": "invoice",
        "subscription": stripe_sub_id,
        "billing_reason": billing_reason,
        "period_start": 1_700_000_000_i64,
        "period_end": 1_702_592_000_i64,
        "currency": "usd",
        "livemode": false,
    })
}

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

    let stripe_sub_id = "sub_fp_invoice_1";
    sqlx::query(
        r"INSERT INTO fan_plus_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
           VALUES ($1, $2, 'cus_fp_invoice', 'active', NOW())",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    // billing_reason != "subscription_cycle" → not a renewal, just updates period.
    let invoice = make_invoice(stripe_sub_id, "subscription_create");
    let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Period_end should now match the invoice's period_end (2024-12-14T22:13:20Z = 1_702_592_000).
    let period_end: chrono::DateTime<chrono::Utc> = sqlx::query_scalar(
        "SELECT current_period_end FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(period_end.timestamp(), 1_702_592_000);
}

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

    let stripe_sub_id = "sub_fp_fail_1";
    sqlx::query(
        r"INSERT INTO fan_plus_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
           VALUES ($1, $2, 'cus_fp_fail', 'active', NOW() + interval '30 days')",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
    let resp = post_event_json(&mut h, "invoice.payment_failed", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    let status: String = sqlx::query_scalar(
        "SELECT status::text FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        status, "past_due",
        "Fan+ must be flipped to past_due on payment failure"
    );
}

#[tokio::test]
async fn webhook_subscription_updated_unknown_id_returns_200() {
    // Pins the fall-through: an event for a stripe_sub_id that has no fan_plus,
    // creator_tier, or app_sync row should still return 200 (no-op).
    let mut h = TestHarness::with_stripe().await;
    let sub = make_subscription("sub_does_not_exist", "active");
    let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
    // Generic path runs `update_subscription_status` which finds nothing,
    // current behavior is to return 200 (idempotent / unknown-sub tolerance).
    assert_eq!(
        resp.status.as_u16(),
        200,
        "Unknown sub_id should not error: {}",
        resp.text
    );
}

// Creator-tier subscription webhook lifecycle (test-fuzz Phase 2.1)
//
// The platform's own revenue ($16-60/mo). Before this block, the creator_tier
// branch in every billing/subscription handler was untested, only fan_plus and
// the generic creator-sub fallback were exercised. These pin that the
// creator_tier branch runs AND keeps the denormalized `users.creator_tier`
// column in sync (sync_user_creator_tier sets it to the active sub's tier, or
// NULL when no active sub remains).

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

    let stripe_sub_id = "sub_ct_checkout_1";
    let mut meta = HashMap::new();
    meta.insert("checkout_type".to_string(), "creator_tier".to_string());
    meta.insert("user_id".to_string(), user_id.to_string());
    meta.insert("tier".to_string(), "small_files".to_string());
    let session = serde_json::json!({
        "id": "cs_ct_checkout_1",
        "object": "checkout_session",
        "mode": "subscription",
        "metadata": meta,
        "subscription": stripe_sub_id,
        "customer": "cus_ct_checkout_1",
    });

    let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Subscription row created with the right tier + active status.
    let (status, tier): (String, String) = sqlx::query_as(
        "SELECT status, tier FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(status, "active");
    assert_eq!(tier, "small_files");

    // Denormalized column synced on the user.
    let user_tier: Option<String> =
        sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
            .bind(user_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        user_tier.as_deref(),
        Some("small_files"),
        "users.creator_tier must be synced"
    );
}

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

    let stripe_sub_id = "sub_ct_invoice_1";
    insert_active_creator_sub(&h, user_id, stripe_sub_id, "big_files").await;

    let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
    let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    let (period_start, period_end): (
        Option<chrono::DateTime<chrono::Utc>>,
        Option<chrono::DateTime<chrono::Utc>>,
    ) = sqlx::query_as(
        "SELECT current_period_start, current_period_end FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(period_start.unwrap().timestamp(), 1_700_000_000);
    assert_eq!(period_end.unwrap().timestamp(), 1_702_592_000);

    // Tier remains live after a successful renewal.
    let user_tier: Option<String> =
        sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
            .bind(user_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(user_tier.as_deref(), Some("big_files"));
}

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

    let stripe_sub_id = "sub_ct_fail_1";
    insert_active_creator_sub(&h, user_id, stripe_sub_id, "everything").await;

    let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
    let resp = post_event_json(&mut h, "invoice.payment_failed", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    let status: String = sqlx::query_scalar(
        "SELECT status FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(status, "past_due");

    // A past_due creator sub is no longer active, so the denormalized tier is cleared,
    // this is the gate that downstream enforcement reads to start the grace clock.
    let user_tier: Option<String> =
        sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
            .bind(user_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        user_tier, None,
        "users.creator_tier must clear when the sub goes past_due"
    );
}

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

    let stripe_sub_id = "sub_ct_upd_1";
    insert_active_creator_sub(&h, user_id, stripe_sub_id, "small_files").await;

    let sub = make_subscription(stripe_sub_id, "past_due");
    let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    let (status, period_end): (String, Option<chrono::DateTime<chrono::Utc>>) = sqlx::query_as(
        "SELECT status, current_period_end FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(status, "past_due");
    assert_eq!(
        period_end.unwrap().timestamp(),
        1_702_592_000,
        "period must be updated from the event"
    );

    let user_tier: Option<String> =
        sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
            .bind(user_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        user_tier, None,
        "tier sync must run on the creator_tier update branch"
    );
}

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

    let stripe_sub_id = "sub_ct_del_1";
    insert_active_creator_sub(&h, user_id, stripe_sub_id, "everything").await;

    let sub = make_subscription(stripe_sub_id, "canceled");
    let resp = post_event_json(&mut h, "customer.subscription.deleted", sub).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    let status: String = sqlx::query_scalar(
        "SELECT status FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(status, "canceled");

    let user_tier: Option<String> =
        sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
            .bind(user_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        user_tier, None,
        "users.creator_tier must clear on cancellation"
    );
}

// Fan+ renewal credit code (test-fuzz Phase 2.1)
//
// The earlier fan+ invoice test deliberately uses billing_reason=subscription_create
// to avoid the renewal branch. THIS pins the renewal branch: a subscription_cycle
// invoice generates the $5 single-use platform promo code that funds the Fan+
// monthly credit. That code is a real DB write nothing previously exercised.

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

    let stripe_sub_id = "sub_fp_credit_1";
    sqlx::query(
        r"INSERT INTO fan_plus_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
           VALUES ($1, $2, 'cus_fp_credit', 'active', NOW())",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    // subscription_cycle == a renewal, which triggers the credit-code path.
    let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
    let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Exactly one $5 single-use fixed-discount promo code now exists for this user.
    let (count, discount_type, discount_value, max_uses): (
        i64,
        Option<String>,
        Option<i32>,
        Option<i32>,
    ) = sqlx::query_as(
        "SELECT COUNT(*), MIN(discount_type), MIN(discount_value), MIN(max_uses) \
             FROM promo_codes WHERE creator_id = $1 AND code_purpose = 'discount'",
    )
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(count, 1, "renewal must mint exactly one Fan+ credit code");
    assert_eq!(discount_type.as_deref(), Some("fixed"));
    assert_eq!(discount_value, Some(500), "$5 credit");
    assert_eq!(max_uses, Some(1), "single use");
}

// Generic creator-sub renewal email gating (test-fuzz Phase 2.1)
//
// The generic invoice.payment_succeeded test only asserts the period write. The
// renewal-email side effect, sent on a renewal, suppressed on the first invoice
//, was untested. These use with_mocks() to capture the email.

#[tokio::test]
async fn webhook_invoice_payment_succeeded_renewal_sends_email() {
    let mut h = TestHarness::with_mocks().await;
    let fix = setup_subscription_fixture(&mut h).await;
    let stripe_sub_id = "sub_renewal_email_1";
    insert_active_subscription(&h, &fix, stripe_sub_id).await;

    let invoice = make_invoice(stripe_sub_id, "subscription_cycle");
    let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Fire-and-forget email task, give it a beat to land.
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    let mock_email = h.mock_email.as_ref().unwrap();
    let mails = mock_email.sent_to("subuser@test.com");
    assert!(
        mails.iter().any(|e| e.subject.contains("renewed")),
        "renewal must send a 'renewed' email, got: {:?}",
        mails.iter().map(|e| &e.subject).collect::<Vec<_>>()
    );
}

#[tokio::test]
async fn webhook_invoice_payment_succeeded_first_invoice_sends_no_renewal_email() {
    let mut h = TestHarness::with_mocks().await;
    let fix = setup_subscription_fixture(&mut h).await;
    let stripe_sub_id = "sub_first_invoice_1";
    insert_active_subscription(&h, &fix, stripe_sub_id).await;

    // subscription_create is the FIRST invoice, not a renewal.
    let invoice = make_invoice(stripe_sub_id, "subscription_create");
    let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    let mock_email = h.mock_email.as_ref().unwrap();
    let mails = mock_email.sent_to("subuser@test.com");
    assert!(
        !mails.iter().any(|e| e.subject.contains("renewed")),
        "first invoice must NOT send a renewal email, got: {:?}",
        mails.iter().map(|e| &e.subject).collect::<Vec<_>>()
    );
}

// charge.refunded edge paths (test-fuzz Phase 2.1)
//
// Only the full-refund-with-matching-transaction path was tested. The partial
// refund (preserve access) and no-match (queue a pending refund) branches,
// both money-critical, were not.

#[tokio::test]
async fn webhook_charge_refunded_partial_preserves_access() {
    let mut h = TestHarness::with_stripe().await;

    let buyer_id = h.signup("prbuyer", "prb@test.com", "password123").await;
    h.client.post_form("/logout", "").await;
    let seller_id = h.signup("prseller", "prs@test.com", "password123").await;
    h.grant_creator(seller_id).await;
    h.client.post_form("/logout", "").await;
    h.login("prseller", "password123").await;

    let resp = h
        .client
        .post_form("/api/projects", "slug=partialproj&title=Partial+Project")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();
    let resp = h
        .client
        .post_form(
            &format!("/api/projects/{project_id}/items"),
            "title=Partial+Track&price_cents=1000&item_type=audio",
        )
        .await;
    let item: Value = resp.json();
    let item_id = item["id"].as_str().unwrap().to_string();

    let pi_id = "pi_partial_refund_1";
    sqlx::query("UPDATE items SET sales_count = 1 WHERE id = $1::uuid")
        .bind(&item_id)
        .execute(&h.db)
        .await
        .unwrap();
    sqlx::query(
        r"INSERT INTO transactions
           (buyer_id, seller_id, item_id, amount_cents, status,
            stripe_payment_intent_id, stripe_checkout_session_id,
            item_title, seller_username, completed_at)
           VALUES ($1, $2, $3::uuid, 1000, 'completed', $4, 'cs_partial', 'Partial Track', 'prseller', NOW())",
    )
    .bind(buyer_id)
    .bind(seller_id)
    .bind(&item_id)
    .bind(pi_id)
    .execute(&h.db)
    .await
    .unwrap();

    // Refund only 400 of 1000 cents, a partial refund.
    let charge = serde_json::json!({
        "id": "ch_partial_1",
        "object": "charge",
        "amount": 1000,
        "amount_refunded": 400,
        "payment_intent": pi_id,
    });
    let resp = post_event_json(&mut h, "charge.refunded", charge).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Access preserved: the transaction is NOT marked refunded and the sale stands.
    let status: String =
        sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_payment_intent_id = $1")
            .bind(pi_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(status, "completed", "partial refund must NOT revoke access");

    let sales: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
        .bind(&item_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(sales, 1, "partial refund must NOT decrement the sale");
}

#[tokio::test]
async fn webhook_charge_refunded_no_transaction_queues_pending() {
    let mut h = TestHarness::with_stripe().await;

    // A full refund whose payment_intent matches no transaction or tip,
    // the payment webhook likely hasn't arrived yet. Must be queued, not dropped.
    let pi_id = "pi_orphan_refund_1";
    let charge = serde_json::json!({
        "id": "ch_orphan_1",
        "object": "charge",
        "amount": 1500,
        "amount_refunded": 1500,
        "payment_intent": pi_id,
    });
    let resp = post_event_json(&mut h, "charge.refunded", charge).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    let (amount, amount_refunded): (i64, i64) = sqlx::query_as(
        "SELECT amount, amount_refunded FROM pending_refunds WHERE payment_intent_id = $1",
    )
    .bind(pi_id)
    .fetch_one(&h.db)
    .await
    .expect("an unmatched full refund must be queued as a pending refund");
    assert_eq!(amount, 1500);
    assert_eq!(amount_refunded, 1500);
}

#[tokio::test]
async fn webhook_refund_created_line_scoped_does_not_reverse_cart() {
    // Run #2 Payments SERIOUS: a cart puts every line of an order under ONE
    // payment_intent. A self-service refund of one line tags the Stripe refund
    // with mnw_transaction_id; the refund.created webhook must mark/revoke ONLY
    // that transaction and leave the order's other lines untouched.
    let mut h = TestHarness::with_stripe().await;

    let buyer_id = h.signup("clbuyer", "clb@test.com", "password123").await;
    h.client.post_form("/logout", "").await;
    let seller_id = h.signup("clseller", "cls@test.com", "password123").await;
    h.grant_creator(seller_id).await;
    h.client.post_form("/logout", "").await;
    h.login("clseller", "password123").await;

    let resp = h
        .client
        .post_form("/api/projects", "slug=cartproj&title=Cart+Project")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();

    // Two items, both sold to the same buyer in one cart (shared payment_intent).
    let mut item_ids = Vec::new();
    for (n, title) in [("Line+One", "a"), ("Line+Two", "b")] {
        let _ = title;
        let resp = h
            .client
            .post_form(
                &format!("/api/projects/{project_id}/items"),
                &format!("title={n}&price_cents=500&item_type=audio"),
            )
            .await;
        let item: Value = resp.json();
        item_ids.push(item["id"].as_str().unwrap().to_string());
    }

    let pi_id = "pi_cart_line_refund";
    let mut tx_ids = Vec::new();
    for item_id in &item_ids {
        sqlx::query("UPDATE items SET sales_count = 1 WHERE id = $1::uuid")
            .bind(item_id)
            .execute(&h.db)
            .await
            .unwrap();
        let tx_id: uuid::Uuid = sqlx::query_scalar(
            r"INSERT INTO transactions
               (buyer_id, seller_id, item_id, amount_cents, status,
                stripe_payment_intent_id, stripe_checkout_session_id,
                item_title, seller_username, completed_at)
               VALUES ($1, $2, $3::uuid, 500, 'completed', $4, 'cs_cart', 'Line', 'clseller', NOW())
               RETURNING id",
        )
        .bind(buyer_id)
        .bind(seller_id)
        .bind(item_id)
        .bind(pi_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
        tx_ids.push(tx_id);
    }

    // Refund ONLY the first line via refund.created carrying its transaction id.
    let refund = serde_json::json!({
        "id": "re_cart_line_1",
        "object": "refund",
        "amount": 500,
        "status": "succeeded",
        "payment_intent": pi_id,
        "metadata": { "mnw_transaction_id": tx_ids[0].to_string() },
    });
    let resp = post_event_json(&mut h, "refund.created", refund).await;
    assert_eq!(resp.status.as_u16(), 200, "Webhook failed: {}", resp.text);

    // Line one is refunded; line two is untouched (the bug would refund both).
    let status_one: String = sqlx::query_scalar("SELECT status FROM transactions WHERE id = $1")
        .bind(tx_ids[0])
        .fetch_one(&h.db)
        .await
        .unwrap();
    let status_two: String = sqlx::query_scalar("SELECT status FROM transactions WHERE id = $1")
        .bind(tx_ids[1])
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(
        status_one, "refunded",
        "refunded line must be marked refunded"
    );
    assert_eq!(
        status_two, "completed",
        "sibling cart line must NOT be reversed"
    );

    // Sales count: only the refunded line's item decremented.
    let sales_one: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
        .bind(&item_ids[0])
        .fetch_one(&h.db)
        .await
        .unwrap();
    let sales_two: i32 = sqlx::query_scalar("SELECT sales_count FROM items WHERE id = $1::uuid")
        .bind(&item_ids[1])
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(sales_one, 0, "refunded line's sale decremented");
    assert_eq!(sales_two, 1, "sibling line's sale stands");
}

// Subscription / invoice edge cases (test-fuzz Phase 2.1)

#[tokio::test]
async fn webhook_subscription_updated_unknown_status_is_noop() {
    // Stripe periodically adds statuses (e.g. "paused"). The handler must treat
    // an unknown status as a no-op, returning Err would pin Stripe in an
    // infinite retry storm. The existing subscription must keep its prior status.
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;
    let stripe_sub_id = "sub_unknown_status_1";
    insert_active_subscription(&h, &fix, stripe_sub_id).await;

    let sub = make_subscription(stripe_sub_id, "paused"); // not a known SubscriptionStatus
    let resp = post_event_json(&mut h, "customer.subscription.updated", sub).await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "unknown status must not error: {}",
        resp.text
    );

    let status: String =
        sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
            .bind(stripe_sub_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        status, "active",
        "unknown status must be a no-op, leaving status untouched"
    );
}

#[tokio::test]
async fn webhook_invoice_without_subscription_is_noop() {
    // A non-subscription invoice (no subscription id anywhere) must short-circuit
    // to Ok without touching any subscription table.
    let mut h = TestHarness::with_stripe().await;
    let invoice = serde_json::json!({
        "id": "in_no_sub_1",
        "object": "invoice",
        "period_start": 1_700_000_000_i64,
        "period_end": 1_702_592_000_i64,
        "billing_reason": "manual",
        "livemode": false,
    });
    let resp = post_event_json(&mut h, "invoice.payment_succeeded", invoice).await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "invoice without subscription must be a 200 no-op: {}",
        resp.text
    );
}

// Adversarial: malformed fulfillment metadata (test-fuzz Phase 2.1)
//
// A checkout.session.completed whose metadata is missing the required buyer_id
// makes the handler error. The dispatcher must queue it for retry (a row in
// webhook_events) and still 200 Stripe, never drop it, never 500.

#[tokio::test]
async fn webhook_purchase_missing_buyer_id_metadata_is_queued() {
    let mut h = TestHarness::with_stripe().await;
    let seller_id = h.signup("mmseller", "mms@test.com", "password123").await;

    // No checkout_type → routed to the purchase handler, whose
    // CheckoutMetadata::from_metadata requires buyer_id and errors without it.
    let mut meta = HashMap::new();
    meta.insert("seller_id".to_string(), seller_id.to_string());
    let session = serde_json::json!({
        "id": "cs_missing_buyer_1",
        "object": "checkout_session",
        "mode": "payment",
        "metadata": meta,
        "payment_intent": "pi_missing_buyer_1",
    });

    let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "malformed event must still 200 after queueing: {}",
        resp.text
    );

    // The failed handler queued the event for retry rather than dropping it.
    let queued: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM webhook_events WHERE source = 'stripe' AND event_type = 'checkout.session.completed'",
    )
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        queued, 1,
        "a handler error must enqueue exactly one retry row"
    );
}

// Canceled is terminal: an out-of-order update must not revive it (Run #11 fix)

#[tokio::test]
async fn webhook_subscription_updated_cannot_revive_canceled_sub() {
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;
    let stripe_sub_id = "sub_revival_guard_1";
    insert_active_subscription(&h, &fix, stripe_sub_id).await;

    // Cancel it (terminal state).
    let resp = post_event_json_with_id(
        &mut h,
        "evt_revive_del",
        "customer.subscription.deleted",
        make_subscription(stripe_sub_id, "canceled"),
    )
    .await;
    assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);
    let status: String =
        sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
            .bind(stripe_sub_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(status, "canceled");

    // An out-of-order `updated`(active) arrives AFTER the cancellation, distinct
    // event id, so the dedup layer does NOT short-circuit it. The DB guard must
    // refuse to revive the canceled subscription.
    let resp = post_event_json_with_id(
        &mut h,
        "evt_revive_upd",
        "customer.subscription.updated",
        make_subscription(stripe_sub_id, "active"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "update must not error: {}",
        resp.text
    );
    let status: String =
        sqlx::query_scalar("SELECT status FROM subscriptions WHERE stripe_subscription_id = $1")
            .bind(stripe_sub_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        status, "canceled",
        "canceled is terminal, an out-of-order update must not revive it"
    );
}

#[tokio::test]
async fn webhook_subscription_updated_cannot_revive_canceled_creator_tier() {
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("ctrevive", "ctrevive@test.com", "password123")
        .await;
    let stripe_sub_id = "sub_ct_revive_1";
    insert_active_creator_sub(&h, user_id, stripe_sub_id, "small_files").await;

    // Cancel, clears the denormalized tier.
    let resp = post_event_json_with_id(
        &mut h,
        "evt_ctrevive_del",
        "customer.subscription.deleted",
        make_subscription(stripe_sub_id, "canceled"),
    )
    .await;
    assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);

    // Out-of-order active update must not revive the canceled creator sub.
    let resp = post_event_json_with_id(
        &mut h,
        "evt_ctrevive_upd",
        "customer.subscription.updated",
        make_subscription(stripe_sub_id, "active"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "update must not error: {}",
        resp.text
    );

    let status: String = sqlx::query_scalar(
        "SELECT status FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        status, "canceled",
        "canceled creator sub must not be revived"
    );
    let tier: Option<String> = sqlx::query_scalar("SELECT creator_tier FROM users WHERE id = $1")
        .bind(user_id)
        .fetch_one(&h.db)
        .await
        .unwrap();
    assert_eq!(
        tier, None,
        "denormalized tier must stay cleared after a refused revival"
    );
}

#[tokio::test]
async fn webhook_subscription_updated_cannot_revive_canceled_fan_plus() {
    // Fan+ was the sibling the Run #11 revival guard missed (Run #12 SERIOUS).
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("fprevive", "fprevive@test.com", "password123")
        .await;

    let stripe_sub_id = "sub_fp_revive_1";
    sqlx::query(
        r"INSERT INTO fan_plus_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
           VALUES ($1, $2, 'cus_fp_revive', 'active', NOW() + interval '30 days')",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    // Cancel (terminal).
    let resp = post_event_json_with_id(
        &mut h,
        "evt_fp_revive_del",
        "customer.subscription.deleted",
        make_subscription(stripe_sub_id, "canceled"),
    )
    .await;
    assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);

    // Out-of-order active update must NOT revive it.
    let resp = post_event_json_with_id(
        &mut h,
        "evt_fp_revive_upd",
        "customer.subscription.updated",
        make_subscription(stripe_sub_id, "active"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "update must not error: {}",
        resp.text
    );

    // The downstream "is this user Fan+?" check must classify them as NOT active.
    let active: bool = sqlx::query_scalar(
        "SELECT EXISTS(SELECT 1 FROM fan_plus_subscriptions \
         WHERE user_id = $1 AND status = 'active' AND canceled_at IS NULL)",
    )
    .bind(user_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert!(
        !active,
        "a canceled Fan+ sub must not be revived by an out-of-order update"
    );
}

// Canceled is terminal for the PERIOD too: an out-of-order invoice.paid must
// not refresh current_period_* on a canceled row. This is the sibling the
// status guard used to miss, now status + period are written together under
// one guard, so the period write can't bypass it. (Run #13 chronic fix.)

#[tokio::test]
async fn webhook_invoice_paid_cannot_refresh_period_on_canceled_sub() {
    let mut h = TestHarness::with_stripe().await;
    let fix = setup_subscription_fixture(&mut h).await;
    let stripe_sub_id = "sub_period_guard_1";
    insert_active_subscription(&h, &fix, stripe_sub_id).await;

    // Cancel (terminal).
    let resp = post_event_json_with_id(
        &mut h,
        "evt_periodguard_del",
        "customer.subscription.deleted",
        make_subscription(stripe_sub_id, "canceled"),
    )
    .await;
    assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);

    let before: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();

    // A stray invoice.payment_succeeded (period_end 1_702_592_000) arrives after cancel.
    let resp = post_event_json(
        &mut h,
        "invoice.payment_succeeded",
        make_invoice(stripe_sub_id, "subscription_cycle"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "invoice must not error: {}",
        resp.text
    );

    let after: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT current_period_end FROM subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        after, before,
        "period must not be refreshed on a canceled subscription"
    );
    assert_ne!(
        after.map(|d| d.timestamp()),
        Some(1_702_592_000),
        "invoice period must not land on a canceled row"
    );
}

#[tokio::test]
async fn webhook_invoice_paid_cannot_refresh_period_on_canceled_creator_tier() {
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("ctperiod", "ctperiod@test.com", "password123")
        .await;
    let stripe_sub_id = "sub_ct_period_1";
    insert_active_creator_sub(&h, user_id, stripe_sub_id, "small_files").await;

    let resp = post_event_json_with_id(
        &mut h,
        "evt_ctperiod_del",
        "customer.subscription.deleted",
        make_subscription(stripe_sub_id, "canceled"),
    )
    .await;
    assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);

    let before: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT current_period_end FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();

    let resp = post_event_json(
        &mut h,
        "invoice.payment_succeeded",
        make_invoice(stripe_sub_id, "subscription_cycle"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "invoice must not error: {}",
        resp.text
    );

    let after: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT current_period_end FROM creator_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        after, before,
        "period must not be refreshed on a canceled creator sub"
    );
}

#[tokio::test]
async fn webhook_invoice_paid_cannot_refresh_period_on_canceled_fan_plus() {
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("fpperiod", "fpperiod@test.com", "password123")
        .await;
    let stripe_sub_id = "sub_fp_period_1";
    sqlx::query(
        r"INSERT INTO fan_plus_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
           VALUES ($1, $2, 'cus_fp_period', 'active', to_timestamp(1_000_000_000))",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    let resp = post_event_json_with_id(
        &mut h,
        "evt_fpperiod_del",
        "customer.subscription.deleted",
        make_subscription(stripe_sub_id, "canceled"),
    )
    .await;
    assert_eq!(resp.status.as_u16(), 200, "delete failed: {}", resp.text);

    // Stray invoice.paid with a DIFFERENT period_end (1_702_592_000) must be refused.
    let resp = post_event_json(
        &mut h,
        "invoice.payment_succeeded",
        make_invoice(stripe_sub_id, "subscription_cycle"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "invoice must not error: {}",
        resp.text
    );

    let after: chrono::DateTime<chrono::Utc> = sqlx::query_scalar(
        "SELECT current_period_end FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
    )
    .bind(stripe_sub_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    assert_eq!(
        after.timestamp(),
        1_000_000_000,
        "period must stay frozen on a canceled Fan+ sub, not jump to the invoice's period_end"
    );
}

#[tokio::test]
async fn fan_plus_renewal_credit_issued_once_across_duplicate_deliveries() {
    // Regression for the Run #22 SERIOUS: webhook event-dedup is a check-then-act
    // read keyed on event id; a duplicate `invoice.payment_succeeded` delivery
    // that escapes it (distinct event id, same invoice/period) must NOT mint a
    // second $5 Fan+ credit. The per-(stripe_sub_id, period_end) idempotency
    // guard makes the renewal credit a single DB-level write.
    let mut h = TestHarness::with_stripe().await;
    let user_id = h
        .signup("fpcredit", "fpcredit@test.com", "password123")
        .await;
    let stripe_sub_id = "sub_fp_credit_idem";
    sqlx::query(
        r"INSERT INTO fan_plus_subscriptions
           (user_id, stripe_subscription_id, stripe_customer_id, status, current_period_end)
           VALUES ($1, $2, 'cus_fp_credit', 'active', to_timestamp(1_700_000_000))",
    )
    .bind(user_id)
    .bind(stripe_sub_id)
    .execute(&h.db)
    .await
    .unwrap();

    // The platform-wide $5 credit codes minted for this user.
    async fn credit_count(h: &TestHarness, user_id: UserId) -> i64 {
        sqlx::query_scalar(
            "SELECT COUNT(*) FROM promo_codes \
             WHERE creator_id = $1 AND is_platform_wide = true AND discount_value = 500",
        )
        .bind(user_id)
        .fetch_one(&h.db)
        .await
        .unwrap()
    }

    // First renewal delivery mints exactly one credit.
    let resp = post_event_json_with_id(
        &mut h,
        "evt_fpcredit_1",
        "invoice.payment_succeeded",
        make_invoice(stripe_sub_id, "subscription_cycle"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "first renewal failed: {}",
        resp.text
    );
    assert_eq!(
        credit_count(&h, user_id).await,
        1,
        "first renewal must mint one credit"
    );

    // A duplicate delivery of the SAME renewal period under a different event id
    // (so event-dedup does not catch it) must not mint a second credit.
    let resp = post_event_json_with_id(
        &mut h,
        "evt_fpcredit_2",
        "invoice.payment_succeeded",
        make_invoice(stripe_sub_id, "subscription_cycle"),
    )
    .await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "duplicate renewal failed: {}",
        resp.text
    );
    assert_eq!(
        credit_count(&h, user_id).await,
        1,
        "a duplicate delivery of the same renewal must not double-issue the credit"
    );
}

// PAY-S1: a failed pending-refund process must RELEASE its claim
//
// check_pending_refund claims the pending refund (matched_at = NOW()) before the
// fallible handle_charge_refunded. If that work fails, the claim must be released
// (matched_at -> NULL) so the stale-refund sweep escalates it and a later
// delivery can retry. Without the release the row is lost forever: the sweep
// filters `matched_at IS NULL`, and the original charge.refunded event was
// already marked processed so Stripe never redelivers. We force the failure with
// a trigger that raises when the refund flips the transaction to 'refunded'.

#[tokio::test]
async fn pending_refund_releases_claim_when_processing_fails() {
    let mut h = TestHarness::with_stripe().await;

    let buyer_id = h
        .signup("prfbuyer", "prfbuyer@test.com", "password123")
        .await;
    h.client.post_form("/logout", "").await;
    let seller_id = h
        .signup("prfseller", "prfseller@test.com", "password123")
        .await;
    h.grant_creator(seller_id).await;
    h.client.post_form("/logout", "").await;
    h.login("prfseller", "password123").await;

    let resp = h
        .client
        .post_form("/api/projects", "slug=prfproj&title=PRF+Project")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();
    let resp = h
        .client
        .post_form(
            &format!("/api/projects/{project_id}/items"),
            "title=PRF+Track&price_cents=999&item_type=audio",
        )
        .await;
    let item: Value = resp.json();
    let item_id = item["id"].as_str().unwrap().to_string();

    let session_id = "cs_prf_001";
    let pi = "pi_faultrefund_001";

    // Pending transaction for this checkout session, carrying the sentinel PI.
    sqlx::query(
        r"INSERT INTO transactions
           (buyer_id, seller_id, item_id, amount_cents, status,
            stripe_checkout_session_id, stripe_payment_intent_id, item_title, seller_username)
           VALUES ($1, $2, $3::uuid, 999, 'pending', $4, $5, 'PRF Track', 'prfseller')",
    )
    .bind(buyer_id)
    .bind(seller_id)
    .bind(&item_id)
    .bind(session_id)
    .bind(pi)
    .execute(&h.db)
    .await
    .unwrap();

    // A full refund for this PI arrived first and was queued.
    sqlx::query("INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded) VALUES ($1, 999, 999)")
        .bind(pi).execute(&h.db).await.unwrap();

    // Force handle_charge_refunded to fail: raise when the refund flips this PI's
    // transaction to 'refunded' (the completion -> 'completed' update is untouched).
    sqlx::query(
        r"CREATE OR REPLACE FUNCTION test_refund_fault() RETURNS trigger AS $$
           BEGIN
             IF NEW.status = 'refunded' AND COALESCE(NEW.stripe_payment_intent_id,'') LIKE '%faultrefund%' THEN
               RAISE EXCEPTION 'injected refund fault';
             END IF;
             RETURN NEW;
           END; $$ LANGUAGE plpgsql",
    ).execute(&h.db).await.unwrap();
    sqlx::query("CREATE TRIGGER test_refund_fault BEFORE UPDATE ON transactions FOR EACH ROW EXECUTE FUNCTION test_refund_fault()")
        .execute(&h.db).await.unwrap();

    // Complete the checkout: fulfillment completes the tx, then check_pending_refund
    // claims the refund and tries to process it -> trigger raises -> claim released.
    let mut meta = HashMap::new();
    meta.insert("buyer_id".to_string(), buyer_id.to_string());
    meta.insert("seller_id".to_string(), seller_id.to_string());
    meta.insert("item_id".to_string(), item_id.clone());
    let session = serde_json::json!({
        "id": session_id,
        "object": "checkout_session",
        "mode": "payment",
        "metadata": meta,
        "payment_intent": pi,
    });
    let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "checkout webhook should still 200: {}",
        resp.text
    );

    // The claim must be RELEASED so the stale-refund sweep can pick it up.
    let matched_at: Option<chrono::DateTime<chrono::Utc>> =
        sqlx::query_scalar("SELECT matched_at FROM pending_refunds WHERE payment_intent_id = $1")
            .bind(pi)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert!(
        matched_at.is_none(),
        "a failed pending-refund process must release its claim (matched_at NULL), else the refund is lost"
    );

    // The refund rolled back atomically: the transaction is NOT marked refunded.
    let status: String =
        sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_payment_intent_id = $1")
            .bind(pi)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_ne!(
        status, "refunded",
        "the failed refund must not have committed"
    );
}

// Complement: a pending refund that processes SUCCESSFULLY on completion is
// consumed (matched_at set, transaction refunded), pins that the happy path
// still claims-and-keeps, so the release above is failure-only.
#[tokio::test]
async fn pending_refund_processed_on_completion_marks_matched() {
    let mut h = TestHarness::with_stripe().await;

    let buyer_id = h
        .signup("prfok_buyer", "prfokb@test.com", "password123")
        .await;
    h.client.post_form("/logout", "").await;
    let seller_id = h
        .signup("prfok_seller", "prfoks@test.com", "password123")
        .await;
    h.grant_creator(seller_id).await;
    h.client.post_form("/logout", "").await;
    h.login("prfok_seller", "password123").await;

    let resp = h
        .client
        .post_form("/api/projects", "slug=prfokproj&title=PRFOK")
        .await;
    let project: Value = resp.json();
    let project_id = project["id"].as_str().unwrap().to_string();
    let resp = h
        .client
        .post_form(
            &format!("/api/projects/{project_id}/items"),
            "title=PRFOK+Track&price_cents=999&item_type=audio",
        )
        .await;
    let item: Value = resp.json();
    let item_id = item["id"].as_str().unwrap().to_string();

    let session_id = "cs_prfok_001";
    let pi = "pi_prfok_001";
    sqlx::query(
        r"INSERT INTO transactions
           (buyer_id, seller_id, item_id, amount_cents, status,
            stripe_checkout_session_id, stripe_payment_intent_id, item_title, seller_username)
           VALUES ($1, $2, $3::uuid, 999, 'pending', $4, $5, 'PRFOK Track', 'prfok_seller')",
    )
    .bind(buyer_id)
    .bind(seller_id)
    .bind(&item_id)
    .bind(session_id)
    .bind(pi)
    .execute(&h.db)
    .await
    .unwrap();
    sqlx::query("INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded) VALUES ($1, 999, 999)")
        .bind(pi).execute(&h.db).await.unwrap();

    let mut meta = HashMap::new();
    meta.insert("buyer_id".to_string(), buyer_id.to_string());
    meta.insert("seller_id".to_string(), seller_id.to_string());
    meta.insert("item_id".to_string(), item_id.clone());
    let session = serde_json::json!({
        "id": session_id, "object": "checkout_session", "mode": "payment",
        "metadata": meta, "payment_intent": pi,
    });
    let resp = post_event_json(&mut h, "checkout.session.completed", session).await;
    assert_eq!(
        resp.status.as_u16(),
        200,
        "checkout webhook failed: {}",
        resp.text
    );

    let matched_at: Option<chrono::DateTime<chrono::Utc>> =
        sqlx::query_scalar("SELECT matched_at FROM pending_refunds WHERE payment_intent_id = $1")
            .bind(pi)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert!(
        matched_at.is_some(),
        "a successfully processed pending refund stays claimed"
    );

    let status: String =
        sqlx::query_scalar("SELECT status FROM transactions WHERE stripe_payment_intent_id = $1")
            .bind(pi)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        status, "refunded",
        "the out-of-order refund must apply on completion"
    );
}

// PAY-S1 crash window: a refund claimed but never completed (process killed
// between claim and completion) must be surfaced by the stale-refund sweep, not
// silently lost. The old sweep filtered `matched_at IS NULL`, so a row that was
// claimed (matched_at set) but whose refund never finished was invisible forever.
// The fix adds `completed_at` and the sweep (`get_stale_refunds`) filters on that.
// This pins the exact selection predicate the sweep relies on (db::pending_refunds
// is pub(crate), so the invariant is asserted at the SQL layer it queries).
#[tokio::test]
async fn pending_refund_claimed_but_uncompleted_is_swept() {
    let h = TestHarness::with_stripe().await;

    // A claimed-but-incomplete row: refund arrived >24h ago (created_at old),
    // claim taken (matched_at set), but completed_at NULL, the process died
    // before recording completion.
    let pi_stuck = "pi_prf_crash_001";
    sqlx::query(
        "INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded, created_at, matched_at, completed_at)
         VALUES ($1, 999, 999, NOW() - INTERVAL '25 hours', NOW() - INTERVAL '25 hours', NULL)",
    ).bind(pi_stuck).execute(&h.db).await.unwrap();

    // A fully-completed row of the same age must NOT be escalated.
    let pi_done = "pi_prf_done_001";
    sqlx::query(
        "INSERT INTO pending_refunds (payment_intent_id, amount, amount_refunded, created_at, matched_at, completed_at)
         VALUES ($1, 999, 999, NOW() - INTERVAL '25 hours', NOW() - INTERVAL '25 hours', NOW() - INTERVAL '25 hours')",
    ).bind(pi_done).execute(&h.db).await.unwrap();

    // The sweep predicate from `get_stale_refunds`: not completed, not escalated,
    // older than the cutoff. The stuck row surfaces; the completed one does not.
    // Under the old `matched_at IS NULL` filter the stuck row would have been
    // invisible (matched_at was set), silently dropping the owed refund.
    let swept: Vec<String> = sqlx::query_scalar(
        "SELECT payment_intent_id FROM pending_refunds
         WHERE completed_at IS NULL AND escalated_at IS NULL AND created_at < NOW() - INTERVAL '24 hours'
           AND payment_intent_id = ANY($1)",
    )
    .bind(vec![pi_stuck.to_string(), pi_done.to_string()])
    .fetch_all(&h.db)
    .await
    .unwrap();

    assert!(
        swept.contains(&pi_stuck.to_string()),
        "a claimed-but-uncompleted refund must surface in the stale sweep (PAY-S1 crash window)"
    );
    assert!(
        !swept.contains(&pi_done.to_string()),
        "a completed refund must not be escalated"
    );
}