fix(billing): enforce one effective subscription per user

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<Utc>, DateTime<Utc>), 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<Utc>, months: i32) -> Result<DateTime<Utc>, 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");
+    }
 }

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<String> = 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(

src/api/webhooks.rs

@@ -642,12 +642,14 @@ async fn write_subscription(
     if subscription.status != "canceled" {
         let conflicting: Option<String> = 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 = [