From f2d490edcee37eb06c4aebf9d13460c67b2d6ac3 Mon Sep 17 00:00:00 2001 From: 237899745 <237899745@users.noreply.git.workyai.cn> Date: Sun, 26 Jul 2026 10:54:01 +0800 Subject: [PATCH] fix(billing): enforce one effective subscription per user --- frontend/src/pages/admin/AdminBillingPage.vue | 2 +- ..._cross_provider_subscription_invariant.sql | 65 +++++ src/api/admin.rs | 244 ++++++++++++++++-- src/api/billing.rs | 29 ++- src/api/webhooks.rs | 48 +++- 5 files changed, 349 insertions(+), 39 deletions(-) create mode 100644 migrations/023_cross_provider_subscription_invariant.sql diff --git a/frontend/src/pages/admin/AdminBillingPage.vue b/frontend/src/pages/admin/AdminBillingPage.vue index 27a6782..8b799d3 100644 --- a/frontend/src/pages/admin/AdminBillingPage.vue +++ b/frontend/src/pages/admin/AdminBillingPage.vue @@ -212,7 +212,7 @@ onMounted(async () => { > {{ subBusy ? '提交中…' : '立即开通' }} - 会取消该用户当前有效订阅，并按月数顺延。 + 会替换当前本地套餐；存在未取消 Stripe 订阅时将拒绝操作。
diff --git a/migrations/023_cross_provider_subscription_invariant.sql b/migrations/023_cross_provider_subscription_invariant.sql new file mode 100644 index 0000000..14afae6 --- /dev/null +++ b/migrations/023_cross_provider_subscription_invariant.sql @@ -0,0 +1,65 @@ +WITH ranked AS ( + SELECT + id, + user_id, + provider, + status::text AS previous_status, + current_period_end, + ROW_NUMBER() OVER ( + PARTITION BY user_id + ORDER BY + CASE WHEN provider = 'stripe' THEN 0 ELSE 1 END, + current_period_end DESC, + updated_at DESC, + id DESC + ) AS position + FROM subscriptions + WHERE status IN ('active', 'trialing', 'past_due') +), duplicates AS ( + SELECT * + FROM ranked + WHERE position > 1 +) +INSERT INTO audit_logs ( + user_id, action, resource_type, resource_id, details +) +SELECT + user_id, + 'migration_subscription_dedup', + 'subscription', + id, + jsonb_build_object( + 'migration', '023_cross_provider_subscription_invariant', + 'provider', provider, + 'previous_status', previous_status, + 'current_period_end', current_period_end, + 'reason', 'cross_provider_single_effective_subscription' + ) +FROM duplicates; + +WITH ranked AS ( + SELECT + id, + ROW_NUMBER() OVER ( + PARTITION BY user_id + ORDER BY + CASE WHEN provider = 'stripe' THEN 0 ELSE 1 END, + current_period_end DESC, + updated_at DESC, + id DESC + ) AS position + FROM subscriptions + WHERE status IN ('active', 'trialing', 'past_due') +) +UPDATE subscriptions AS subscription +SET status = 'canceled', + cancel_at_period_end = false, + canceled_at = COALESCE(subscription.canceled_at, NOW()), + updated_at = NOW() +FROM ranked +WHERE ranked.position > 1 + AND subscription.id = ranked.id; + +CREATE UNIQUE INDEX idx_subscriptions_user_effective_unique + ON subscriptions(user_id) + WHERE status IN ('active', 'trialing', 'past_due'); diff --git a/src/api/admin.rs b/src/api/admin.rs index fc30b31..60e8ae7 100644 --- a/src/api/admin.rs +++ b/src/api/admin.rs @@ -959,28 +959,92 @@ async fn create_manual_subscription( return Err(AppError::new(ErrorCode::Forbidden, "套餐不可用")); } + let (subscription_id, period_start, period_end) = persist_manual_subscription( + &state.db, + admin_id, + user_id, + plan.id, + months, + req.note.as_deref(), + ip, + ) + .await?; + + Ok(Json(Envelope { + success: true, + data: ManualSubscriptionResponse { + message: "套餐已开通".to_string(), + subscription_id, + user_id, + plan_id: plan.id, + plan_name: plan.name, + period_start, + period_end, + status: "active".to_string(), + }, + })) +} + +async fn persist_manual_subscription( + pool: &sqlx::PgPool, + admin_id: Uuid, + user_id: Uuid, + plan_id: Uuid, + months: i32, + note: Option<&str>, + ip: IpAddr, +) -> Result<(Uuid, DateTime, DateTime), AppError> { let period_start = Utc::now(); let period_end = add_months_utc8(period_start, months)?; - - let mut tx = state - .db + let mut tx = pool .begin() .await .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?; - let _ = sqlx::query( + let _: Uuid = sqlx::query_scalar("SELECT id FROM users WHERE id = $1 FOR UPDATE") + .bind(user_id) + .fetch_one(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "锁定订阅用户失败").with_source(err))?; + + let has_open_stripe: bool = sqlx::query_scalar( + r#" + SELECT EXISTS( + SELECT 1 + FROM subscriptions + WHERE user_id = $1 + AND provider = 'stripe' + AND status <> 'canceled' + ) + "#, + ) + .bind(user_id) + .fetch_one(&mut *tx) + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "检查 Stripe 订阅失败").with_source(err))?; + if has_open_stripe { + return Err(AppError::new( + ErrorCode::Forbidden, + "用户存在未取消的 Stripe 订阅，不能直接替换为手工套餐", + )); + } + + sqlx::query( r#" UPDATE subscriptions SET status = 'canceled', cancel_at_period_end = false, canceled_at = NOW(), updated_at = NOW() - WHERE user_id = $1 AND status IN ('active', 'trialing', 'past_due') + WHERE user_id = $1 + AND provider <> 'stripe' + AND status IN ('active', 'trialing', 'past_due') "#, ) .bind(user_id) .execute(&mut *tx) - .await; + .await + .map_err(|err| AppError::new(ErrorCode::Internal, "关闭原本地订阅失败").with_source(err))?; let subscription_id: Uuid = sqlx::query_scalar( r#" @@ -999,7 +1063,7 @@ async fn create_manual_subscription( "#, ) .bind(user_id) - .bind(plan.id) + .bind(plan_id) .bind(period_start) .bind(period_end) .fetch_one(&mut *tx) @@ -1031,9 +1095,9 @@ async fn create_manual_subscription( .bind(subscription_id) .bind(serde_json::json!({ "target_user_id": user_id, - "plan_id": plan.id, + "plan_id": plan_id, "months": months, - "note": req.note, + "note": note, })) .bind(ip.to_string()) .execute(&mut *tx) @@ -1044,19 +1108,7 @@ async fn create_manual_subscription( .await .map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?; - Ok(Json(Envelope { - success: true, - data: ManualSubscriptionResponse { - message: "套餐已开通".to_string(), - subscription_id, - user_id, - plan_id: plan.id, - plan_name: plan.name, - period_start, - period_end, - status: "active".to_string(), - }, - })) + Ok((subscription_id, period_start, period_end)) } fn add_months_utc8(start: DateTime, months: i32) -> Result, AppError> { @@ -1740,6 +1792,7 @@ async fn audit_config_action( #[cfg(test)] mod tests { use super::*; + use sqlx::postgres::PgPoolOptions; #[test] fn secret_masking_never_splits_utf8() { @@ -1747,4 +1800,151 @@ mod tests { assert_eq!(mask_secret("中文密钥测试内容"), "中文密钥测试内容"); assert_eq!(mask_secret("🔑🔑🔑🔑🔑🔑🔑🔑more"), "🔑🔑🔑🔑🔑🔑🔑🔑..."); } + + #[tokio::test] + #[ignore = "requires IMAGEFORGE_TEST_DATABASE_URL"] + async fn manual_subscriptions_are_serialized_and_cannot_replace_stripe() { + let database_url = std::env::var("IMAGEFORGE_TEST_DATABASE_URL") + .expect("IMAGEFORGE_TEST_DATABASE_URL is required"); + let pool = PgPoolOptions::new() + .max_connections(32) + .connect(&database_url) + .await + .expect("connect test database"); + sqlx::migrate!() + .run(&pool) + .await + .expect("apply test migrations"); + + let marker = Uuid::new_v4().simple().to_string(); + let admin_id: Uuid = sqlx::query_scalar( + r#" + INSERT INTO users (email, username, password_hash, role, email_verified_at) + VALUES ($1, $2, 'test', 'admin', NOW()) + RETURNING id + "#, + ) + .bind(format!("admin-{marker}@example.test")) + .bind(format!("admin-{marker}")) + .fetch_one(&pool) + .await + .expect("insert test admin"); + let user_id: Uuid = sqlx::query_scalar( + r#" + INSERT INTO users (email, username, password_hash, email_verified_at) + VALUES ($1, $2, 'test', NOW()) + RETURNING id + "#, + ) + .bind(format!("user-{marker}@example.test")) + .bind(format!("user-{marker}")) + .fetch_one(&pool) + .await + .expect("insert test user"); + let plan_id: Uuid = sqlx::query_scalar("SELECT id FROM plans WHERE code = 'pro_monthly'") + .fetch_one(&pool) + .await + .expect("load test plan"); + let ip: IpAddr = "127.0.0.1".parse().unwrap(); + + let mut joins = Vec::new(); + for _ in 0..20 { + let pool = pool.clone(); + joins.push(tokio::spawn(async move { + persist_manual_subscription( + &pool, + admin_id, + user_id, + plan_id, + 1, + Some("concurrency-test"), + ip, + ) + .await + })); + } + for join in joins { + join.await + .expect("manual subscription task panicked") + .expect("manual subscription failed"); + } + + let effective_manual: i64 = sqlx::query_scalar( + r#" + SELECT COUNT(*) + FROM subscriptions + WHERE user_id = $1 + AND provider = 'manual' + AND status IN ('active', 'trialing', 'past_due') + "#, + ) + .bind(user_id) + .fetch_one(&pool) + .await + .expect("count effective manual subscriptions"); + assert_eq!(effective_manual, 1); + + sqlx::query( + "UPDATE subscriptions SET status = 'canceled', canceled_at = NOW() WHERE user_id = $1", + ) + .bind(user_id) + .execute(&pool) + .await + .expect("cancel test manual subscription"); + 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 '1 month', + 'stripe', $3, $4 + ) + "#, + ) + .bind(user_id) + .bind(plan_id) + .bind(format!("cus_{marker}")) + .bind(format!("sub_{marker}")) + .execute(&pool) + .await + .expect("insert Stripe subscription"); + + let error = persist_manual_subscription( + &pool, + admin_id, + user_id, + plan_id, + 1, + Some("must-not-replace-stripe"), + ip, + ) + .await + .expect_err("manual subscription replaced Stripe"); + assert_eq!(error.code, ErrorCode::Forbidden); + let effective_subscriptions: i64 = sqlx::query_scalar( + r#" + SELECT COUNT(*) + FROM subscriptions + WHERE user_id = $1 + AND status IN ('active', 'trialing', 'past_due') + "#, + ) + .bind(user_id) + .fetch_one(&pool) + .await + .expect("count effective subscriptions"); + assert_eq!(effective_subscriptions, 1); + + sqlx::query("DELETE FROM audit_logs WHERE details->>'target_user_id' = $1") + .bind(user_id.to_string()) + .execute(&pool) + .await + .expect("clean test audit logs"); + sqlx::query("DELETE FROM users WHERE id = ANY($1)") + .bind(vec![user_id, admin_id]) + .execute(&pool) + .await + .expect("clean test users"); + } } diff --git a/src/api/billing.rs b/src/api/billing.rs index ee460ba..2964bc9 100644 --- a/src/api/billing.rs +++ b/src/api/billing.rs @@ -457,23 +457,30 @@ async fn create_checkout_for_user( .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( + let open_subscription_provider: Option = sqlx::query_scalar( r#" - SELECT EXISTS( - SELECT 1 FROM subscriptions - WHERE user_id = $1 AND provider = 'stripe' AND status <> 'canceled' - ) + SELECT provider + FROM subscriptions + WHERE user_id = $1 + AND ( + (provider = 'stripe' AND status <> 'canceled') + OR status IN ('active', 'trialing', 'past_due') + ) + ORDER BY CASE WHEN provider = 'stripe' THEN 0 ELSE 1 END + LIMIT 1 "#, ) .bind(user_id) - .fetch_one(&mut *tx) + .fetch_optional(&mut *tx) .await .map_err(|err| AppError::new(ErrorCode::Internal, "查询订阅状态失败").with_source(err))?; - if has_open_subscription { - return Err(AppError::new( - ErrorCode::IdempotencyConflict, - "已有 Stripe 订阅，请通过账单门户升级、降级或续费", - )); + if let Some(provider) = open_subscription_provider { + let message = if provider == "stripe" { + "已有 Stripe 订阅，请通过账单门户升级、降级或续费" + } else { + "当前已有有效套餐，请在套餐结束后创建 Stripe 订阅" + }; + return Err(AppError::new(ErrorCode::IdempotencyConflict, message)); } sqlx::query( diff --git a/src/api/webhooks.rs b/src/api/webhooks.rs index 3173221..b7f28fd 100644 --- a/src/api/webhooks.rs +++ b/src/api/webhooks.rs @@ -642,12 +642,14 @@ async fn write_subscription( if subscription.status != "canceled" { let conflicting: Option = sqlx::query_scalar( r#" - SELECT provider_subscription_id + SELECT provider || ':' || COALESCE(provider_subscription_id, id::text) FROM subscriptions WHERE user_id = $1 - AND provider = 'stripe' - AND status <> 'canceled' - AND provider_subscription_id <> $2 + AND status IN ('active', 'trialing', 'past_due') + AND NOT ( + provider = 'stripe' + AND provider_subscription_id = $2 + ) FOR UPDATE "#, ) @@ -661,7 +663,7 @@ async fn write_subscription( if conflicting.is_some() { return Err(AppError::new( ErrorCode::StorageUnavailable, - "用户已有其他未取消 Stripe 订阅，事件等待人工对账", + "用户已有其他有效订阅，Stripe 事件等待人工对账", )); } } @@ -1910,6 +1912,42 @@ mod tests { .await .expect("cancel primary subscription"); + let manual_subscription_id: Uuid = sqlx::query_scalar( + r#" + INSERT INTO subscriptions ( + user_id, plan_id, status, current_period_start, current_period_end, provider + ) VALUES ($1, $2, 'active', NOW(), NOW() + INTERVAL '1 month', 'manual') + RETURNING id + "#, + ) + .bind(user_id) + .bind(plan_id) + .fetch_one(&pool) + .await + .expect("insert manual subscription before delayed Stripe event"); + let delayed_active_error = apply_test_event(&state, &secondary) + .await + .expect_err("delayed active Stripe event created cross-provider double entitlement"); + assert_eq!(delayed_active_error.code, ErrorCode::StorageUnavailable); + let effective_after_delay: i64 = sqlx::query_scalar( + r#" + SELECT COUNT(*) FROM subscriptions + WHERE user_id = $1 AND status IN ('active', 'trialing', 'past_due') + "#, + ) + .bind(user_id) + .fetch_one(&pool) + .await + .expect("count effective subscriptions after delayed Stripe event"); + assert_eq!(effective_after_delay, 1); + sqlx::query( + "UPDATE subscriptions SET status = 'canceled', canceled_at = NOW() WHERE id = $1", + ) + .bind(manual_subscription_id) + .execute(&pool) + .await + .expect("cancel delayed-event manual fixture"); + let concurrent_invoice_id = format!("inv_{marker}_concurrent"); let concurrent_invoice_number = format!("INV-{marker}-C"); let concurrent_invoice_events = [