use super::{
    Cents, ClaimToken, DbPurchaseRow, DbTransaction, DbTransactionExportRow, DownloadToken, ItemId,
    KeyCode, PgPool, ProjectId, PromoCodeId, Result, TransactionId, UserId,
};

/// Parameters for creating a pending Stripe checkout transaction.
pub struct CreateTransactionParams<'a> {
    pub buyer_id: Option<UserId>,
    pub seller_id: UserId,
    /// `None` for project-level purchases (no specific item).
    pub item_id: Option<ItemId>,
    pub amount_cents: Cents,
    pub platform_fee_cents: Cents,
    pub stripe_checkout_session_id: &'a str,
    pub item_title: &'a str,
    pub seller_username: &'a str,
    pub share_contact: bool,
    /// Set for project-level purchases; `None` for item purchases.
    pub project_id: Option<ProjectId>,
    /// Promo code used for this checkout (for releasing reservations on cleanup).
    pub promo_code_id: Option<PromoCodeId>,
    /// Guest buyer's email (set for guest checkouts, None for logged-in).
    pub guest_email: Option<&'a str>,
    /// Cents MNW owes the seller as reimbursement for a platform-funded credit
    /// (the Fan+ renewal credit) applied to this sale; `0` for ordinary sales. The
    /// scheduler settles it via a platform -> connected transfer once the
    /// transaction completes (see `db::platform_credits`).
    pub platform_credit_cents: i64,
}

/// Common parameters for claiming a free item (direct, discount code, or download code).
pub struct ClaimParams<'a> {
    pub buyer_id: UserId,
    pub item_id: ItemId,
    pub seller_id: UserId,
    pub item_title: &'a str,
    pub seller_username: &'a str,
    pub share_contact: bool,
    /// If this claim was granted via a bundle purchase, the parent transaction ID.
    pub parent_transaction_id: Option<TransactionId>,
    /// Cents MNW owes the seller when a platform-wide (Fan+) credit made this item
    /// free, the seller is reimbursed the item's price via a platform transfer so
    /// they are still paid. `0` for ordinary free claims (genuinely-free items,
    /// seller-issued free-access codes, bundle grants).
    pub platform_credit_cents: i64,
}

/// Record a new pending transaction for a Stripe checkout session.
#[tracing::instrument(skip_all)]
pub async fn create_transaction<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: &CreateTransactionParams<'_>,
) -> Result<DbTransaction> {
    let tx = sqlx::query_as!(
        DbTransaction,
        r#"
        INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact, project_id, promo_code_id, guest_email, platform_credit_cents)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
        RETURNING
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        "#,
        params.buyer_id as Option<UserId>,
        params.seller_id as UserId,
        params.item_id as Option<ItemId>,
        params.amount_cents as Cents,
        params.platform_fee_cents as Cents,
        params.stripe_checkout_session_id,
        params.item_title,
        params.seller_username,
        params.share_contact,
        params.project_id as Option<ProjectId>,
        params.promo_code_id as Option<PromoCodeId>,
        params.guest_email,
        params.platform_credit_cents,
    )
    .fetch_one(executor)
    .await?;

    Ok(tx)
}

/// Complete a guest transaction: mark it completed, record the guest email, and
/// mint a `claim_token` so the buyer can later attach the purchase to an account.
///
/// Guest purchases always land unclaimed (`buyer_id` NULL); attachment to a user
/// happens out of band via [`attach_guest_purchases_by_email`] at signup/email
/// verification. (A prior `existing_user_id` auto-attach parameter was always
/// passed `None` and has been removed, Run #1 NOTE, dead foot-gun.)
#[tracing::instrument(skip_all)]
pub async fn complete_guest_transaction<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    stripe_checkout_session_id: &str,
    stripe_payment_intent_id: Option<&str>,
    guest_email: &str,
) -> Result<Option<DbTransaction>> {
    let claim_token = ClaimToken::new();

    let tx = sqlx::query_as!(
        DbTransaction,
        r#"
        UPDATE transactions
        SET status = 'completed',
            stripe_payment_intent_id = $2,
            completed_at = NOW(),
            guest_email = $3,
            claim_token = $4,
            buyer_id = NULL
        WHERE stripe_checkout_session_id = $1
          AND status = 'pending'
        RETURNING
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        "#,
        stripe_checkout_session_id,
        stripe_payment_intent_id,
        guest_email,
        claim_token as ClaimToken,
    )
    .fetch_optional(executor)
    .await?;

    Ok(tx)
}

/// Attach all unclaimed guest purchases for an email to a user account.
/// Called during signup/email verification to auto-claim prior guest purchases.
#[tracing::instrument(skip_all)]
pub async fn attach_guest_purchases_by_email(
    pool: &PgPool,
    email: &str,
    user_id: UserId,
) -> Result<u64> {
    let result = sqlx::query!(
        r#"
        UPDATE transactions
        SET buyer_id = $1, claimed_by = $1, claim_token = NULL
        WHERE LOWER(guest_email) = LOWER($2)
          AND buyer_id IS NULL
          AND status = 'completed'
        "#,
        user_id as UserId,
        email,
    )
    .execute(pool)
    .await?;

    Ok(result.rows_affected())
}

/// Claim a single guest purchase by claim token.
#[tracing::instrument(skip_all)]
pub async fn claim_guest_purchase(
    pool: &PgPool,
    claim_token: ClaimToken,
    user_id: UserId,
) -> Result<Option<DbTransaction>> {
    let tx = sqlx::query_as!(
        DbTransaction,
        r#"
        UPDATE transactions
        SET buyer_id = $2, claimed_by = $2, claim_token = NULL
        WHERE claim_token = $1
          AND buyer_id IS NULL
          AND status = 'completed'
        RETURNING
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        "#,
        claim_token as ClaimToken,
        user_id as UserId,
    )
    .fetch_optional(pool)
    .await?;

    Ok(tx)
}

/// Look up a completed transaction by download token (for guest download links).
#[tracing::instrument(skip_all)]
pub async fn get_transaction_by_download_token(
    pool: &PgPool,
    download_token: DownloadToken,
) -> Result<Option<DbTransaction>> {
    let tx = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        FROM transactions WHERE download_token = $1 AND status = 'completed'
        "#,
        download_token as DownloadToken,
    )
    .fetch_optional(pool)
    .await?;

    Ok(tx)
}

/// Mark a pending transaction as completed (idempotent; returns `None` if already completed).
///
/// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
/// can include this in a larger transaction when needed.
#[tracing::instrument(skip_all)]
pub async fn complete_transaction<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    stripe_checkout_session_id: &str,
    stripe_payment_intent_id: Option<&str>,
) -> Result<Option<DbTransaction>> {
    // Only update if status is 'pending' for idempotency
    // Returns None if transaction was already completed (duplicate webhook)
    let tx = sqlx::query_as!(
        DbTransaction,
        r#"
        UPDATE transactions
        SET status = 'completed',
            stripe_payment_intent_id = $2,
            completed_at = NOW()
        WHERE stripe_checkout_session_id = $1
          AND status = 'pending'
        RETURNING
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        "#,
        stripe_checkout_session_id,
        stripe_payment_intent_id,
    )
    .fetch_optional(executor)
    .await?;

    Ok(tx)
}

/// Complete ALL pending transactions for a cart checkout session.
/// Returns the list of completed transactions (empty if already processed).
#[tracing::instrument(skip_all)]
pub async fn complete_cart_transactions<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    stripe_checkout_session_id: &str,
    stripe_payment_intent_id: Option<&str>,
) -> Result<Vec<DbTransaction>> {
    let txs = sqlx::query_as!(
        DbTransaction,
        r#"
        UPDATE transactions
        SET status = 'completed',
            stripe_payment_intent_id = $2,
            completed_at = NOW()
        WHERE stripe_checkout_session_id = $1
          AND status = 'pending'
        RETURNING
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        "#,
        stripe_checkout_session_id,
        stripe_payment_intent_id,
    )
    .fetch_all(executor)
    .await?;

    Ok(txs)
}

/// Fetch all completed transactions for a checkout session.
///
/// Used on the crash-recovery branch of the purchase/cart webhook handlers: when
/// `complete_transaction` / `complete_cart_transactions` return nothing (the
/// rows were already flipped to completed by a first attempt that crashed before
/// running finalize), this re-reads those completed rows so finalize can re-run
/// idempotently. Covers single and cart purchases since both key on the session.
#[tracing::instrument(skip_all)]
pub async fn get_completed_transactions_for_session<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    stripe_checkout_session_id: &str,
) -> Result<Vec<DbTransaction>> {
    let txs = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        FROM transactions
        WHERE stripe_checkout_session_id = $1 AND status = 'completed'
        "#,
        stripe_checkout_session_id,
    )
    .fetch_all(executor)
    .await?;

    Ok(txs)
}

/// List transactions where the user is the buyer, newest first.
///
/// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
#[tracing::instrument(skip_all)]
pub async fn get_transactions_by_buyer(
    pool: &PgPool,
    buyer_id: UserId,
    limit: Option<i64>,
) -> Result<Vec<DbTransaction>> {
    let txs = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        FROM transactions WHERE buyer_id = $1 ORDER BY created_at DESC LIMIT $2
        "#,
        buyer_id as UserId,
        limit,
    )
    .fetch_all(pool)
    .await?;

    Ok(txs)
}

/// One page of a buyer's purchases for CSV export, newest first.
///
/// Paginated so the purchases export streams in bounded batches rather than
/// loading the buyer's whole history with `limit: None` (ultra-fuzz Run 4 S1).
/// Stable `(created_at, id)` ordering keeps OFFSET batches consistent.
pub async fn get_buyer_transactions_for_export_page(
    pool: &PgPool,
    buyer_id: UserId,
    limit: i64,
    offset: i64,
) -> Result<Vec<DbTransaction>> {
    let txs = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        FROM transactions WHERE buyer_id = $1
        ORDER BY created_at DESC, id DESC
        LIMIT $2 OFFSET $3
        "#,
        buyer_id as UserId,
        limit,
        offset,
    )
    .fetch_all(pool)
    .await?;

    Ok(txs)
}

/// List transactions where the user is the seller, newest first.
///
/// Pass `limit: None` for all rows (exports), or `Some(n)` for dashboard display.
#[tracing::instrument(skip_all)]
pub async fn get_transactions_by_seller(
    pool: &PgPool,
    seller_id: UserId,
    limit: Option<i64>,
) -> Result<Vec<DbTransaction>> {
    let txs = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        FROM transactions WHERE seller_id = $1 ORDER BY created_at DESC LIMIT $2
        "#,
        seller_id as UserId,
        limit,
    )
    .fetch_all(pool)
    .await?;

    Ok(txs)
}

/// Check whether a user has a completed purchase for a given item.
#[tracing::instrument(skip_all)]
pub async fn has_purchased_item(pool: &PgPool, user_id: UserId, item_id: ItemId) -> Result<bool> {
    let count: i64 = sqlx::query_scalar!(
        r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND item_id = $2 AND status = 'completed'"#,
        user_id as UserId,
        item_id as ItemId,
    )
    .fetch_one(pool)
    .await?;

    Ok(count > 0)
}

/// Bulk variant of `has_purchased_item`. Returns the subset of `item_ids` that
/// the buyer has a completed purchase for. Single DB roundtrip vs. N calls.
#[tracing::instrument(skip_all)]
pub async fn purchased_subset(
    pool: &PgPool,
    user_id: UserId,
    item_ids: &[ItemId],
) -> Result<std::collections::HashSet<ItemId>> {
    if item_ids.is_empty() {
        return Ok(std::collections::HashSet::new());
    }
    let rows = sqlx::query_scalar!(
        r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
         WHERE buyer_id = $1 AND status = 'completed' AND item_id = ANY($2)"#,
        user_id as UserId,
        item_ids as &[ItemId],
    )
    .fetch_all(pool)
    .await?;
    Ok(rows.into_iter().collect())
}

/// Get all item IDs that a user has purchased (for batch access checks)
#[tracing::instrument(skip_all)]
pub async fn get_user_purchased_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
    let item_ids: Vec<ItemId> = sqlx::query_scalar!(
        r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions WHERE buyer_id = $1 AND status = 'completed' AND item_id IS NOT NULL"#,
        user_id as UserId,
    )
    .fetch_all(pool)
    .await?;

    Ok(item_ids)
}

/// Claims a free item by creating a zero-cost completed transaction.
/// Returns true if claimed successfully, false if already in library.
///
/// Uses `ON CONFLICT DO NOTHING` against the partial unique index on
/// `(buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL` to prevent duplicate
/// claims under concurrent requests.
#[tracing::instrument(skip_all)]
pub async fn claim_free_item<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: &ClaimParams<'_>,
) -> Result<bool> {
    let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
    let result = sqlx::query!(
        r#"
        INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, status, completed_at, item_title, seller_username, share_contact, parent_transaction_id, platform_credit_cents)
        VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8, $9)
        ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
        "#,
        params.buyer_id as UserId,
        params.seller_id as UserId,
        params.item_id as ItemId,
        claim_id,
        params.item_title,
        params.seller_username,
        params.share_contact,
        params.parent_transaction_id as Option<TransactionId>,
        params.platform_credit_cents,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected() > 0)
}

/// Batch variant of [`claim_free_item`] for bundle grants: claims every child
/// item for one buyer in a single INSERT instead of N round-trips (each with its
/// own pool acquire) on the Stripe webhook / checkout hot path (Perf-S4, Run 9).
/// Each row is idempotent via the same partial-unique ON CONFLICT as the single
/// claim, and child items deliberately do not increment `sales_count`. Returns the
/// number of rows actually inserted. No-op on an empty slice.
#[tracing::instrument(skip_all)]
pub async fn claim_free_items_batch<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    buyer_id: UserId,
    seller_id: UserId,
    seller_username: &str,
    parent_transaction_id: Option<TransactionId>,
    items: &[(ItemId, &str)],
) -> Result<u64> {
    if items.is_empty() {
        return Ok(0);
    }
    let item_ids: Vec<ItemId> = items.iter().map(|(id, _)| *id).collect();
    let item_titles: Vec<&str> = items.iter().map(|(_, title)| *title).collect();

    // The per-row claim id mirrors the single claim's `free-claim-{buyer}-{item}`
    // so a later single claim of the same item still collides idempotently.
    let result = sqlx::query(
        r"
        INSERT INTO transactions
            (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents,
             stripe_checkout_session_id, status, completed_at, item_title,
             seller_username, share_contact, parent_transaction_id)
        SELECT
            $1, $2, t.item_id, 0, 0,
            'free-claim-' || $1::text || '-' || t.item_id::text,
            'completed', NOW(), t.item_title, $3, false, $4
        FROM UNNEST($5::uuid[], $6::text[]) AS t(item_id, item_title)
        ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
        ",
    )
    .bind(buyer_id)
    .bind(seller_id)
    .bind(seller_username)
    .bind(parent_transaction_id)
    .bind(&item_ids)
    .bind(&item_titles)
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// Optional parameters for generating a license key inside a claim transaction.
pub struct LicenseKeyParams<'a> {
    pub key_code: &'a KeyCode,
    pub max_activations: Option<i32>,
}

/// Atomically claim a free item and increment the promo code's use count.
///
/// Claims the item FIRST (INSERT transaction), then increments use_count.
/// If the user already owns the item (rows_affected == 0), rolls back without
/// consuming the code. If the code limit is reached, rolls back the claim too.
///
/// When `license_key_params` is `Some`, a license key is created inside the
/// same transaction so that the claim and key are always consistent.
///
/// Returns `(code_accepted, item_claimed)`:
/// - `code_accepted = false` → promo code hit its usage limit (nothing changed)
/// - `item_claimed = false` → user already owns the item (code was NOT consumed)
#[tracing::instrument(skip_all)]
pub async fn claim_free_with_promo_code(
    pool: &PgPool,
    promo_code_id: PromoCodeId,
    params: &ClaimParams<'_>,
    license_key_params: Option<&LicenseKeyParams<'_>>,
) -> Result<(bool, bool)> {
    let mut tx = pool.begin().await?;

    // Step 1: Attempt to claim the item first. When a platform-wide credit (Fan+)
    // made the item free, `platform_credit_cents` carries the item's full price so
    // the scheduler reimburses the creator via transfer, the fan pays nothing but
    // the creator is still paid (MNW funds it).
    let claim_id = format!("free-claim-{}-{}", params.buyer_id, params.item_id);
    let result = sqlx::query!(
        r#"
        INSERT INTO transactions (buyer_id, seller_id, item_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, status, completed_at, item_title, seller_username, share_contact, promo_code_id, platform_credit_cents)
        VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, $7, $8, $9)
        ON CONFLICT (buyer_id, item_id) WHERE status = 'completed' AND item_id IS NOT NULL DO NOTHING
        "#,
        params.buyer_id as UserId,
        params.seller_id as UserId,
        params.item_id as ItemId,
        claim_id,
        params.item_title,
        params.seller_username,
        params.share_contact,
        promo_code_id as PromoCodeId,
        params.platform_credit_cents,
    )
    .execute(&mut *tx)
    .await?;

    let claimed = result.rows_affected() > 0;

    // Step 2: If the user already owns the item, rollback without consuming the code
    if !claimed {
        tx.rollback().await?;
        return Ok((true, false));
    }

    // Step 3: Increment the promo code use count. Re-check the full validity
    // window (max_uses AND starts_at/expires_at) inside the atomic UPDATE, the
    // pre-flight check ran outside this transaction, so a code expiring in the
    // sub-ms gap must still be rejected here, not just on use-count (Run 11 Pay
    // MINOR / TOCTOU).
    let code_result = sqlx::query!(
        r#"
        UPDATE promo_codes SET use_count = use_count + 1
        WHERE id = $1
          AND (max_uses IS NULL OR use_count < max_uses)
          AND (starts_at IS NULL OR starts_at <= NOW())
          AND (expires_at IS NULL OR expires_at > NOW())
        "#,
        promo_code_id as PromoCodeId,
    )
    .execute(&mut *tx)
    .await?;

    // Step 4: If the code limit was reached, rollback the claim too
    if code_result.rows_affected() == 0 {
        tx.rollback().await?;
        return Ok((false, false));
    }

    crate::db::items::increment_sales_count(&mut *tx, params.item_id).await?;

    // Step 5: Create license key inside the same transaction if requested.
    // Retry once on a unique-violation: the wordlist generator has ~6B-coin-
    // flip headroom, so an actual collision is vanishingly rare, but the
    // alternative is surfacing a 500 to a buyer mid-claim, cheap to handle.
    if let Some(lk) = license_key_params {
        let attempt = sqlx::query!(
            r#"
            INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
            VALUES ($1, $2, NULL, $3, $4)
            "#,
            params.item_id as ItemId,
            params.buyer_id as UserId,
            lk.key_code as &KeyCode,
            lk.max_activations,
        )
        .execute(&mut *tx)
        .await;

        if let Err(sqlx::Error::Database(e)) = &attempt
            && e.code().as_deref() == Some("23505")
        {
            let retry_code = crate::helpers::generate_key_code();
            tracing::warn!(item_id = %params.item_id, "license key 23505 collision; retrying once");
            sqlx::query!(
                r#"
                INSERT INTO license_keys (item_id, owner_id, transaction_id, key_code, max_activations)
                VALUES ($1, $2, NULL, $3, $4)
                "#,
                params.item_id as ItemId,
                params.buyer_id as UserId,
                retry_code as KeyCode,
                lk.max_activations,
            )
            .execute(&mut *tx)
            .await?;
        } else {
            attempt?;
        }
    }

    tx.commit().await?;
    Ok((true, true))
}

// ── Project purchases ──

/// Check whether a user has a completed purchase for a given project.
#[tracing::instrument(skip_all)]
pub async fn has_purchased_project(
    pool: &PgPool,
    user_id: UserId,
    project_id: ProjectId,
) -> Result<bool> {
    let count: i64 = sqlx::query_scalar!(
        r#"SELECT COUNT(*) AS "count!" FROM transactions WHERE buyer_id = $1 AND project_id = $2 AND status = 'completed'"#,
        user_id as UserId,
        project_id as ProjectId,
    )
    .fetch_one(pool)
    .await?;

    Ok(count > 0)
}

/// Parameters for creating a pending project purchase transaction.
pub struct CreateProjectTransactionParams<'a> {
    pub buyer_id: UserId,
    pub seller_id: UserId,
    pub project_id: ProjectId,
    pub amount_cents: i32,
    pub stripe_checkout_session_id: &'a str,
    pub project_title: &'a str,
    pub seller_username: &'a str,
    pub share_contact: bool,
}

/// Record a new pending transaction for a project purchase.
#[tracing::instrument(skip_all)]
pub async fn create_project_transaction(
    pool: &PgPool,
    params: &CreateProjectTransactionParams<'_>,
) -> Result<DbTransaction> {
    let tx = sqlx::query_as!(
        DbTransaction,
        r#"
        INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents, stripe_checkout_session_id, item_title, seller_username, share_contact)
        VALUES ($1, $2, $3, $4, 0, $5, $6, $7, $8)
        RETURNING
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        "#,
        params.buyer_id as UserId,
        params.seller_id as UserId,
        params.project_id as ProjectId,
        params.amount_cents,
        params.stripe_checkout_session_id,
        params.project_title,
        params.seller_username,
        params.share_contact,
    )
    .fetch_one(pool)
    .await?;

    Ok(tx)
}

/// Get items purchased by a user, including any associated license key.
///
/// Reads from the `purchases` VIEW (which filters `transactions` to
/// `status = 'completed'`), then JOINs through `items → projects → users`
/// for display fields. The LEFT JOIN on `license_keys` attaches the most
/// recent non-revoked key code so the buyer can see it in their library
/// without a separate lookup. Capped at 20 rows for the dashboard summary.
#[tracing::instrument(skip_all)]
pub async fn get_user_purchases(pool: &PgPool, user_id: UserId) -> Result<Vec<DbPurchaseRow>> {
    let purchases = sqlx::query_as!(
        DbPurchaseRow,
        r#"
        SELECT
            transaction_id AS "transaction_id!: TransactionId",
            item_id AS "item_id!: ItemId",
            title AS "title!",
            creator AS "creator!",
            item_type AS "item_type!: crate::db::ItemType",
            purchased_at AS "purchased_at!: chrono::DateTime<chrono::Utc>",
            is_free AS "is_free!",
            license_key_code AS "license_key_code?: KeyCode",
            has_new_version AS "has_new_version!"
        FROM (
            SELECT DISTINCT ON (p.item_id)
                p.transaction_id,
                p.item_id,
                i.title,
                u.username as creator,
                i.item_type,
                p.purchased_at,
                -- Badge from what the buyer actually paid, not the item's current
                -- price: a later re-price to $0 must not retroactively badge a paid
                -- purchase "Free" (nor vice-versa). The purchases view carries the
                -- transaction's own amount_cents.
                (p.amount_cents = 0) as is_free,
                lk.key_code as license_key_code,
                (vc.total_versions > 0 AND vc.total_versions > COALESCE(dc.downloaded_count, 0)) as has_new_version
            FROM purchases p
            JOIN items i ON p.item_id = i.id
            JOIN projects proj ON i.project_id = proj.id
            JOIN users u ON proj.user_id = u.id
            LEFT JOIN license_keys lk ON lk.item_id = p.item_id AND lk.owner_id = p.buyer_id AND lk.revoked_at IS NULL
            LEFT JOIN LATERAL (
                SELECT COUNT(*) AS total_versions
                FROM versions v
                WHERE v.item_id = i.id AND v.s3_key IS NOT NULL
            ) vc ON true
            LEFT JOIN LATERAL (
                SELECT COUNT(*) AS downloaded_count
                FROM user_downloads ud
                WHERE ud.user_id = p.buyer_id AND ud.item_id = i.id
            ) dc ON true
            WHERE p.buyer_id = $1
            ORDER BY p.item_id, p.purchased_at DESC
        ) deduped
        ORDER BY purchased_at DESC
        LIMIT 20
        "#,
        user_id as UserId,
    )
    .fetch_all(pool)
    .await?;

    Ok(purchases)
}

/// Remove a free item from library (deletes the claim transaction).
/// If the claim was via a promo code, decrements the code's use_count.
#[tracing::instrument(skip_all)]
pub async fn remove_free_item_from_library(
    pool: &PgPool,
    user_id: UserId,
    item_id: ItemId,
) -> Result<bool> {
    // Delete the free claim and return the promo_code_id if one was used
    let row: Option<Option<crate::db::PromoCodeId>> = sqlx::query_scalar!(
        r#"
        DELETE FROM transactions
        WHERE buyer_id = $1 AND item_id = $2 AND amount_cents = 0 AND status = 'completed'
        RETURNING promo_code_id AS "promo_code_id: crate::db::PromoCodeId"
        "#,
        user_id as UserId,
        item_id as ItemId,
    )
    .fetch_optional(pool)
    .await?;

    let deleted = row.is_some();

    if let Some(Some(pc_id)) = row {
        crate::db::promo_codes::release_use_count(pool, pc_id)
            .await
            .ok();
    }

    Ok(deleted)
}

/// Fetch a single transaction by ID.
///
/// # Authorization
///
/// This lookup is intentionally **unscoped**, it does not filter by buyer or
/// seller, because the two callers need the row to *decide* authorization
/// (a receipt page shown to buyer-or-seller; a refund restricted to the
/// seller). Every caller MUST therefore check ownership against the returned
/// `buyer_id`/`seller_id` before acting on it. A new caller that returns this
/// row's contents without such a check would be an IDOR, there is no implicit
/// scoping here to lean on.
#[tracing::instrument(skip_all)]
pub async fn get_transaction_by_id(
    pool: &PgPool,
    id: TransactionId,
) -> Result<Option<DbTransaction>> {
    let tx = sqlx::query_as!(
        DbTransaction,
        r#"
        SELECT
            id AS "id: TransactionId", buyer_id AS "buyer_id: UserId", seller_id AS "seller_id: UserId",
            item_id AS "item_id: ItemId", amount_cents AS "amount_cents: Cents", platform_fee_cents AS "platform_fee_cents: Cents",
            currency, status AS "status: crate::db::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
            created_at AS "created_at: chrono::DateTime<chrono::Utc>", completed_at AS "completed_at: chrono::DateTime<chrono::Utc>",
            item_title, seller_username, share_contact, project_id AS "project_id: ProjectId",
            parent_transaction_id AS "parent_transaction_id: TransactionId", promo_code_id AS "promo_code_id: PromoCodeId",
            guest_email, claim_token AS "claim_token: ClaimToken", claimed_by AS "claimed_by: UserId",
            download_token AS "download_token: DownloadToken"
        FROM transactions WHERE id = $1
        "#,
        id as TransactionId,
    )
    .fetch_optional(pool)
    .await?;
    Ok(tx)
}

/// Atomically claim a completed transaction for refund (`completed -> refunding`).
///
/// Returns `Some(id)` only if THIS call won the transition; returns `None` if the
/// row was not `completed` (already refunding, already refunded, or gone). The
/// self-service refund handler must call this BEFORE issuing the Stripe refund so
/// a rapid double-submit cannot pass the refundability check twice and over-refund
/// a shared-cart PaymentIntent (Pay-S1, Run 9). On Stripe error the handler calls
/// [`release_refund_claim`] to roll the row back to `completed`; on success the
/// `refund.created` webhook finalizes `refunding -> refunded`.
#[tracing::instrument(skip_all)]
pub async fn claim_transaction_for_refund(
    pool: &PgPool,
    id: TransactionId,
) -> Result<Option<TransactionId>> {
    let row = sqlx::query_scalar!(
        r#"
        UPDATE transactions
        SET status = 'refunding'
        WHERE id = $1 AND status = 'completed'
        RETURNING id AS "id: TransactionId"
        "#,
        id as TransactionId,
    )
    .fetch_optional(pool)
    .await?;

    Ok(row)
}

/// Release a refund claim (`refunding -> completed`) after a Stripe refund call
/// failed, so the creator can retry. Idempotent: only a row still in `refunding`
/// transitions; a row the webhook already finalized to `refunded` is left alone.
#[tracing::instrument(skip_all)]
pub async fn release_refund_claim(pool: &PgPool, id: TransactionId) -> Result<()> {
    sqlx::query!(
        r#"
        UPDATE transactions
        SET status = 'completed'
        WHERE id = $1 AND status = 'refunding'
        "#,
        id as TransactionId,
    )
    .execute(pool)
    .await?;

    Ok(())
}

/// Mark a transaction as refunded, returning its ID and item_id for downstream cleanup.
///
/// The WHERE clause requires `status IN ('completed', 'refunding')` so that
/// already-refunded or pending transactions are not double-processed, while a row
/// the self-service handler has claimed (`refunding`) still finalizes. Returns an
/// empty vec if no matching transactions were found (idempotent for webhook retries).
///
/// Returns ALL refunded transactions (handles cart checkouts where multiple
/// transactions share the same payment_intent_id).
///
/// FULL-INTENT scope, and `pub(crate)` so only in-crate webhook handlers can
/// mint it: a single cart line must use the line-scoped
/// [`refund_transaction_by_id`] instead, never this PI-wide UPDATE (the Run #2
/// Payments SERIOUS that refunded a whole cart from one line's event).
#[tracing::instrument(skip_all)]
pub(crate) async fn refund_transaction_by_payment_intent<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    payment_intent_id: &str,
) -> Result<Vec<(crate::db::TransactionId, Option<ItemId>)>> {
    // item_id is nullable on project-level transactions (routes/stripe/checkout/project.rs);
    // returning non-Optional ItemId would cause sqlx decode failures and infinite Stripe retries.
    let rows = sqlx::query!(
        r#"
        UPDATE transactions
        SET status = 'refunded'
        WHERE stripe_payment_intent_id = $1 AND status IN ('completed', 'refunding')
        RETURNING id AS "id: crate::db::TransactionId", item_id AS "item_id: ItemId"
        "#,
        payment_intent_id,
    )
    .fetch_all(executor)
    .await?;

    Ok(rows.into_iter().map(|r| (r.id, r.item_id)).collect())
}

/// Mark a SINGLE transaction refunded by id, returning `(id, item_id)` if it
/// transitioned from `completed` or `refunding` (the self-service handler claims
/// the row to `refunding` before calling Stripe). Returns `None` if it was already
/// refunded or otherwise not refundable (idempotent for webhook re-delivery).
///
/// Used by the line-scoped `refund.created` handler: cart lines share a
/// payment_intent, so refunding one line must touch only its own row, never the
/// PI-wide [`refund_transaction_by_payment_intent`] (Run #2 Payments SERIOUS).
#[tracing::instrument(skip_all)]
pub(crate) async fn refund_transaction_by_id<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    id: TransactionId,
) -> Result<Option<(crate::db::TransactionId, Option<ItemId>)>> {
    let row = sqlx::query!(
        r#"
        UPDATE transactions
        SET status = 'refunded'
        WHERE id = $1 AND status IN ('completed', 'refunding')
        RETURNING id AS "id: crate::db::TransactionId", item_id AS "item_id: ItemId"
        "#,
        id as TransactionId,
    )
    .fetch_optional(executor)
    .await?;

    Ok(row.map(|r| (r.id, r.item_id)))
}

/// True if any transaction (any status) references this payment_intent. Lets the
/// `charge.refunded` handler tell "already refunded" (line-scoped refunds marked
/// the rows) apart from "genuinely unmatched" before queuing a pending refund.
pub async fn transaction_exists_for_payment_intent<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    payment_intent_id: &str,
) -> Result<bool> {
    let exists = sqlx::query_scalar!(
        r#"SELECT EXISTS(SELECT 1 FROM transactions WHERE stripe_payment_intent_id = $1) AS "exists!""#,
        payment_intent_id,
    )
    .fetch_one(executor)
    .await?;

    Ok(exists)
}

/// True if any transaction (any status) references this checkout session. Lets
/// the cart-completion webhook tell a benign duplicate delivery (rows already
/// completed) apart from an ORPHANED paid session (rows never created, buyer
/// charged, got nothing) so the latter is escalated (Run #2 Payments SERIOUS).
pub async fn transaction_exists_for_checkout_session<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    checkout_session_id: &str,
) -> Result<bool> {
    let exists = sqlx::query_scalar!(
        r#"SELECT EXISTS(SELECT 1 FROM transactions WHERE stripe_checkout_session_id = $1) AS "exists!""#,
        checkout_session_id,
    )
    .fetch_one(executor)
    .await?;

    Ok(exists)
}

/// Revoke all child transactions linked to a parent (bundle) transaction.
///
/// Returns the item IDs of revoked children so callers can decrement sales counts.
#[tracing::instrument(skip_all)]
pub async fn revoke_child_transactions<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    parent_transaction_id: TransactionId,
) -> Result<Vec<ItemId>> {
    let item_ids = sqlx::query_scalar!(
        r#"
        UPDATE transactions
        SET status = 'refunded'
        WHERE parent_transaction_id = $1 AND status = 'completed'
        RETURNING item_id AS "item_id: ItemId"
        "#,
        parent_transaction_id as TransactionId,
    )
    .fetch_all(executor)
    .await?;

    Ok(item_ids.into_iter().flatten().collect())
}

/// Get seller transactions for CSV export, with conditional buyer email.
///
/// Respects contact revocations: if a buyer revoked sharing, their email
/// is hidden even if `share_contact` was true on the transaction.
#[tracing::instrument(skip_all)]
/// One page of a seller's sales for CSV export, newest first.
///
/// Paginated (`LIMIT`/`OFFSET`) so the export streams in bounded batches instead
/// of loading the seller's entire transaction history into memory in one query
/// (ultra-fuzz Run 4 S1). The `(created_at, id)` ordering is stable so OFFSET
/// batches don't reorder. (Keyset pagination would avoid OFFSET's deep-scan cost
/// and is the future optimization; OFFSET is sufficient at current scale and
/// keeps peak memory + per-query result bounded, which is the DoS fix.)
pub async fn get_seller_transactions_for_export_page(
    pool: &PgPool,
    seller_id: UserId,
    limit: i64,
    offset: i64,
) -> Result<Vec<DbTransactionExportRow>> {
    let rows = sqlx::query_as!(
        DbTransactionExportRow,
        r#"
        SELECT
            t.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
            t.item_id AS "item_id: ItemId",
            t.item_title,
            t.amount_cents AS "amount_cents: Cents",
            t.status AS "status: crate::db::TransactionStatus",
            CASE WHEN t.share_contact AND NOT EXISTS (
                SELECT 1 FROM contact_revocations cr
                WHERE cr.buyer_id = t.buyer_id AND cr.seller_id = t.seller_id
            ) THEN u.email ELSE NULL END as buyer_email
        FROM transactions t
        LEFT JOIN users u ON u.id = t.buyer_id
        WHERE t.seller_id = $1
        ORDER BY t.created_at DESC, t.id DESC
        LIMIT $2 OFFSET $3
        "#,
        seller_id as UserId,
        limit,
        offset,
    )
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

/// All of a seller's export rows accumulated into one Vec, bounded to
/// `EXPORT_ACCUMULATE_CAP` rows. For admin / internal-API callers that need the
/// full set in memory; the public creator-facing export streams page-by-page via
/// [`get_seller_transactions_for_export_page`] instead of materializing here.
pub async fn get_seller_transactions_for_export(
    pool: &PgPool,
    seller_id: UserId,
) -> Result<Vec<DbTransactionExportRow>> {
    /// Page size for the accumulating fetch.
    const PAGE: i64 = 5_000;
    /// Cap so even an admin/internal export can't load an unbounded result set.
    const EXPORT_ACCUMULATE_CAP: usize = 1_000_000;

    let mut all = Vec::new();
    let mut offset = 0i64;
    loop {
        let page = get_seller_transactions_for_export_page(pool, seller_id, PAGE, offset).await?;
        let n = page.len();
        all.extend(page);
        offset += n as i64;
        if (n as i64) < PAGE || all.len() >= EXPORT_ACCUMULATE_CAP {
            break;
        }
    }
    Ok(all)
}

/// Create a pending "placeholder" transaction for a subscription checkout that
/// used a promo code. This row exists solely so `cleanup_stale_pending` can
/// release the promo code reservation if the buyer abandons the Stripe session.
/// It is deleted (not completed) when the subscription webhook fires.
#[tracing::instrument(skip_all)]
pub async fn create_subscription_pending_transaction(
    pool: &PgPool,
    buyer_id: UserId,
    seller_id: UserId,
    project_id: ProjectId,
    stripe_checkout_session_id: &str,
    promo_code_id: PromoCodeId,
) -> Result<()> {
    sqlx::query!(
        r#"
        INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents,
                                  stripe_checkout_session_id, item_title, seller_username, share_contact, promo_code_id)
        VALUES ($1, $2, $3, 0, 0, $4, 'subscription-promo-hold', '', false, $5)
        "#,
        buyer_id as UserId,
        seller_id as UserId,
        project_id as ProjectId,
        stripe_checkout_session_id,
        promo_code_id as PromoCodeId,
    )
    .execute(pool)
    .await?;

    Ok(())
}

/// Delete a pending subscription promo-hold transaction by checkout session ID.
/// Called from the subscription webhook after the subscription is created.
#[tracing::instrument(skip_all)]
pub async fn delete_subscription_pending_transaction<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    stripe_checkout_session_id: &str,
) -> Result<()> {
    sqlx::query!(
        "DELETE FROM transactions WHERE stripe_checkout_session_id = $1 AND status = 'pending'",
        stripe_checkout_session_id,
    )
    .execute(executor)
    .await?;

    Ok(())
}

/// Delete stale pending transactions (older than the given threshold) and return
/// the promo_code_ids that need their use_count decremented.
///
/// Stripe checkout sessions expire after 24 hours, so pending transactions older
/// than that will never complete. This releases the pending purchase uniqueness
/// slot and any reserved promo code use_count.
#[tracing::instrument(skip_all)]
pub async fn cleanup_stale_pending(
    pool: &PgPool,
    older_than: chrono::Duration,
) -> Result<Vec<Option<crate::db::PromoCodeId>>> {
    let cutoff = chrono::Utc::now() - older_than;
    // runtime-checked: binds a chrono DateTime<Utc> param; a bind param's type can't be overridden in the macro when sqlx time+chrono features are unified.
    let rows: Vec<(Option<crate::db::PromoCodeId>,)> = sqlx::query_as(
        r"
        DELETE FROM transactions
        WHERE status = 'pending'
          AND created_at < $1
        RETURNING promo_code_id
        ",
    )
    .bind(cutoff)
    .fetch_all(pool)
    .await?;

    Ok(rows.into_iter().map(|(id,)| id).collect())
}

/// Bulk variant of `get_pending_item_purchase`. Returns the subset of `item_ids`
/// for which the buyer already has a `pending` transaction. Used by cart
/// checkout to abort early when any line item would collide with the partial
/// unique index on `(buyer_id, item_id) WHERE status = 'pending'`.
#[tracing::instrument(skip_all)]
pub async fn pending_subset(
    pool: &PgPool,
    buyer_id: UserId,
    item_ids: &[ItemId],
) -> Result<std::collections::HashSet<ItemId>> {
    if item_ids.is_empty() {
        return Ok(std::collections::HashSet::new());
    }
    let rows = sqlx::query_scalar!(
        r#"SELECT DISTINCT item_id AS "item_id!: ItemId" FROM transactions
         WHERE buyer_id = $1 AND status = 'pending' AND item_id = ANY($2)"#,
        buyer_id as UserId,
        item_ids as &[ItemId],
    )
    .fetch_all(pool)
    .await?;
    Ok(rows.into_iter().collect())
}

/// Returns the buyer's pending transaction for a specific item, if any.
/// Used to surface in-progress checkouts on the purchase page.
#[tracing::instrument(skip_all)]
pub async fn get_pending_item_purchase(
    pool: &PgPool,
    buyer_id: UserId,
    item_id: ItemId,
) -> Result<Option<(TransactionId, chrono::DateTime<chrono::Utc>)>> {
    let row = sqlx::query!(
        r#"
        SELECT id AS "id: TransactionId", created_at AS "created_at: chrono::DateTime<chrono::Utc>"
        FROM transactions
        WHERE buyer_id = $1 AND item_id = $2 AND status = 'pending'
        LIMIT 1
        "#,
        buyer_id as UserId,
        item_id as ItemId,
    )
    .fetch_optional(pool)
    .await?;

    Ok(row.map(|r| (r.id, r.created_at)))
}

/// Delete the buyer's pending transaction for a specific item.
/// Returns any released `promo_code_id` so the caller can release its
/// reservation.
#[tracing::instrument(skip_all)]
pub async fn delete_pending_item_purchase(
    pool: &PgPool,
    buyer_id: UserId,
    item_id: ItemId,
) -> Result<Option<crate::db::PromoCodeId>> {
    let row: Option<Option<crate::db::PromoCodeId>> = sqlx::query_scalar!(
        r#"
        DELETE FROM transactions
        WHERE buyer_id = $1 AND item_id = $2 AND status = 'pending'
        RETURNING promo_code_id AS "promo_code_id: crate::db::PromoCodeId"
        "#,
        buyer_id as UserId,
        item_id as ItemId,
    )
    .fetch_optional(pool)
    .await?;

    Ok(row.flatten())
}

/// Create a completed free guest transaction.
///
/// Returns the number of rows inserted (0 if already claimed via ON CONFLICT).
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(skip_all)]
pub async fn create_free_guest_transaction(
    pool: &PgPool,
    buyer_id: Option<UserId>,
    seller_id: UserId,
    item_id: ItemId,
    checkout_session_id: &str,
    item_title: &str,
    seller_username: &str,
    guest_email: &str,
    claim_token: Option<ClaimToken>,
    download_token: DownloadToken,
) -> std::result::Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"
        INSERT INTO transactions (
            buyer_id, seller_id, item_id, amount_cents, platform_fee_cents,
            stripe_checkout_session_id, status, completed_at,
            item_title, seller_username, share_contact,
            guest_email, claim_token, download_token
        )
        VALUES ($1, $2, $3, 0, 0, $4, 'completed', NOW(), $5, $6, false, $7, $8, $9)
        ON CONFLICT (guest_email, item_id) WHERE status = 'completed' AND guest_email IS NOT NULL DO NOTHING
        "#,
        buyer_id as Option<UserId>,
        seller_id as UserId,
        item_id as ItemId,
        checkout_session_id,
        item_title,
        seller_username,
        guest_email,
        claim_token as Option<ClaimToken>,
        download_token as DownloadToken,
    )
    .execute(pool)
    .await?;

    Ok(result.rows_affected())
}

/// Record a free project claim (PWYW with $0 min or free project).
///
/// Returns `true` if the claim was actually inserted, `false` if the buyer
/// already owned the project. Mirrors the `claim_free_item` shape so callers
/// can gate downstream side-effects (contact-revocation clear, sale-notification
/// email, etc.) on the winner of a concurrent-claim race, without this signal,
/// two concurrent `/checkout/project` POSTs both fire those side-effects
/// regardless of which one's INSERT actually landed (Run #7 deferred SERIOUS).
#[tracing::instrument(skip_all)]
pub async fn claim_free_project(
    pool: &PgPool,
    buyer_id: UserId,
    seller_id: UserId,
    project_id: ProjectId,
    item_title: &str,
    seller_username: &str,
    share_contact: bool,
) -> Result<bool> {
    let result = sqlx::query!(
        r#"
        INSERT INTO transactions (buyer_id, seller_id, project_id, amount_cents, platform_fee_cents,
                                  status, completed_at, item_title, seller_username, share_contact)
        VALUES ($1, $2, $3, 0, 0, 'completed', NOW(), $4, $5, $6)
        ON CONFLICT (buyer_id, project_id) WHERE status = 'completed' AND project_id IS NOT NULL DO NOTHING
        "#,
        buyer_id as UserId,
        seller_id as UserId,
        project_id as ProjectId,
        item_title,
        seller_username,
        share_contact,
    )
    .execute(pool)
    .await?;

    Ok(result.rows_affected() > 0)
}