//! Crash-recovery integration tests for the purchase webhook finalizer
//! (ultra-fuzz Payments A+, finding M-Pay1).
//!
//! A `checkout.session.completed` first attempt can crash AFTER committing the
//! completed transaction status but BEFORE running the secondary effects
//! (license-key mint, revenue splits, emails). Because the event is not marked
//! processed until the handler returns Ok, Stripe redelivers. On redelivery the
//! transaction is already completed, so `complete_transaction` returns None; the
//! recovery branch must re-fetch the completed rows and re-run the idempotent
//! finalizer so the buyer ends up with their key and the collaborators with
//! their splits. Running it again must NOT create duplicates.

use crate::harness::TestHarness;
use makenotwork::db;
use serde_json::Value;
use std::collections::HashMap;

/// Create a Stripe-connected creator with a published paid item that has license
/// keys enabled. Returns (seller_id, project_id, item_id).
async fn setup_keyed_item(h: &mut TestHarness, price_cents: i32) -> (db::UserId, String, String) {
    let seller_id = h.signup("kseller", "kseller@test.com", "pass1234").await;
    h.grant_creator(seller_id).await;
    sqlx::query("UPDATE users SET stripe_account_id = 'acct_mock_kseller', stripe_charges_enabled = true WHERE id = $1")
        .bind(seller_id)
        .execute(&h.db)
        .await
        .unwrap();

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

    let resp = h
        .client
        .post_form("/api/projects", "slug=kshop&title=KShop")
        .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"),
            &format!("title=Plugin&price_cents={price_cents}&item_type=plugin"),
        )
        .await;
    let item: Value = resp.json();
    let item_id = item["id"].as_str().unwrap().to_string();

    // Enable license keys so finalize mints one on purchase.
    h.client
        .put_form(
            &format!("/api/items/{item_id}/license-settings"),
            "enable_license_keys=on&default_max_activations=3",
        )
        .await;

    h.client
        .put_form(&format!("/api/projects/{project_id}"), "is_public=true")
        .await;
    h.client
        .put_form(&format!("/api/items/{item_id}"), "is_public=true")
        .await;

    (seller_id, project_id, item_id)
}

/// Deliver a `checkout.session.completed` webhook for a single-item purchase.
async fn deliver_purchase_webhook(
    h: &mut TestHarness,
    event_id: &str,
    session_id: &str,
    buyer_id: db::UserId,
    seller_id: db::UserId,
    item_id: &str,
) -> u16 {
    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.to_string());
    let session = serde_json::json!({
        "id": session_id,
        "object": "checkout_session",
        "mode": "payment",
        "metadata": meta,
        "payment_intent": "pi_crash_recovery",
    });
    let payload = serde_json::json!({
        "id": event_id,
        "type": "checkout.session.completed",
        "data": {"object": session},
    })
    .to_string();
    let signature = crate::harness::stripe::sign_webhook_payload(
        &payload,
        crate::harness::stripe::TEST_WEBHOOK_SECRET,
    );
    h.client
        .request_with_headers(
            "POST",
            "/stripe/webhook",
            Some(&payload),
            &[
                ("stripe-signature", &signature),
                ("content-type", "application/json"),
            ],
        )
        .await
        .status
        .as_u16()
}

#[tokio::test]
async fn crash_recovery_reruns_finalize_idempotently() {
    let mut h = TestHarness::with_mocks().await;
    let (seller_id, project_id, item_id) = setup_keyed_item(&mut h, 1000).await;

    // Collaborator with a 50% split.
    let collab_id = h.signup("kcollab", "kcollab@test.com", "pass1234").await;
    h.client.post_form("/logout", "").await;
    sqlx::query(
        "INSERT INTO project_members (project_id, user_id, role, split_percent, added_by) VALUES ($1::uuid, $2, 'member', 50, $3)",
    )
    .bind(&project_id)
    .bind(collab_id)
    .bind(seller_id)
    .execute(&h.db)
    .await
    .unwrap();

    // Buyer initiates checkout (creates the pending transaction).
    let buyer_id = h.signup("kbuyer", "kbuyer@test.com", "pass1234").await;
    h.client
        .post_form(
            &format!("/stripe/checkout/{item_id}"),
            "share_contact=false",
        )
        .await;
    let session_id: String = sqlx::query_scalar(
        "SELECT stripe_checkout_session_id FROM transactions WHERE buyer_id = $1 AND status = 'pending'",
    )
    .bind(buyer_id)
    .fetch_one(&h.db)
    .await
    .unwrap();

    // First delivery completes the purchase and runs finalize normally.
    let status = deliver_purchase_webhook(
        &mut h,
        "evt_crash_1",
        &session_id,
        buyer_id,
        seller_id,
        &item_id,
    )
    .await;
    assert_eq!(status, 200, "first webhook should succeed");
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    let tx_id: db::TransactionId =
        sqlx::query_scalar("SELECT id FROM transactions WHERE stripe_checkout_session_id = $1")
            .bind(&session_id)
            .fetch_one(&h.db)
            .await
            .unwrap();

    // --- Simulate a first attempt that crashed before finalize: the tx is
    // completed but the secondary effects never landed. Wipe key + splits. ---
    sqlx::query("DELETE FROM license_keys WHERE transaction_id = $1")
        .bind(tx_id)
        .execute(&h.db)
        .await
        .unwrap();
    sqlx::query("DELETE FROM revenue_splits WHERE transaction_id = $1")
        .bind(tx_id)
        .execute(&h.db)
        .await
        .unwrap();

    let key_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM license_keys WHERE transaction_id = $1")
            .bind(tx_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(key_count, 0, "precondition: key wiped");
    let split_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE transaction_id = $1")
            .bind(tx_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(split_count, 0, "precondition: splits wiped");

    // --- Redelivery (new event id, so dedup passes; the tx is already completed
    // so complete_transaction returns None -> the recovery branch re-runs the
    // idempotent finalizer). ---
    let status = deliver_purchase_webhook(
        &mut h,
        "evt_crash_2",
        &session_id,
        buyer_id,
        seller_id,
        &item_id,
    )
    .await;
    assert_eq!(status, 200, "recovery webhook should succeed");
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    let key_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM license_keys WHERE transaction_id = $1")
            .bind(tx_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(key_count, 1, "recovery should mint the missing license key");
    let split_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE transaction_id = $1")
            .bind(tx_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(split_count, 1, "recovery should record the missing split");

    // --- A SECOND recovery (another redelivery) must NOT create duplicates. ---
    let status = deliver_purchase_webhook(
        &mut h,
        "evt_crash_3",
        &session_id,
        buyer_id,
        seller_id,
        &item_id,
    )
    .await;
    assert_eq!(status, 200, "second recovery webhook should succeed");
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    let key_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM license_keys WHERE transaction_id = $1")
            .bind(tx_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(key_count, 1, "idempotent: still exactly one license key");
    let split_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE transaction_id = $1")
            .bind(tx_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(split_count, 1, "idempotent: still exactly one split");
}

/// Deliver a `checkout.session.completed` webhook for a tip on `session_id`.
async fn deliver_tip_webhook(
    h: &mut TestHarness,
    event_id: &str,
    session_id: &str,
    tipper_id: db::UserId,
    recipient_id: db::UserId,
) -> u16 {
    let mut meta = HashMap::new();
    meta.insert("checkout_type".to_string(), "tip".to_string());
    meta.insert("tipper_id".to_string(), tipper_id.to_string());
    meta.insert("recipient_id".to_string(), recipient_id.to_string());
    let session = serde_json::json!({
        "id": session_id,
        "object": "checkout_session",
        "mode": "payment",
        "metadata": meta,
        "payment_intent": "pi_tip_crash_recovery",
    });
    let payload = serde_json::json!({
        "id": event_id,
        "type": "checkout.session.completed",
        "data": {"object": session},
    })
    .to_string();
    let signature = crate::harness::stripe::sign_webhook_payload(
        &payload,
        crate::harness::stripe::TEST_WEBHOOK_SECRET,
    );
    h.client
        .request_with_headers(
            "POST",
            "/stripe/webhook",
            Some(&payload),
            &[
                ("stripe-signature", &signature),
                ("content-type", "application/json"),
            ],
        )
        .await
        .status
        .as_u16()
}

/// Tip revenue splits must survive a crash between `complete_tip` and the split
/// write, and a redelivery must not double-credit collaborators (Run 20 chronic
/// money-loss). Mirrors the purchase recovery test for the tip path.
#[tokio::test]
async fn tip_split_crash_recovery_is_idempotent() {
    let mut h = TestHarness::with_mocks().await;

    // Recipient (creator) with a project and a 50% collaborator.
    let recipient_id = h.signup("trecip", "trecip@test.com", "pass1234").await;
    let project_id: db::ProjectId = sqlx::query_scalar(
        "INSERT INTO projects (user_id, slug, title) VALUES ($1, 'tshop', 'TShop') RETURNING id",
    )
    .bind(recipient_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    let collab_id = h.signup("tcollab", "tcollab@test.com", "pass1234").await;
    sqlx::query(
        "INSERT INTO project_members (project_id, user_id, role, split_percent, added_by) VALUES ($1, $2, 'member', 50, $3)",
    )
    .bind(project_id)
    .bind(collab_id)
    .bind(recipient_id)
    .execute(&h.db)
    .await
    .unwrap();

    // A tipper and a pending tip attributed to the project.
    let tipper_id = h.signup("ttipper", "ttipper@test.com", "pass1234").await;
    let session_id = "cs_tip_crash_001";
    db::tips::create_tip(
        &h.db,
        tipper_id,
        recipient_id,
        Some(project_id),
        1000,
        None,
        session_id,
    )
    .await
    .unwrap();
    let tip_id: db::TipId =
        sqlx::query_scalar("SELECT id FROM tips WHERE stripe_checkout_session_id = $1")
            .bind(session_id)
            .fetch_one(&h.db)
            .await
            .unwrap();

    // First delivery completes the tip and records the split.
    let status =
        deliver_tip_webhook(&mut h, "evt_tip_1", session_id, tipper_id, recipient_id).await;
    assert_eq!(status, 200, "first tip webhook should succeed");
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    let split_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE tip_id = $1")
            .bind(tip_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(
        split_count, 1,
        "first delivery records the collaborator split"
    );

    // Simulate a crash before the split write: tip stays completed, split gone.
    sqlx::query("DELETE FROM revenue_splits WHERE tip_id = $1")
        .bind(tip_id)
        .execute(&h.db)
        .await
        .unwrap();

    // Redelivery (new event id): complete_tip returns None; the recovery branch
    // re-fetches the tip and re-runs the idempotent split write.
    let status =
        deliver_tip_webhook(&mut h, "evt_tip_2", session_id, tipper_id, recipient_id).await;
    assert_eq!(status, 200, "recovery tip webhook should succeed");
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    let split_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE tip_id = $1")
            .bind(tip_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(split_count, 1, "recovery re-records the missing tip split");

    // A second recovery must NOT duplicate (ON CONFLICT DO NOTHING, migration 163).
    let status =
        deliver_tip_webhook(&mut h, "evt_tip_3", session_id, tipper_id, recipient_id).await;
    assert_eq!(status, 200, "second recovery tip webhook should succeed");
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    let split_count: i64 =
        sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE tip_id = $1")
            .bind(tip_id)
            .fetch_one(&h.db)
            .await
            .unwrap();
    assert_eq!(split_count, 1, "idempotent: still exactly one tip split");
}

/// The partial unique index on license_keys (transaction_id) structurally rejects
/// a second auto-minted key for the same transaction even if the pre-check is
/// bypassed.
#[tokio::test]
async fn license_keys_transaction_id_unique_index_rejects_double_mint() {
    let h = TestHarness::new().await;

    // Minimal fixtures: a user, project, item, and completed transaction.
    let owner_id: db::UserId = sqlx::query_scalar(
        "INSERT INTO users (username, email, password_hash, email_verified) \
         VALUES ('idxowner', 'idxowner@test.com', 'x', true) RETURNING id",
    )
    .fetch_one(&h.db)
    .await
    .unwrap();
    let project_id: db::ProjectId = sqlx::query_scalar(
        "INSERT INTO projects (user_id, slug, title) VALUES ($1, 'idxproj', 'Idx') RETURNING id",
    )
    .bind(owner_id)
    .fetch_one(&h.db)
    .await
    .unwrap();
    let item_id: db::ItemId = sqlx::query_scalar(
        "INSERT INTO items (project_id, title, item_type, price_cents, slug) VALUES ($1, 'Idx Item', 'plugin', 0, 'idx-item') RETURNING id",
    ).bind(project_id).fetch_one(&h.db).await.unwrap();
    let tx_id: db::TransactionId = sqlx::query_scalar(
        "INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, status) \
         VALUES ($1, $1, $2, 0, 'completed') RETURNING id",
    )
    .bind(owner_id)
    .bind(item_id)
    .fetch_one(&h.db)
    .await
    .unwrap();

    // First key for the transaction: OK.
    let first = sqlx::query(
        "INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code) VALUES ($1, $2, $3, 'aaa-bbb-ccc')",
    ).bind(item_id).bind(owner_id).bind(tx_id).execute(&h.db).await;
    assert!(first.is_ok(), "first key insert should succeed: {first:?}");

    // Second key for the SAME transaction: rejected by the partial unique index.
    let second = sqlx::query(
        "INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code) VALUES ($1, $2, $3, 'ddd-eee-fff')",
    ).bind(item_id).bind(owner_id).bind(tx_id).execute(&h.db).await;
    assert!(
        second.is_err(),
        "second key for same transaction must be rejected by unique index"
    );

    // A manually-created key (transaction_id NULL) is unconstrained: many allowed.
    for code in ["man-1", "man-2"] {
        let manual = sqlx::query(
            "INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code) VALUES ($1, $2, NULL, $3)",
        ).bind(item_id).bind(owner_id).bind(code).execute(&h.db).await;
        assert!(
            manual.is_ok(),
            "manual key {code} should be allowed (partial index excludes NULL)"
        );
    }
}