From 08000cc16ecdfce83526423f9c93673f12e91c43 Mon Sep 17 00:00:00 2001 From: 237899745 <237899745@users.noreply.git.workyai.cn> Date: Sun, 26 Jul 2026 03:07:30 +0800 Subject: [PATCH] fix: serialize Stripe checkout provisioning --- .env.example | 2 + docker/.env.production.example | 1 + docker/docker-compose.prod.yml | 1 + docs/api.md | 3 +- .../017_billing_checkout_invariants.sql | 59 ++ src/api/billing.rs | 920 ++++++++++++++---- src/config.rs | 6 + src/services/idempotency.rs | 10 - 8 files changed, 805 insertions(+), 197 deletions(-) create mode 100644 migrations/017_billing_checkout_invariants.sql diff --git a/.env.example b/.env.example index 9732e85..541bad7 100644 --- a/.env.example +++ b/.env.example @@ -40,6 +40,8 @@ STORAGE_PATH=./uploads BILLING_PROVIDER=stripe STRIPE_SECRET_KEY=sk_test_xxx STRIPE_WEBHOOK_SECRET=whsec_xxx +# Keep the official endpoint in production; override only for isolated mocks. +STRIPE_API_BASE_URL=https://api.stripe.com # 邮件服务（注册验证 + 密码重置） MAIL_ENABLED=false diff --git a/docker/.env.production.example b/docker/.env.production.example index 1328ffd..590eccf 100644 --- a/docker/.env.production.example +++ b/docker/.env.production.example @@ -48,6 +48,7 @@ MAIL_ENABLED=false MAIL_LOG_LINKS_WHEN_DISABLED=false # STRIPE_SECRET_KEY=sk_live_replace_me # STRIPE_WEBHOOK_SECRET=whsec_replace_me +# STRIPE_API_BASE_URL=https://api.stripe.com # MAIL_PROVIDER=custom # MAIL_FROM=noreply@example.com # MAIL_PASSWORD=replace-with-smtp-authorization-code diff --git a/docker/docker-compose.prod.yml b/docker/docker-compose.prod.yml index e57a102..d0c4606 100644 --- a/docker/docker-compose.prod.yml +++ b/docker/docker-compose.prod.yml @@ -74,6 +74,7 @@ services: ADMIN_PASSWORD: ${ADMIN_PASSWORD:-} STRIPE_SECRET_KEY: "${STRIPE_SECRET_KEY:-}" STRIPE_WEBHOOK_SECRET: "${STRIPE_WEBHOOK_SECRET:-}" + STRIPE_API_BASE_URL: ${STRIPE_API_BASE_URL:-https://api.stripe.com} MAIL_ENABLED: ${MAIL_ENABLED:-false} MAIL_LOG_LINKS_WHEN_DISABLED: ${MAIL_LOG_LINKS_WHEN_DISABLED:-false} MAIL_PROVIDER: ${MAIL_PROVIDER:-qq} diff --git a/docs/api.md b/docs/api.md index 0339185..ccea2ac 100644 --- a/docs/api.md +++ b/docs/api.md @@ -592,7 +592,6 @@ Authorization: Bearer POST /billing/checkout Authorization: Bearer Content-Type: application/json -Idempotency-Key: ``` 请求体： @@ -605,6 +604,8 @@ Idempotency-Key: { "success": true, "data": { "checkout_url": "https://pay.example.com/..." } } ``` +Checkout 的 Customer 与 Session 幂等键由服务端按用户和待支付记录生成，客户端无需也不能决定该幂等边界。同一用户同一时间只允许一个未过期的 Checkout；已有未取消 Stripe 订阅时返回 `409 IDEMPOTENCY_CONFLICT`，套餐调整必须使用 Portal。 + ### 9.5 打开客户 Portal（管理支付方式/取消订阅） ```http POST /billing/portal diff --git a/migrations/017_billing_checkout_invariants.sql b/migrations/017_billing_checkout_invariants.sql new file mode 100644 index 0000000..db64bee --- /dev/null +++ b/migrations/017_billing_checkout_invariants.sql @@ -0,0 +1,59 @@ +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM users + WHERE billing_customer_id IS NOT NULL AND billing_customer_id <> '' + GROUP BY billing_customer_id + HAVING COUNT(*) > 1 + ) THEN + RAISE EXCEPTION 'duplicate users.billing_customer_id values require Stripe reconciliation before migration 017'; + END IF; + + IF EXISTS ( + SELECT 1 + FROM subscriptions + WHERE provider = 'stripe' AND status <> 'canceled' + GROUP BY user_id + HAVING COUNT(*) > 1 + ) THEN + RAISE EXCEPTION 'multiple open Stripe subscriptions per user require reconciliation before migration 017'; + END IF; +END $$; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_users_billing_customer_unique + ON users(billing_customer_id) + WHERE billing_customer_id IS NOT NULL AND billing_customer_id <> ''; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_subscriptions_user_open_stripe_unique + ON subscriptions(user_id) + WHERE provider = 'stripe' AND status <> 'canceled'; + +CREATE TABLE IF NOT EXISTS billing_checkout_sessions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + plan_id UUID NOT NULL REFERENCES plans(id), + stripe_customer_id VARCHAR(200), + stripe_session_id VARCHAR(200), + checkout_url TEXT, + status VARCHAR(20) NOT NULL DEFAULT 'pending', + expires_at TIMESTAMPTZ NOT NULL, + lease_owner UUID, + lease_until TIMESTAMPTZ, + error_message TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + completed_at TIMESTAMPTZ, + CONSTRAINT billing_checkout_sessions_status_check + CHECK (status IN ('pending', 'completed', 'expired', 'failed', 'canceled')), + CONSTRAINT billing_checkout_sessions_stripe_session_unique + UNIQUE (stripe_session_id) +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_billing_checkout_sessions_user_pending + ON billing_checkout_sessions(user_id) + WHERE status = 'pending'; + +CREATE INDEX IF NOT EXISTS idx_billing_checkout_sessions_expiry + ON billing_checkout_sessions(expires_at) + WHERE status = 'pending'; diff --git a/src/api/billing.rs b/src/api/billing.rs index 09099fd..600e325 100644 --- a/src/api/billing.rs +++ b/src/api/billing.rs @@ -2,7 +2,6 @@ use crate::api::context; use crate::api::envelope::Envelope; use crate::error::{AppError, ErrorCode}; use crate::services::billing; -use crate::services::idempotency; use crate::services::quota; use crate::services::settings; use crate::state::AppState; @@ -11,7 +10,7 @@ use axum::extract::{ConnectInfo, State}; use axum::http::HeaderMap; use axum::routing::{get, post}; use axum::{Json, Router}; -use chrono::{DateTime, Utc}; +use chrono::{DateTime, Duration, TimeZone, Utc}; use serde::{Deserialize, Serialize}; use sqlx::FromRow; use std::net::SocketAddr; @@ -390,193 +389,396 @@ async fn create_checkout( _ => return Err(AppError::new(ErrorCode::Unauthorized, "未登录")), }; - let idempotency_key = headers - .get("idempotency-key") - .and_then(|v| v.to_str().ok()) - .map(str::trim) - .filter(|v| !v.is_empty()) - .map(str::to_string); + let checkout_url = create_checkout_for_user(&state, user_id, req.plan_id).await?; + Ok(Json(Envelope { + success: true, + data: CheckoutResponse { checkout_url }, + })) +} - let request_hash = idempotency_key.as_ref().map(|_| { - let plan = req.plan_id.to_string(); - idempotency::sha256_hex(&[b"billing_checkout", plan.as_bytes()]) - }); +#[derive(Debug, FromRow)] +struct PlanStripeRow { + stripe_price_id: Option, + amount_cents: i32, + is_active: bool, +} - let mut idem_acquired = false; - if let (Some(idem), Some(request_hash)) = (idempotency_key.as_deref(), request_hash.as_deref()) - { - match idempotency::begin( - &state, - idempotency::Scope::User(user_id), - idem, - request_hash, - state.config.idempotency_ttl_hours as i64, +#[derive(Debug, FromRow)] +struct UserStripeRow { + email: String, + billing_customer_id: Option, +} + +struct CheckoutProvision { + attempt_id: Uuid, + lease_owner: Uuid, + email: String, + customer_id: Option, + price_id: String, +} + +enum CheckoutClaim { + Ready(String), + Wait(Uuid), + Provision(CheckoutProvision), +} + +async fn create_checkout_for_user( + state: &AppState, + user_id: Uuid, + plan_id: Uuid, +) -> Result { + let plan = sqlx::query_as::<_, PlanStripeRow>( + "SELECT stripe_price_id, amount_cents, is_active FROM plans WHERE id = $1", + ) + .bind(plan_id) + .fetch_optional(&state.db) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询套餐失败").with_source(err))? + .ok_or_else(|| AppError::new(ErrorCode::NotFound, "套餐不存在"))?; + if !plan.is_active || plan.amount_cents <= 0 { + return Err(AppError::new(ErrorCode::Forbidden, "套餐不可用")); + } + let price_id = plan + .stripe_price_id + .filter(|value| !value.trim().is_empty()) + .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "该套餐不可订阅"))?; + + let lease_owner = Uuid::new_v4(); + let mut tx = state.db.begin().await.map_err(|err| { + AppError::new(ErrorCode::Internal, "开启 Checkout 事务失败").with_source(err) + })?; + let user = sqlx::query_as::<_, UserStripeRow>( + "SELECT email, billing_customer_id FROM users WHERE id = $1 FOR UPDATE", + ) + .bind(user_id) + .fetch_optional(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "锁定用户失败").with_source(err))? + .ok_or_else(|| AppError::new(ErrorCode::Unauthorized, "用户不存在"))?; + + let has_open_subscription: bool = sqlx::query_scalar( + r#" + SELECT EXISTS( + SELECT 1 FROM subscriptions + WHERE user_id = $1 AND provider = 'stripe' AND status <> 'canceled' ) - .await? - { - idempotency::BeginResult::Replay { response_body, .. } => { - let resp: CheckoutResponse = - serde_json::from_value(response_body).map_err(|err| { - AppError::new(ErrorCode::Internal, "幂等结果解析失败").with_source(err) - })?; - return Ok(Json(Envelope { - success: true, - data: resp, - })); - } - idempotency::BeginResult::InProgress => { - if let Some((_status, body)) = idempotency::wait_for_replay( - &state, - idempotency::Scope::User(user_id), - idem, - request_hash, - 10_000, - ) - .await? - { - let resp: CheckoutResponse = serde_json::from_value(body).map_err(|err| { - AppError::new(ErrorCode::Internal, "幂等结果解析失败").with_source(err) - })?; - return Ok(Json(Envelope { - success: true, - data: resp, - })); - } - return Err(AppError::new( - ErrorCode::InvalidRequest, - "请求正在处理中，请稍后重试", - )); - } - idempotency::BeginResult::Acquired => { - idem_acquired = true; - } - } + "#, + ) + .bind(user_id) + .fetch_one(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询订阅状态失败").with_source(err))?; + if has_open_subscription { + return Err(AppError::new( + ErrorCode::IdempotencyConflict, + "已有 Stripe 订阅，请通过账单门户升级、降级或续费", + )); } - let session_result: Result = (async { - let stripe_secret = settings::get_stripe_secret(&state) - .await - .map_err(|err| err.with_source("stripe secret not configured"))?; + sqlx::query( + "UPDATE billing_checkout_sessions SET status = 'expired', updated_at = NOW(), lease_owner = NULL, lease_until = NULL WHERE user_id = $1 AND status = 'pending' AND expires_at <= NOW()", + ) + .bind(user_id) + .execute(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "清理过期 Checkout 失败").with_source(err))?; - #[derive(Debug, FromRow)] - struct PlanStripeRow { - stripe_price_id: Option, - amount_cents: i32, - is_active: bool, + let existing: Option<(Uuid, Uuid, Option, bool)> = sqlx::query_as( + r#" + SELECT id, plan_id, checkout_url, COALESCE(lease_until > NOW(), false) + FROM billing_checkout_sessions + WHERE user_id = $1 AND status = 'pending' + FOR UPDATE + "#, + ) + .bind(user_id) + .fetch_optional(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "查询 Checkout 状态失败").with_source(err))?; + + let claim = if let Some((attempt_id, existing_plan_id, checkout_url, lease_valid)) = existing { + if existing_plan_id != plan_id { + return Err(AppError::new( + ErrorCode::IdempotencyConflict, + "已有其他套餐的支付会话，请先完成或等待其过期", + )); } - - let plan = sqlx::query_as::<_, PlanStripeRow>( + if let Some(checkout_url) = checkout_url { + CheckoutClaim::Ready(checkout_url) + } else if lease_valid { + CheckoutClaim::Wait(attempt_id) + } else { + sqlx::query( + "UPDATE billing_checkout_sessions SET lease_owner = $2, lease_until = NOW() + INTERVAL '30 seconds', error_message = NULL, updated_at = NOW() WHERE id = $1 AND status = 'pending'", + ) + .bind(attempt_id) + .bind(lease_owner) + .execute(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "重新领取 Checkout 失败").with_source(err))?; + CheckoutClaim::Provision(CheckoutProvision { + attempt_id, + lease_owner, + email: user.email, + customer_id: user.billing_customer_id, + price_id, + }) + } + } else { + let attempt_id = Uuid::new_v4(); + sqlx::query( r#" - SELECT stripe_price_id, amount_cents, is_active - FROM plans + INSERT INTO billing_checkout_sessions ( + id, user_id, plan_id, status, expires_at, lease_owner, lease_until + ) VALUES ( + $1, $2, $3, 'pending', NOW() + INTERVAL '30 minutes', + $4, NOW() + INTERVAL '30 seconds' + ) + "#, + ) + .bind(attempt_id) + .bind(user_id) + .bind(plan_id) + .bind(lease_owner) + .execute(&mut *tx) + .await + .map_err(|err| { + AppError::new(ErrorCode::Internal, "创建 Checkout 状态失败").with_source(err) + })?; + CheckoutClaim::Provision(CheckoutProvision { + attempt_id, + lease_owner, + email: user.email, + customer_id: user.billing_customer_id, + price_id, + }) + }; + tx.commit().await.map_err(|err| { + AppError::new(ErrorCode::Internal, "提交 Checkout 状态失败").with_source(err) + })?; + + match claim { + CheckoutClaim::Ready(url) => Ok(url), + CheckoutClaim::Wait(attempt_id) => wait_for_checkout_url(state, attempt_id).await, + CheckoutClaim::Provision(provision) => { + let attempt_id = provision.attempt_id; + let lease_owner = provision.lease_owner; + let result = provision_checkout(state, user_id, provision).await; + if let Err(err) = &result { + let _ = sqlx::query( + "UPDATE billing_checkout_sessions SET lease_owner = NULL, lease_until = NOW(), error_message = $4, updated_at = NOW() WHERE id = $1 AND user_id = $2 AND status = 'pending' AND lease_owner = $3", + ) + .bind(attempt_id) + .bind(user_id) + .bind(lease_owner) + .bind(err.to_string()) + .execute(&state.db) + .await; + } + result + } + } +} + +async fn wait_for_checkout_url(state: &AppState, attempt_id: Uuid) -> Result { + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + let row: Option<(String, Option, bool, Option)> = sqlx::query_as( + r#" + SELECT status, checkout_url, COALESCE(lease_until > NOW(), false), error_message + FROM billing_checkout_sessions WHERE id = $1 "#, ) - .bind(req.plan_id) + .bind(attempt_id) .fetch_optional(&state.db) .await - .map_err(|err| AppError::new(ErrorCode::Internal, "查询套餐失败").with_source(err))? - .ok_or_else(|| AppError::new(ErrorCode::NotFound, "套餐不存在"))?; - - if !plan.is_active { - return Err(AppError::new(ErrorCode::Forbidden, "套餐不可用")); - } - let Some(price_id) = plan.stripe_price_id.filter(|v| !v.trim().is_empty()) else { - return Err(AppError::new(ErrorCode::InvalidRequest, "该套餐不可订阅")); + .map_err(|err| AppError::new(ErrorCode::Internal, "等待 Checkout 失败").with_source(err))?; + let Some((status, checkout_url, lease_valid, error_message)) = row else { + return Err(AppError::new(ErrorCode::NotFound, "Checkout 状态不存在")); }; - if plan.amount_cents <= 0 { - return Err(AppError::new(ErrorCode::InvalidRequest, "该套餐不可订阅")); + if let Some(checkout_url) = checkout_url { + return Ok(checkout_url); } - - #[derive(Debug, FromRow)] - struct UserStripeRow { - email: String, - billing_customer_id: Option, + if status != "pending" { + return Err(AppError::new( + ErrorCode::IdempotencyConflict, + "Checkout 已结束，请重新发起", + )); } + if !lease_valid { + return Err(AppError::new( + ErrorCode::StorageUnavailable, + error_message.unwrap_or_else(|| "Checkout 创建未完成，请重试".to_string()), + )); + } + if tokio::time::Instant::now() >= deadline { + return Err(AppError::new( + ErrorCode::IdempotencyConflict, + "Checkout 正在创建，请稍后重试", + )); + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } +} - let user = sqlx::query_as::<_, UserStripeRow>( - "SELECT email, billing_customer_id FROM users WHERE id = $1", +async fn provision_checkout( + state: &AppState, + user_id: Uuid, + provision: CheckoutProvision, +) -> Result { + let stripe_secret = settings::get_stripe_secret(state) + .await + .map_err(|err| err.with_source("stripe secret not configured"))?; + let proposed_customer_id = match provision.customer_id { + Some(customer_id) if !customer_id.trim().is_empty() => customer_id, + _ => { + let idempotency_key = format!("imageforge-customer-{user_id}"); + stripe_create_customer( + state, + &stripe_secret, + &provision.email, + user_id, + &idempotency_key, + ) + .await? + } + }; + let customer_id = persist_customer_identity( + state, + user_id, + provision.attempt_id, + provision.lease_owner, + &proposed_customer_id, + ) + .await?; + + let success_url = format!( + "{}/dashboard/billing?checkout=success", + state.config.public_base_url + ); + let cancel_url = format!("{}/pricing?checkout=cancel", state.config.public_base_url); + let idempotency_key = format!("imageforge-checkout-{}", provision.attempt_id); + let session = stripe_create_checkout_session( + state, + &stripe_secret, + StripeCheckoutRequest { + customer_id: &customer_id, + price_id: &provision.price_id, + success_url: &success_url, + cancel_url: &cancel_url, + user_id, + attempt_id: provision.attempt_id, + idempotency_key: &idempotency_key, + }, + ) + .await?; + + let updated = sqlx::query( + r#" + UPDATE billing_checkout_sessions + SET stripe_customer_id = $4, + stripe_session_id = $5, + checkout_url = $6, + expires_at = $7, + lease_owner = NULL, + lease_until = NULL, + error_message = NULL, + updated_at = NOW() + WHERE id = $1 + AND user_id = $2 + AND status = 'pending' + AND lease_owner = $3 + "#, + ) + .bind(provision.attempt_id) + .bind(user_id) + .bind(provision.lease_owner) + .bind(&customer_id) + .bind(&session.id) + .bind(&session.url) + .bind(session.expires_at) + .execute(&state.db) + .await + .map_err(|err| { + AppError::new(ErrorCode::Internal, "保存 Stripe Checkout 失败").with_source(err) + })?; + if updated.rows_affected() != 1 { + return Err(AppError::new( + ErrorCode::IdempotencyConflict, + "Checkout 状态已变化，请重试", + )); + } + Ok(session.url) +} + +async fn persist_customer_identity( + state: &AppState, + user_id: Uuid, + attempt_id: Uuid, + lease_owner: Uuid, + proposed_customer_id: &str, +) -> Result { + let mut tx = state.db.begin().await.map_err(|err| { + AppError::new(ErrorCode::Internal, "开启 Customer 映射事务失败").with_source(err) + })?; + let current: Option> = + sqlx::query_scalar("SELECT billing_customer_id FROM users WHERE id = $1 FOR UPDATE") + .bind(user_id) + .fetch_optional(&mut *tx) + .await + .map_err(|err| { + AppError::new(ErrorCode::Internal, "锁定 Customer 映射失败").with_source(err) + })?; + let current = current.ok_or_else(|| AppError::new(ErrorCode::Unauthorized, "用户不存在"))?; + let customer_id = if let Some(current) = current.filter(|value| !value.trim().is_empty()) { + current + } else { + let updated: String = sqlx::query_scalar( + r#" + UPDATE users + SET billing_customer_id = $2, updated_at = NOW() + WHERE id = $1 + AND (billing_customer_id IS NULL OR billing_customer_id = '') + RETURNING billing_customer_id + "#, ) .bind(user_id) - .fetch_one(&state.db) + .bind(proposed_customer_id) + .fetch_one(&mut *tx) .await - .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?; - - let customer_id = if let Some(cus) = user.billing_customer_id { - cus - } else { - let cus = stripe_create_customer(&stripe_secret, &user.email, user_id).await?; - let _ = sqlx::query("UPDATE users SET billing_customer_id = $2 WHERE id = $1") - .bind(user_id) - .bind(&cus) - .execute(&state.db) - .await; - cus - }; - - let success_url = format!( - "{}/dashboard/billing?checkout=success", - state.config.public_base_url - ); - let cancel_url = format!("{}/pricing?checkout=cancel", state.config.public_base_url); - - stripe_create_checkout_session( - &stripe_secret, - &customer_id, - &price_id, - &success_url, - &cancel_url, - user_id, - ) - .await - }) - .await; - - match session_result { - Ok(session) => { - if let (Some(idem), Some(request_hash)) = - (idempotency_key.as_deref(), request_hash.as_deref()) - { - if idem_acquired { - let _ = idempotency::complete( - &state, - idempotency::Scope::User(user_id), - idem, - request_hash, - 200, - serde_json::to_value(&CheckoutResponse { - checkout_url: session.clone(), - }) - .unwrap_or(serde_json::Value::Null), - ) - .await; - } - } - - Ok(Json(Envelope { - success: true, - data: CheckoutResponse { - checkout_url: session, - }, - })) - } - Err(err) => { - if let (Some(idem), Some(request_hash)) = - (idempotency_key.as_deref(), request_hash.as_deref()) - { - if idem_acquired { - let _ = idempotency::abort( - &state, - idempotency::Scope::User(user_id), - idem, - request_hash, - ) - .await; - } - } - Err(err) - } + .map_err(|err| { + AppError::new(ErrorCode::Internal, "保存 Stripe Customer 失败").with_source(err) + })?; + updated + }; + let checkout_updated = sqlx::query( + r#" + UPDATE billing_checkout_sessions + SET stripe_customer_id = $4, updated_at = NOW() + WHERE id = $1 AND user_id = $2 AND status = 'pending' AND lease_owner = $3 + "#, + ) + .bind(attempt_id) + .bind(user_id) + .bind(lease_owner) + .bind(&customer_id) + .execute(&mut *tx) + .await + .map_err(|err| { + AppError::new(ErrorCode::Internal, "保存 Checkout Customer 失败").with_source(err) + })?; + if checkout_updated.rows_affected() != 1 { + return Err(AppError::new( + ErrorCode::IdempotencyConflict, + "Checkout 处理租约已失效", + )); } + tx.commit().await.map_err(|err| { + AppError::new(ErrorCode::Internal, "提交 Customer 映射失败").with_source(err) + })?; + Ok(customer_id) } #[derive(Debug, Serialize, Deserialize)] @@ -617,7 +819,8 @@ async fn create_portal( }; let return_url = format!("{}/dashboard/billing", state.config.public_base_url); - let url = stripe_create_portal_session(&stripe_secret, &customer_id, &return_url).await?; + let url = + stripe_create_portal_session(&state, &stripe_secret, &customer_id, &return_url).await?; Ok(Json(Envelope { success: true, @@ -626,17 +829,21 @@ async fn create_portal( } async fn stripe_create_customer( + state: &AppState, secret: &str, email: &str, user_id: Uuid, + idempotency_key: &str, ) -> Result { let resp: serde_json::Value = stripe_post_form( + state, secret, "/v1/customers", vec![ ("email".to_string(), email.to_string()), ("metadata[user_id]".to_string(), user_id.to_string()), ], + Some(idempotency_key), ) .await?; @@ -648,51 +855,100 @@ async fn stripe_create_customer( Ok(id.to_string()) } -async fn stripe_create_checkout_session( - secret: &str, - customer_id: &str, - price_id: &str, - success_url: &str, - cancel_url: &str, +struct StripeCheckoutSession { + id: String, + url: String, + expires_at: DateTime, +} + +struct StripeCheckoutRequest<'a> { + customer_id: &'a str, + price_id: &'a str, + success_url: &'a str, + cancel_url: &'a str, user_id: Uuid, -) -> Result { + attempt_id: Uuid, + idempotency_key: &'a str, +} + +async fn stripe_create_checkout_session( + state: &AppState, + secret: &str, + request: StripeCheckoutRequest<'_>, +) -> Result { let resp: serde_json::Value = stripe_post_form( + state, secret, "/v1/checkout/sessions", vec![ ("mode".to_string(), "subscription".to_string()), - ("customer".to_string(), customer_id.to_string()), - ("line_items[0][price]".to_string(), price_id.to_string()), + ("customer".to_string(), request.customer_id.to_string()), + ( + "line_items[0][price]".to_string(), + request.price_id.to_string(), + ), ("line_items[0][quantity]".to_string(), "1".to_string()), - ("success_url".to_string(), success_url.to_string()), - ("cancel_url".to_string(), cancel_url.to_string()), + ("success_url".to_string(), request.success_url.to_string()), + ("cancel_url".to_string(), request.cancel_url.to_string()), ("allow_promotion_codes".to_string(), "true".to_string()), - ("client_reference_id".to_string(), user_id.to_string()), - ("metadata[user_id]".to_string(), user_id.to_string()), + ( + "client_reference_id".to_string(), + request.user_id.to_string(), + ), + ("metadata[user_id]".to_string(), request.user_id.to_string()), + ( + "metadata[checkout_attempt_id]".to_string(), + request.attempt_id.to_string(), + ), + ( + "subscription_data[metadata][user_id]".to_string(), + request.user_id.to_string(), + ), + ( + "subscription_data[metadata][checkout_attempt_id]".to_string(), + request.attempt_id.to_string(), + ), ], + Some(request.idempotency_key), ) .await?; + let id = resp + .get("id") + .and_then(|v| v.as_str()) + .ok_or_else(|| AppError::new(ErrorCode::Internal, "Stripe checkout id 缺失"))?; let url = resp .get("url") .and_then(|v| v.as_str()) .ok_or_else(|| AppError::new(ErrorCode::Internal, "Stripe checkout 创建失败"))?; + let expires_at = resp + .get("expires_at") + .and_then(|value| value.as_i64()) + .and_then(|value| Utc.timestamp_opt(value, 0).single()) + .unwrap_or_else(|| Utc::now() + Duration::minutes(30)); - Ok(url.to_string()) + Ok(StripeCheckoutSession { + id: id.to_string(), + url: url.to_string(), + expires_at, + }) } async fn stripe_create_portal_session( + state: &AppState, secret: &str, customer_id: &str, return_url: &str, ) -> Result { let resp: serde_json::Value = stripe_post_form( + state, secret, "/v1/billing_portal/sessions", vec![ ("customer".to_string(), customer_id.to_string()), ("return_url".to_string(), return_url.to_string()), ], + None, ) .await?; @@ -705,17 +961,24 @@ async fn stripe_create_portal_session( } async fn stripe_post_form( + state: &AppState, secret: &str, path: &str, form: Vec<(String, String)>, + idempotency_key: Option<&str>, ) -> Result { - let url = format!("https://api.stripe.com{path}"); + let url = format!("{}{path}", state.config.stripe_api_base_url); let client = reqwest::Client::new(); - let resp = client + let mut request = client .post(url) .bearer_auth(secret) - .form(&form) + .timeout(std::time::Duration::from_secs(15)) + .form(&form); + if let Some(idempotency_key) = idempotency_key { + request = request.header("Idempotency-Key", idempotency_key); + } + let resp = request .send() .await .map_err(|err| AppError::new(ErrorCode::Internal, "Stripe 请求失败").with_source(err))?; @@ -733,3 +996,288 @@ async fn stripe_post_form( serde_json::from_str(&body) .map_err(|err| AppError::new(ErrorCode::Internal, "Stripe 响应解析失败").with_source(err)) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Config; + use crate::services::mail::Mailer; + use axum::extract::State; + use axum::http::HeaderMap; + use sqlx::postgres::PgPoolOptions; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + use tokio::sync::{Barrier, Semaphore}; + + #[derive(Clone, Default)] + struct CheckoutStripeMock { + customer_requests: Arc, + session_requests: Arc, + } + + async fn create_mock_customer( + State(mock): State, + headers: HeaderMap, + ) -> Json { + let key = headers + .get("Idempotency-Key") + .and_then(|value| value.to_str().ok()) + .expect("Customer request must have Idempotency-Key"); + assert!(key.starts_with("imageforge-customer-")); + let request_number = mock.customer_requests.fetch_add(1, Ordering::SeqCst) + 1; + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + Json(serde_json::json!({ + "id": format!("cus_mock_{request_number}") + })) + } + + async fn create_mock_session( + State(mock): State, + headers: HeaderMap, + ) -> Json { + let key = headers + .get("Idempotency-Key") + .and_then(|value| value.to_str().ok()) + .expect("Checkout request must have Idempotency-Key"); + assert!(key.starts_with("imageforge-checkout-")); + let request_number = mock.session_requests.fetch_add(1, Ordering::SeqCst) + 1; + Json(serde_json::json!({ + "id": format!("cs_mock_{request_number}"), + "url": format!("https://checkout.example.test/{request_number}"), + "expires_at": (Utc::now() + Duration::minutes(30)).timestamp() + })) + } + + async fn spawn_checkout_stripe_mock( + ) -> (String, CheckoutStripeMock, tokio::task::JoinHandle<()>) { + let mock = CheckoutStripeMock::default(); + let app = Router::new() + .route("/v1/customers", axum::routing::post(create_mock_customer)) + .route( + "/v1/checkout/sessions", + axum::routing::post(create_mock_session), + ) + .with_state(mock.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind checkout Stripe mock"); + let address = listener.local_addr().expect("read Stripe mock address"); + let task = tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("serve checkout Stripe mock"); + }); + (format!("http://{address}"), mock, task) + } + + async fn build_test_state( + pool: sqlx::PgPool, + database_url: String, + redis_url: String, + stripe_api_base_url: String, + ) -> AppState { + let mut config = Config::from_env().expect("load test config"); + config.database_url = database_url; + config.redis_url = redis_url; + config.stripe_secret_key = Some("sk_test_checkout".to_string()); + config.stripe_api_base_url = stripe_api_base_url; + config.mail_enabled = false; + config.mail_log_links_when_disabled = false; + let redis = redis::Client::open(config.redis_url.clone()) + .expect("create test Redis client") + .get_connection_manager() + .await + .expect("connect test Redis"); + AppState { + mailer: Arc::new(Mailer::new(&config).expect("create disabled test mailer")), + image_processing_semaphore: Arc::new(Semaphore::new(2)), + runtime_policy_cache: crate::services::settings::RuntimePolicyCache::new(), + storage_cache: crate::services::storage::StorageCache::new(), + config, + db: pool, + redis, + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[ignore = "requires isolated IMAGEFORGE_TEST_DATABASE_URL and IMAGEFORGE_TEST_REDIS_URL"] + async fn checkout_customer_and_subscription_invariants_are_serialized() { + let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL") + .expect("IMAGEFORGE_TEST_DATABASE_URL must be set"); + assert!( + database_url.to_ascii_lowercase().contains("test"), + "refusing to run destructive integration test outside a test database" + ); + let redis_url = std::env::var("IMAGEFORGE_TEST_REDIS_URL") + .expect("IMAGEFORGE_TEST_REDIS_URL must be set"); + let pool = PgPoolOptions::new() + .max_connections(16) + .connect(&database_url) + .await + .expect("connect test database"); + sqlx::migrate!().run(&pool).await.expect("run migrations"); + let (stripe_api_base_url, stripe_mock, stripe_mock_task) = + spawn_checkout_stripe_mock().await; + let state = + build_test_state(pool.clone(), database_url, redis_url, stripe_api_base_url).await; + + let marker = Uuid::new_v4().simple().to_string(); + let user_id = Uuid::new_v4(); + let failed_user_id = Uuid::new_v4(); + let plan_id = Uuid::new_v4(); + sqlx::query( + r#" + INSERT INTO plans ( + id, code, name, stripe_price_id, amount_cents, + included_units_per_period, max_file_size_mb, + max_files_per_batch, concurrency_limit, retention_days + ) VALUES ($1, $2, 'Checkout invariant test', $3, 100, + 10, 10, 10, 1, 1) + "#, + ) + .bind(plan_id) + .bind(format!("checkout_test_{marker}")) + .bind(format!("price_checkout_{marker}")) + .execute(&pool) + .await + .expect("insert checkout test plan"); + sqlx::query( + r#" + INSERT INTO users (id, email, username, password_hash) + VALUES ($1, $2, $3, 'test-only'), ($4, $5, $6, 'test-only') + "#, + ) + .bind(user_id) + .bind(format!("checkout-{marker}@example.test")) + .bind(format!("checkout_{marker}")) + .bind(failed_user_id) + .bind(format!("checkout-fail-{marker}@example.test")) + .bind(format!("checkout_fail_{marker}")) + .execute(&pool) + .await + .expect("insert checkout test users"); + + let barrier = Arc::new(Barrier::new(2)); + let mut joins = Vec::new(); + for _ in 0..2 { + let state = state.clone(); + let barrier = barrier.clone(); + joins.push(tokio::spawn(async move { + barrier.wait().await; + create_checkout_for_user(&state, user_id, plan_id).await + })); + } + let mut urls = Vec::new(); + for join in joins { + urls.push( + join.await + .expect("join concurrent checkout") + .expect("create concurrent checkout"), + ); + } + assert_eq!(urls[0], urls[1]); + assert_eq!(stripe_mock.customer_requests.load(Ordering::SeqCst), 1); + assert_eq!(stripe_mock.session_requests.load(Ordering::SeqCst), 1); + let persisted_customer: Option = + sqlx::query_scalar("SELECT billing_customer_id FROM users WHERE id = $1") + .bind(user_id) + .fetch_one(&pool) + .await + .expect("query persisted Customer"); + assert_eq!(persisted_customer.as_deref(), Some("cus_mock_1")); + + sqlx::query( + r#" + INSERT INTO subscriptions ( + user_id, plan_id, status, current_period_start, current_period_end, + provider, provider_customer_id, provider_subscription_id + ) VALUES ($1, $2, 'active', NOW(), NOW() + INTERVAL '30 days', + 'stripe', 'cus_mock_1', $3) + "#, + ) + .bind(user_id) + .bind(plan_id) + .bind(format!("sub_checkout_{marker}")) + .execute(&pool) + .await + .expect("insert active Stripe subscription"); + let active_error = create_checkout_for_user(&state, user_id, plan_id) + .await + .expect_err("active subscriber received another Checkout Session"); + assert_eq!(active_error.code, ErrorCode::IdempotencyConflict); + assert_eq!(stripe_mock.customer_requests.load(Ordering::SeqCst), 1); + assert_eq!(stripe_mock.session_requests.load(Ordering::SeqCst), 1); + + let function_name = format!("test_reject_customer_{marker}"); + let trigger_name = format!("test_reject_customer_trigger_{marker}"); + sqlx::query(&format!( + r#" + CREATE FUNCTION {function_name}() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.id = '{failed_user_id}'::uuid + AND NEW.billing_customer_id IS DISTINCT FROM OLD.billing_customer_id THEN + RAISE EXCEPTION 'injected billing customer persistence failure'; + END IF; + RETURN NEW; + END + $$ + "#, + )) + .execute(&pool) + .await + .expect("create Customer failure function"); + sqlx::query(&format!( + "CREATE TRIGGER {trigger_name} BEFORE UPDATE OF billing_customer_id ON users FOR EACH ROW EXECUTE FUNCTION {function_name}()" + )) + .execute(&pool) + .await + .expect("create Customer failure trigger"); + + create_checkout_for_user(&state, failed_user_id, plan_id) + .await + .expect_err("Checkout Session returned after Customer persistence failed"); + assert_eq!(stripe_mock.customer_requests.load(Ordering::SeqCst), 2); + assert_eq!( + stripe_mock.session_requests.load(Ordering::SeqCst), + 1, + "Stripe Session was created after Customer persistence failed" + ); + let failed_mapping: Option = + sqlx::query_scalar("SELECT billing_customer_id FROM users WHERE id = $1") + .bind(failed_user_id) + .fetch_one(&pool) + .await + .expect("query failed Customer mapping"); + assert!(failed_mapping.is_none()); + let failed_checkout: (Option, Option) = sqlx::query_as( + "SELECT checkout_url, lease_owner FROM billing_checkout_sessions WHERE user_id = $1 AND status = 'pending'", + ) + .bind(failed_user_id) + .fetch_one(&pool) + .await + .expect("query failed Checkout attempt"); + assert_eq!(failed_checkout, (None, None)); + + sqlx::query(&format!("DROP TRIGGER {trigger_name} ON users")) + .execute(&pool) + .await + .expect("drop Customer failure trigger"); + sqlx::query(&format!("DROP FUNCTION {function_name}()")) + .execute(&pool) + .await + .expect("drop Customer failure function"); + sqlx::query("DELETE FROM users WHERE id IN ($1, $2)") + .bind(user_id) + .bind(failed_user_id) + .execute(&pool) + .await + .expect("delete checkout test users"); + sqlx::query("DELETE FROM plans WHERE id = $1") + .bind(plan_id) + .execute(&pool) + .await + .expect("delete checkout test plan"); + stripe_mock_task.abort(); + } +} diff --git a/src/config.rs b/src/config.rs index b86bbec..72a15ca 100644 --- a/src/config.rs +++ b/src/config.rs @@ -25,6 +25,7 @@ pub struct Config { pub stripe_secret_key: Option, pub stripe_webhook_secret: Option, + pub stripe_api_base_url: String, pub storage_path: String, @@ -99,6 +100,10 @@ impl Config { } let stripe_secret_key = env_string("STRIPE_SECRET_KEY"); let stripe_webhook_secret = env_string("STRIPE_WEBHOOK_SECRET"); + let stripe_api_base_url = env_string("STRIPE_API_BASE_URL") + .unwrap_or_else(|| "https://api.stripe.com".to_string()) + .trim_end_matches('/') + .to_string(); let storage_path = env_string("STORAGE_PATH").unwrap_or_else(|| "./uploads".to_string()); @@ -140,6 +145,7 @@ impl Config { api_key_pepper, stripe_secret_key, stripe_webhook_secret, + stripe_api_base_url, storage_path, allow_anonymous_upload, anon_max_file_size_mb, diff --git a/src/services/idempotency.rs b/src/services/idempotency.rs index 80b01ce..fa0d9e6 100644 --- a/src/services/idempotency.rs +++ b/src/services/idempotency.rs @@ -3,7 +3,6 @@ use crate::state::AppState; use chrono::{DateTime, Duration, Utc}; use serde_json::Value as JsonValue; -use sha2::{Digest, Sha256}; use sqlx::FromRow; use uuid::Uuid; @@ -27,15 +26,6 @@ struct IdemRow { response_body: Option, } -pub fn sha256_hex(parts: &[&[u8]]) -> String { - let mut hasher = Sha256::new(); - for p in parts { - hasher.update(p); - hasher.update([0u8]); // separator - } - hex::encode(hasher.finalize()) -} - pub async fn begin( state: &AppState, scope: Scope,