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

use crate::harness::stripe::{sign_webhook_payload, TEST_WEBHOOK_SECRET, TEST_WEBHOOK_SECRET_V2};
use crate::harness::TestHarness;
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/{}/items", project_id),
            "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/{}/items", project_id),
            "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": 1700000000,
                "current_period_end": 1702592000,
                "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
// ---------------------------------------------------------------------------

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": 1702592000,
                "current_period_end": 1705184000,
                "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(), 1702592000);
    assert_eq!(period_end.unwrap().timestamp(), 1705184000);
}

#[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": 1702592000,
        "period_end": 1705184000,
        "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(), 1702592000);
    assert_eq!(period_end.unwrap().timestamp(), 1705184000);
}

#[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": 1700000000,
        "period_end": 1702592000,
        "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/{}/items", project_id),
            "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");
}

/// N1 regression: the dedup mark commits BEFORE the handler runs, so the
/// mark/unmark layering must be exact — a wrong move either double-processes a
/// redelivered event or silently short-circuits it forever. This pins the
/// event-ID dedup primitive directly (the `_idempotent` tests above use
/// distinct event IDs, so they only exercise the handler's SQL idempotency, not
/// this layer).
#[tokio::test]
async fn webhook_event_dedup_and_unmark_layering() {
    use makenotwork::db::webhook_events;

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

    // First sighting claims the event.
    assert!(
        webhook_events::try_mark_event_processed(&h.db, id).await.unwrap(),
        "first mark should claim the event"
    );
    // Redelivery of the same id is deduped — the handler must not run again.
    assert!(
        !webhook_events::try_mark_event_processed(&h.db, id).await.unwrap(),
        "second mark of the same id must be deduped"
    );
    // Handler + retry-queue both failed → unmark so Stripe redelivery re-runs
    // (otherwise it would short-circuit at the dedup check and 200 without work).
    webhook_events::unmark_event_processed(&h.db, id).await.unwrap();
    assert!(
        webhook_events::try_mark_event_processed(&h.db, id).await.unwrap(),
        "after unmark, the event must be claimable again so redelivery actually processes"
    );
}

// ---------------------------------------------------------------------------
// 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": 1700000000_i64,
                "current_period_end": 1702592000_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": 1700000000_i64,
        "period_end": 1702592000_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 = 1702592000).
    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(), 1702592000);
}

#[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(), 1700000000);
    assert_eq!(period_end.unwrap().timestamp(), 1702592000);

    // 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(), 1702592000, "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/{}/items", project_id),
            "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);
}

// ---------------------------------------------------------------------------
// 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": 1700000000_i64,
        "period_end": 1702592000_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 1702592000) 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(1702592000), "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(1000000000))"#,
    )
    .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 (1702592000) 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(), 1000000000, "period must stay frozen on a canceled Fan+ sub, not jump to the invoice's period_end");
}