//! Cart checkout: multi-item purchase from one seller in a single Stripe session.

use axum::{
    Form,
    extract::State,
    response::{IntoResponse, Redirect, Response},
};
use serde::Deserialize;

use crate::{
    Billing, Integrations,
    auth::AuthUser,
    config::Config,
    db::{self, Cents, PromoCodeId, UserId},
    error::{AppError, Result, ResultExt},
    helpers,
    wam_client::WamClient,
};
use sqlx::PgPool;

use super::grant_bundle_items;

/// Release a promo reservation on a checkout-abort path, logging on failure.
///
/// These releases run only after another error has already aborted the
/// checkout, so the caller can't surface a failure to the user, but a silent
/// drop leaves the promo's use-count incremented (a stuck reservation). Log it
/// so an orphaned reservation is traceable rather than invisible.
async fn release_promo_quietly(db: &PgPool, pc_id: PromoCodeId, user_id: UserId) {
    if let Err(e) = db::promo_codes::release_use_count_and_detach(db, pc_id, user_id).await {
        tracing::warn!(
            promo_code_id = %pc_id,
            %user_id,
            error = %e,
            "failed to release promo reservation on checkout abort; use-count may be stuck"
        );
    }
}

/// Form data for cart checkout.
#[derive(Debug, Deserialize)]
pub(in crate::routes::stripe) struct CartCheckoutForm {
    pub seller_id: String,
    #[serde(default)]
    pub share_contact: bool,
    pub promo_code: Option<String>,
}

/// POST /stripe/checkout/cart - Checkout all cart items from one seller.
///
/// Thin wrapper over [`checkout_seller_cart`]: enforces the buyer-side
/// preconditions (suspended/sandbox/self-purchase) the chained path doesn't
/// need, then maps the core's `Option<url>` onto a redirect.
#[tracing::instrument(skip_all, name = "stripe::cart_checkout", fields(user_id = %user.id))]
pub(in crate::routes::stripe) async fn create_cart_checkout(
    State(db): State<PgPool>,
    State(integrations): State<Integrations>,
    State(payments): State<Billing>,
    State(config): State<Config>,
    AuthUser(user): AuthUser,
    Form(form): Form<CartCheckoutForm>,
) -> Result<Response> {
    user.check_not_suspended()?;
    user.check_not_sandbox()?;

    let seller_id: UserId = form
        .seller_id
        .parse()
        .map_err(|_| AppError::BadRequest("Invalid seller ID".to_string()))?;

    if user.id == seller_id {
        return Err(AppError::BadRequest(
            "You cannot purchase your own items".to_string(),
        ));
    }

    match checkout_seller_cart(
        &db,
        integrations.wam.as_ref(),
        &payments,
        &config,
        &user,
        seller_id,
        form.share_contact,
        form.promo_code.as_deref(),
    )
    .await?
    {
        Some(url) => Ok(Redirect::to(&url).into_response()),
        None => Ok(Redirect::to("/library?purchase=success").into_response()),
    }
}

/// Form data for checkout-all (cross-seller).
#[derive(Debug, Deserialize)]
pub(in crate::routes::stripe) struct CartCheckoutAllForm {
    #[serde(default)]
    pub share_contact: bool,
}

/// POST /stripe/checkout/cart/all - Checkout all cart items across all sellers.
///
/// Queues seller IDs in the session, processes the first seller, then chains
/// through the rest via checkout_success redirects.
#[tracing::instrument(skip_all, name = "stripe::cart_checkout_all", fields(user_id = %user.id))]
pub(in crate::routes::stripe) async fn create_cart_checkout_all(
    State(db): State<PgPool>,
    State(integrations): State<Integrations>,
    State(payments): State<Billing>,
    State(config): State<Config>,
    AuthUser(user): AuthUser,
    session: tower_sessions::Session,
    Form(form): Form<CartCheckoutAllForm>,
) -> Result<Response> {
    user.check_not_suspended()?;
    user.check_not_sandbox()?;

    let cart_items = db::cart::get_cart_items(&db, user.id)
        .await
        .context("fetch all cart items")?;

    if cart_items.is_empty() {
        return Ok(Redirect::to("/cart").into_response());
    }

    // Group by seller, collect unique seller IDs in order
    let mut seen = std::collections::HashSet::new();
    let mut seller_ids: Vec<String> = Vec::new();
    for item in &cart_items {
        let sid = item.seller_id.to_string();
        if seen.insert(sid.clone()) {
            seller_ids.push(sid);
        }
    }

    if seller_ids.is_empty() {
        return Ok(Redirect::to("/cart").into_response());
    }

    // Queue remaining sellers (all except the first) in session
    let first_seller = seller_ids.remove(0);
    if !seller_ids.is_empty() {
        session
            .insert("cart_queue", seller_ids)
            .await
            .map_err(|e| AppError::BadRequest(format!("session error: {e}")))?;
        session
            .insert("cart_share_contact", form.share_contact)
            .await
            .map_err(|e| AppError::BadRequest(format!("session error: {e}")))?;
    }

    // Process the first seller and chain through the queue until we hit a
    // paid seller (return its Stripe URL) or exhaust everything as free.
    match drain_to_paid(
        &db,
        integrations.wam.as_ref(),
        &payments,
        &config,
        &user,
        first_seller,
        form.share_contact,
        &session,
    )
    .await?
    {
        Some(url) => Ok(Redirect::to(&url).into_response()),
        None => Ok(Redirect::to("/library?purchase=success").into_response()),
    }
}

/// Claim a set of free (or discount-zeroed) cart items: insert the free
/// transaction, bump the sales count, and grant bundle items + a license key
/// when applicable, then bulk-remove the claimed rows from the cart.
///
/// Bundle/license fields come from `CartItem`, so this does no per-item
/// `get_item_by_id`, and the cart rows are removed in one bulk DELETE after the
/// loop (Run #8 perf MED). Shared by the free-by-price and discount-zeroed
/// passes so the claim logic exists in exactly one place.
async fn claim_free_cart_items(
    db: &PgPool,
    wam: Option<&WamClient>,
    user_id: UserId,
    seller_id: UserId,
    items: &[(&db::cart::CartItem, i64)],
    share_contact: bool,
) -> Result<()> {
    if items.is_empty() {
        return Ok(());
    }
    // Claim every free item in ONE transaction instead of a begin/commit per item
    // (Run 11 Perf MOD tail), collapses N round-trips to one and makes the cart
    // claim atomic (all free items or none). Post-commit side effects (bundle
    // grants, license-key minting) run after, only for the newly-claimed items.
    let mut to_remove: Vec<db::ItemId> = Vec::with_capacity(items.len());
    let mut claimed_items: Vec<&db::cart::CartItem> = Vec::with_capacity(items.len());

    let mut tx = db.begin().await.context("begin free-claim transaction")?;
    for (item, platform_credit_cents) in items {
        let claim = db::transactions::ClaimParams {
            buyer_id: user_id,
            item_id: item.item_id,
            seller_id,
            item_title: &item.title,
            seller_username: &item.creator_username,
            share_contact,
            parent_transaction_id: None,
            platform_credit_cents: *platform_credit_cents,
        };

        let claimed = db::transactions::claim_free_item(&mut *tx, &claim)
            .await
            .context("claim free item")?;
        if claimed {
            db::items::increment_sales_count(&mut *tx, item.item_id)
                .await
                .context("increment sales count")?;
            claimed_items.push(*item);
        }
        to_remove.push(item.item_id);
    }
    tx.commit().await.context("commit free-claim transaction")?;

    for item in claimed_items {
        if item.item_type == "bundle" {
            grant_bundle_items(db, item.item_id, user_id, seller_id, None).await;
        }
        if item.enable_license_keys {
            let key_code = helpers::generate_key_code();
            if let Err(e) = db::license_keys::create_license_key(
                db,
                item.item_id,
                user_id,
                None,
                &key_code,
                item.default_max_activations,
            )
            .await
            {
                // Mirror the paid path (webhook/checkout_helpers.rs): a buyer who
                // claimed the item but got no key is silent data loss, so escalate
                // to WAM for manual issuance rather than swallowing it (audit Run
                // 17 Observability). Free claims have no transaction id, so key the
                // ticket on the item.
                tracing::error!(user_id = %user_id, item_id = %item.item_id, error = ?e, "failed to generate license key for free claim");
                if let Some(wam) = wam {
                    let title =
                        format!("License key not issued (free claim): item {}", item.item_id);
                    let body = format!(
                        "User {user_id} claimed free item {} but license key generation \
                         failed: {e}\n\nManually issue a key.",
                        item.item_id,
                    );
                    wam.create_ticket(
                        &title,
                        Some(&body),
                        "critical",
                        "license-key-gen-failed",
                        Some(&item.item_id.to_string()),
                    )
                    .await;
                }
            }
        }
    }
    if let Err(e) = db::cart::remove_from_cart_bulk(db, user_id, &to_remove).await {
        // Non-fatal: the items were claimed; a failed cart cleanup just leaves
        // stale rows the user can remove. Log rather than drop silently.
        tracing::warn!(user_id = %user_id, error = ?e, "failed to clear claimed free items from cart");
    }
    Ok(())
}

/// Create the pending transactions for every paid item in one DB transaction,
/// so the buyer gets all items or none (no partial delivery on a mid-loop
/// failure).
///
/// A 23505 means another tab raced past the pre-check; abort the whole cart
/// rather than leave a paid Stripe line item with no pending row to fulfill. On
/// any error the promo reservation (if any) is released, since the Stripe
/// session was already created but no fulfilling rows landed.
async fn create_cart_pending_transactions(
    db: &PgPool,
    user_id: UserId,
    seller_id: UserId,
    session_id: &str,
    items: &[(&db::cart::CartItem, i32, i64)],
    share_contact: bool,
    promo_code_id: Option<PromoCodeId>,
) -> Result<()> {
    let mut db_tx = db
        .begin()
        .await
        .context("begin cart transaction creation")?;
    for (item, final_price, platform_credit_cents) in items {
        match db::transactions::create_transaction(
            &mut *db_tx,
            &db::transactions::CreateTransactionParams {
                buyer_id: Some(user_id),
                seller_id,
                item_id: Some(item.item_id),
                amount_cents: Cents::new(*final_price as i64),
                platform_fee_cents: Cents::ZERO,
                stripe_checkout_session_id: session_id,
                item_title: &item.title,
                seller_username: &item.creator_username,
                share_contact,
                project_id: None,
                promo_code_id,
                guest_email: None,
                platform_credit_cents: *platform_credit_cents,
            },
        )
        .await
        {
            Ok(_) => {}
            Err(AppError::Database(sqlx::Error::Database(ref db_err)))
                if db_err.code().as_deref() == Some("23505") =>
            {
                tracing::warn!(
                    buyer_id = %user_id, item_id = %item.item_id,
                    "23505 raced past pre-check during cart pending insert"
                );
                if let Some(pc_id) = promo_code_id {
                    release_promo_quietly(db, pc_id, user_id).await;
                }
                return Err(AppError::BadRequest(
                    "Another checkout for one of these items started while this one was loading. \
                     Please refresh and try again."
                        .to_string(),
                ));
            }
            Err(e) => {
                // Transaction auto-rolls back on drop.
                if let Some(pc_id) = promo_code_id {
                    release_promo_quietly(db, pc_id, user_id).await;
                }
                return Err(e).context("create pending transaction for cart item");
            }
        }
    }
    db_tx
        .commit()
        .await
        .context("commit cart pending transactions")?;
    Ok(())
}

/// Core per-seller cart checkout, shared by the single-seller form
/// ([`create_cart_checkout`]) and the cross-seller chain ([`drain_to_paid`]).
///
/// Returns `Ok(None)` when every item for this seller was free (claimed inline,
/// no Stripe session needed, the chain advances to the next seller), or
/// `Ok(Some(url))` with the Stripe Checkout URL for the paid remainder.
///
/// Ordering matters: the promo reservation is taken as late as possible, after
/// the Stripe-ready, minimum-charge, and pending-collision checks, so an abort
/// on any of those can't burn a single-use code. The previous chained copy
/// reserved early and leaked the reservation on the Stripe-ready and min-charge
/// rejects (inert only because the chain never passed a promo); folding the two
/// copies into one removes that divergence.
#[tracing::instrument(skip_all, name = "stripe::checkout_seller_cart", fields(user_id = %user.id, %seller_id))]
#[allow(clippy::too_many_arguments)]
async fn checkout_seller_cart(
    db: &PgPool,
    wam: Option<&WamClient>,
    payments: &Billing,
    config: &Config,
    user: &crate::auth::SessionUser,
    seller_id: UserId,
    share_contact: bool,
    promo_code: Option<&str>,
) -> Result<Option<String>> {
    let cart_items = db::cart::get_cart_items_for_seller(db, user.id, seller_id)
        .await
        .context("fetch cart items for seller")?;
    if cart_items.is_empty() {
        return Err(AppError::BadRequest(
            "No items in cart for this creator".to_string(),
        ));
    }

    let seller = db::users::get_user_by_id(db, seller_id)
        .await
        .context("fetch seller")?
        .ok_or(AppError::NotFound)?;
    if seller.is_suspended() || seller.is_deactivated() || seller.is_creator_paused() {
        return Err(AppError::BadRequest(
            "This creator's account is not active".to_string(),
        ));
    }

    // Bulk-check ownership in a single query instead of N sequential roundtrips.
    let cart_item_ids: Vec<db::ItemId> = cart_items.iter().map(|c| c.item_id).collect();
    let already_owned = db::transactions::purchased_subset(db, user.id, &cart_item_ids)
        .await
        .context("bulk check existing purchases")?;

    let mut free_items: Vec<&db::cart::CartItem> = Vec::new();
    let mut paid_items: Vec<&db::cart::CartItem> = Vec::new();
    for item in &cart_items {
        if already_owned.contains(&item.item_id) {
            if let Err(e) = db::cart::remove_from_cart(db, user.id, item.item_id).await {
                tracing::warn!(
                    user_id = %user.id, item_id = %item.item_id, error = ?e,
                    "failed to remove already-purchased item from cart; buyer will see it lingering on /cart"
                );
            }
            continue;
        }
        if item.is_free() {
            free_items.push(item);
        } else {
            paid_items.push(item);
        }
    }

    // Validate an optional promo code and compute per-item discounted prices plus
    // the platform credit MNW owes the seller when the code is a platform-wide
    // (Fan+) credit, so the creator is reimbursed the discount and still nets the
    // full price (same invariant as the single-item path; both destructure the
    // shared `AppliedDiscount`).
    let mut promo_code_id: Option<PromoCodeId> = None;
    let mut discounted_prices: std::collections::HashMap<db::ItemId, i32> =
        std::collections::HashMap::new();
    let mut platform_credits: std::collections::HashMap<db::ItemId, i64> =
        std::collections::HashMap::new();
    if let Some(code_str) = promo_code.map(str::trim).filter(|s| !s.is_empty())
        && let Some(validated) =
            db::promo_codes::lookup_and_validate_promo(db, seller_id, Some(user.id), code_str)
                .await?
    {
        use db::promo_codes::PromoApplication;
        // Apply to each eligible paid item; ineligible items (scope/min-price)
        // are skipped so the rest of the cart can still qualify.
        //
        // A platform-wide credit (the $5 Fan+ renewal credit) is a monetary
        // BALANCE spent at most once across the whole cart, NOT a per-line
        // coupon: without a budget it would discount the buyer and reimburse the
        // seller once per eligible line, an N-times payout from a single-use
        // credit (ultra-fuzz Run 13 Payments SERIOUS). `credit_budget` is the
        // code's face value (`None` for seller-funded/percentage codes, which
        // have no balance to over-spend and stay per-line); each line's credit is
        // capped to the remainder and the uncovered discount reverts to the buyer.
        let mut credit_budget = validated.platform_credit_budget_cents();
        for item in &paid_items {
            if item.pwyw_enabled {
                continue; // PWYW items can't take a promo (single-item behavior)
            }
            if let PromoApplication::Apply(applied) = db::promo_codes::apply_promo_to_item(
                &validated,
                item.item_id,
                item.project_id,
                item.effective_price_cents(),
            )? {
                let (final_price, credit) =
                    db::promo_codes::cap_line_to_credit_budget(applied, &mut credit_budget);
                discounted_prices.insert(item.item_id, final_price);
                if credit > 0 {
                    platform_credits.insert(item.item_id, credit);
                }
            }
        }
        promo_code_id = Some(validated.id());
    }

    // Re-classify after discount: some paid items may now be free. Each carries the
    // platform credit (0 for creator-funded discounts) so the seller is reimbursed
    // whether the item ends up paid-at-a-discount or free.
    let mut newly_free: Vec<(&db::cart::CartItem, i64)> = Vec::new();
    let mut still_paid: Vec<(&db::cart::CartItem, i32, i64)> = Vec::new();
    for item in &paid_items {
        let final_price = discounted_prices
            .get(&item.item_id)
            .copied()
            .unwrap_or_else(|| item.effective_price_cents());
        let credit = platform_credits.get(&item.item_id).copied().unwrap_or(0);
        if final_price == 0 {
            newly_free.push((item, credit));
        } else {
            still_paid.push((item, final_price, credit));
        }
    }
    let claimed_any_free = !free_items.is_empty() || !newly_free.is_empty();

    // When the whole cart is free after discounts, reserve the promo use BEFORE
    // claiming anything. An all-free cart still consumes exactly ONE use of the
    // code (Run 10 Pay S1: one checkout = one use), and a reached-limit code must
    // reject before any free item is granted, mirroring the paid path's
    // reserve-before-fulfil discipline at the Stripe branch below. Without this,
    // a max_uses-limited or 100%-off code would be redeemable unlimited times via
    // an all-free cart, which returns early past the paid-path reservation.
    if still_paid.is_empty()
        && let Some(pc_id) = promo_code_id
    {
        let reserved = db::promo_codes::try_increment_use_count(db, pc_id)
            .await
            .context("reserve promo code use at free cart checkout")?;
        if !reserved {
            return Err(AppError::BadRequest(
                "This promo code has reached its usage limit".to_string(),
            ));
        }
    }

    // Genuinely-free items carry no platform credit and no promo dependency, so
    // they're safe to grant now on either path.
    let free_with_credit: Vec<(&db::cart::CartItem, i64)> =
        free_items.iter().map(|it| (*it, 0i64)).collect();
    claim_free_cart_items(
        db,
        wam,
        user.id,
        seller_id,
        &free_with_credit,
        share_contact,
    )
    .await?;

    // No paid items remain after discounts: the single promo use is already
    // reserved above, so grant the promo-freed lines and finish.
    if still_paid.is_empty() {
        claim_free_cart_items(db, wam, user.id, seller_id, &newly_free, share_contact).await?;
        if share_contact && claimed_any_free {
            db::transactions::clear_contact_revocation(db, user.id, seller_id)
                .await
                .context("clear contact revocation")?;
        }
        return Ok(None);
    }

    // Mixed cart: the promo-freed lines (`newly_free`) are deliberately NOT
    // granted yet. They must wait until the single promo use is reserved below,
    // otherwise two concurrent single-use checkouts could both claim the freed
    // items before either reserves, and the reservation loser keeps them for free
    // (plus the platform-credit obligation on those lines) (Run 21 payments).

    // Verify Stripe is ready and the total clears the minimum BEFORE reserving
    // the promo, so neither reject burns a single-use code.
    let stripe_account_id = seller
        .stripe_account_id
        .as_deref()
        .ok_or_else(|| AppError::BadRequest("Creator hasn't set up payments yet".to_string()))?;
    if !seller.stripe_charges_enabled {
        return Err(AppError::BadRequest(
            "Creator's payment account is not ready".to_string(),
        ));
    }
    let stripe = payments
        .stripe
        .as_ref()
        .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;

    let line_items: Vec<crate::payments::CartLineItem> = still_paid
        .iter()
        .map(
            |(item, final_price, _credit)| crate::payments::CartLineItem {
                title: &item.title,
                amount_cents: *final_price as i64,
            },
        )
        .collect();

    // Reject sub-Stripe-minimum totals before calling Stripe: chained promo+PWYW
    // combinations can land between 1¢ and 49¢, and Stripe's own error for that
    // is not user-friendly.
    let cart_total: i64 = line_items.iter().map(|li| li.amount_cents).sum();
    if cart_total > 0 && cart_total < crate::constants::STRIPE_MINIMUM_CHARGE_CENTS {
        return Err(AppError::BadRequest(format!(
            "Minimum cart total is {}",
            crate::formatting::format_revenue(crate::constants::STRIPE_MINIMUM_CHARGE_CENTS)
        )));
    }

    // Pre-check the partial unique index `(buyer_id, item_id) WHERE status='pending'`
    // BEFORE creating the Stripe session, so we never charge for an item that
    // can't get a pending row (and would therefore never be fulfilled).
    let paid_item_ids: Vec<db::ItemId> = still_paid.iter().map(|(it, _, _)| it.item_id).collect();
    let pending_collisions = db::transactions::pending_subset(db, user.id, &paid_item_ids)
        .await
        .context("pre-check pending cart purchases")?;
    if !pending_collisions.is_empty() {
        return Err(AppError::BadRequest(
            "You already have a checkout in progress for one or more of these items. \
             Complete or cancel that checkout before starting a new one."
                .to_string(),
        ));
    }

    // Reserve the promo use only now that every cheap reject is behind us.
    //
    // Semantics (ultra-fuzz Run 10 Pay S1, decided): one cart checkout consumes
    // exactly ONE use of the code, even when the code discounts multiple eligible
    // lines above. One redemption = one use, the same accounting as a
    // single-item checkout (one item-checkout = one use). Reserving per
    // discounted line would be a different product, not a bug fix; keep this a
    // single increment.
    if let Some(pc_id) = promo_code_id {
        let reserved = db::promo_codes::try_increment_use_count(db, pc_id)
            .await
            .context("reserve promo code use at cart checkout")?;
        if !reserved {
            return Err(AppError::BadRequest(
                "This promo code has reached its usage limit".to_string(),
            ));
        }
    }

    // Reservation succeeded (or there's no promo), now safe to grant the
    // promo-freed lines. Doing this AFTER the atomic reserve closes the
    // concurrent double-spend: a checkout that loses the single-use reservation
    // errored above and never reaches here.
    if !newly_free.is_empty() {
        claim_free_cart_items(db, wam, user.id, seller_id, &newly_free, share_contact).await?;
    }

    let success_url = format!(
        "{}/stripe/success?session_id={{CHECKOUT_SESSION_ID}}",
        config.host_url
    );
    let cancel_url = format!("{}/cart", config.host_url);
    let cart_params = crate::payments::CartCheckoutParams {
        connected_account_id: stripe_account_id,
        line_items: &line_items,
        buyer_id: user.id,
        seller_id,
        success_url: &success_url,
        cancel_url: &cancel_url,
        enable_stripe_tax: seller.stripe_tax_enabled,
    };

    let result = match stripe.create_cart_checkout_session(&cart_params).await {
        Ok(r) => r,
        Err(e) => {
            if let Some(pc_id) = promo_code_id {
                release_promo_quietly(db, pc_id, user.id).await;
            }
            return Err(e).context("create cart checkout session");
        }
    };

    if let Err(e) = create_cart_pending_transactions(
        db,
        user.id,
        seller_id,
        &result.id,
        &still_paid,
        share_contact,
        promo_code_id,
    )
    .await
    {
        // The Stripe session is already live and cannot be un-created here. If the
        // buyer pays it, the cart-completion webhook finds no pending rows and
        // escalates it as an orphaned paid session (Run #2 Payments SERIOUS).
        // Release the promo reservation so it isn't stuck held by a dead session.
        if let Some(pc_id) = promo_code_id {
            release_promo_quietly(db, pc_id, user.id).await;
        }
        tracing::error!(
            session_id = %result.id, error = ?e,
            "failed to create cart pending transactions after opening Stripe session; session is orphaned if paid"
        );
        return Err(e).context("create cart pending transactions");
    }

    // Cart items are removed by the webhook handler on successful payment, so a
    // canceled Stripe checkout leaves the cart intact.
    result
        .url
        .map(Some)
        .ok_or_else(|| AppError::BadRequest("No checkout URL returned".to_string()))
}

/// Process the cart queue starting with `first_seller_id`. Loops while
/// [`checkout_seller_cart`] returns `Ok(None)` (all items for that seller were
/// free), draining the session queue. Returns the Stripe checkout URL the
/// moment a paid seller is reached, or `None` when the queue is exhausted with
/// every item claimed free.
///
/// Chained checkout never carries a promo (`promo_code = None`); discounts are
/// only applied on direct single-seller form submissions.
#[tracing::instrument(skip_all, name = "stripe::drain_to_paid", fields(user_id = %user.id, first_seller_id = %first_seller_id))]
#[allow(clippy::too_many_arguments)]
pub(super) async fn drain_to_paid(
    db: &PgPool,
    wam: Option<&WamClient>,
    payments: &Billing,
    config: &Config,
    user: &crate::auth::SessionUser,
    first_seller_id: String,
    share_contact: bool,
    session: &tower_sessions::Session,
) -> Result<Option<String>> {
    let mut current = first_seller_id;
    loop {
        let seller_id: UserId = current
            .parse()
            .map_err(|_| AppError::BadRequest("Invalid seller ID".to_string()))?;
        if let Some(url) = checkout_seller_cart(
            db,
            wam,
            payments,
            config,
            user,
            seller_id,
            share_contact,
            None,
        )
        .await?
        {
            return Ok(Some(url));
        }
        // All items for `current` were free. Pop the next queued seller and
        // try again; on empty queue, signal "everything claimed".
        let next: Option<String> = match session.get::<Vec<String>>("cart_queue").await {
            Ok(Some(mut queue)) if !queue.is_empty() => {
                let n = queue.remove(0);
                if queue.is_empty() {
                    session.remove::<Vec<String>>("cart_queue").await.ok();
                    session.remove::<bool>("cart_share_contact").await.ok();
                } else {
                    session.insert("cart_queue", queue).await.ok();
                }
                Some(n)
            }
            _ => None,
        };
        match next {
            Some(n) => current = n,
            None => return Ok(None),
        }
    }
}