fix: reconcile ambiguous Stripe event ordering

.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

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

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}

docs/api.md

@@ -592,7 +592,6 @@ Authorization: Bearer <token>
 POST /billing/checkout
 Authorization: Bearer <token>
 Content-Type: application/json
-Idempotency-Key: <key>
 ```
 
 请求体：
@@ -605,6 +604,8 @@ Idempotency-Key: <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

docs/billing.md

@@ -130,14 +130,18 @@ RETURNING used_units;
 - `payments.provider_payment_id` ↔ Stripe `payment_intent.id`（或 charge id，按实现选）
 
 ### 4.2 Checkout / Portal
-- Checkout：后端创建 Stripe Checkout Session，前端跳转 `checkout_url`。
+- Checkout：后端按用户行锁串行创建 Stripe Checkout Session，前端跳转 `checkout_url`。服务端使用稳定 Customer 幂等键和待支付记录 ID 对应的 Session 幂等键，不依赖客户端 `Idempotency-Key`。
+- 每个用户只能映射一个非空 Stripe Customer，同时只能存在一个未过期待支付记录和一个未取消 Stripe 订阅。并发请求复用同一 Session；Customer 映射未持久化时禁止返回或创建 Session。
+- 已有未取消 Stripe 订阅的用户不能再次进入订阅 Checkout，升级、降级、续费和取消统一走 Portal，避免多重周期扣费。
 - Portal：后端创建 Stripe Billing Portal Session，前端跳转管理支付方式/取消订阅。
 
 ### 4.3 Stripe Webhook（商用必须）
 要求：
 - **验签**：使用 `STRIPE_WEBHOOK_SECRET` 校验 `Stripe-Signature`。
 - **事件幂等**：按 `provider_event_id` 去重（落库 `webhook_events`）。
-- **乱序容忍**：订阅对象按 `(event.created, 事件优先级, event.id)` 保存独立水位；`deleted` 即使先到也会保留 tombstone，旧 `created/updated` 不得恢复已取消订阅。
+- **乱序容忍**：订阅对象按 `(event.created, 事件优先级)` 保存独立水位；`deleted` 即使先到也会保留 tombstone，旧 `created/updated` 不得恢复已取消订阅。
+- **同秒歧义**：两个不同事件具有相同 `(event.created, 事件优先级)` 时，不能用不透明的 Event ID 排序，必须从 Stripe 拉取当前订阅快照并以快照响应时间推进水位。
+- **迁移对账**：历史版本用本地 `subscriptions.updated_at` 播种的非终态水位会标记为待对账；API 后台任务持租约获取 Stripe 快照，成功后才清除标记。未映射 Customer 或 Price 的受管订阅事件返回失败并等待重试，不能标记为已处理。
 - **并发一致性**：`subscriptions(provider, provider_subscription_id)` 唯一，订阅业务写入与 `webhook_events=processed` 在同一事务提交。
 - **可重放**：保存原始 payload（脱敏）用于排查。
 

docs/database.md

@@ -310,6 +310,13 @@ CREATE UNIQUE INDEX idx_webhook_events_unique ON webhook_events(provider, provid
 CREATE INDEX idx_webhook_events_status ON webhook_events(status);
 ```
 
+Stripe 运行时还通过迁移维护三组一致性结构：
+- `billing_checkout_sessions` 持久化每用户唯一的待支付 Session 及处理租约，防止并发创建多个 Customer/Session。
+- `provider_object_event_watermarks` 以 Stripe `event.created` 和事件等级保存对象水位；同秒同等级的不同事件标记为歧义并触发权威快照，不能按 Event ID 字典序决定先后。
+- `stripe_subscription_reconciliations` 保存历史非因果水位的租约化对账任务，允许多 API 实例用 `FOR UPDATE SKIP LOCKED` 安全消费。
+
+数据库唯一索引同时保证非空 `users.billing_customer_id` 全局唯一、`subscriptions(provider, provider_subscription_id)` 唯一，以及每用户最多一条未取消 Stripe 订阅。部署这些索引前必须先清理存量冲突，具体检查见 `docs/deployment.md`。
+
 ### 4.8 tasks - 压缩任务
 ```sql
 CREATE TABLE tasks (

docs/deployment.md

@@ -55,15 +55,48 @@ curl --fail http://127.0.0.1:8080/metrics
 
 ### 更新与回滚
 
-更新代码后保留 `.env.production` 和命名卷，重新构建并滚动重建：
+更新代码后保留 `.env.production` 和命名卷。包含迁移 `017`、`018` 的版本不能直接让旧、新 Worker 并行滚动：先备份数据库并停止旧 Worker，再构建新镜像。
+
+迁移 `017` 会在发现重复 Customer 或同用户多条未取消 Stripe 订阅时主动失败。部署前先检查并人工对账，两个查询都必须返回 0 行：
+
+```sql
+SELECT billing_customer_id, COUNT(*)
+FROM users
+WHERE billing_customer_id IS NOT NULL AND billing_customer_id <> ''
+GROUP BY billing_customer_id
+HAVING COUNT(*) > 1;
+
+SELECT user_id, COUNT(*)
+FROM subscriptions
+WHERE provider = 'stripe' AND status <> 'canceled'
+GROUP BY user_id
+HAVING COUNT(*) > 1;
+```
+
+推荐顺序：
 
 ```bash
 git pull --ff-only
+docker compose --env-file .env.production -f docker/docker-compose.prod.yml stop worker
 docker compose --env-file .env.production -f docker/docker-compose.prod.yml build api
-docker compose --env-file .env.production -f docker/docker-compose.prod.yml up -d
+docker compose --env-file .env.production -f docker/docker-compose.prod.yml up -d postgres redis api
+docker compose --env-file .env.production -f docker/docker-compose.prod.yml up -d worker
 ```
 
-生产镜像应使用不可变的 `IMAGEFORGE_TAG`。回滚时把该值改回上一镜像标签，然后再次运行 `up -d`。
+新 API 启动后会消费迁移 `018` 创建的 Stripe 对账队列。启动 Worker 前应确认 API 健康、`STRIPE_SECRET_KEY` 可用且服务器能访问 `STRIPE_API_BASE_URL`；对账可以后台继续，但必须监控失败项：
+
+```sql
+SELECT status, COUNT(*)
+FROM stripe_subscription_reconciliations
+GROUP BY status;
+
+SELECT provider_object_id, reconciliation_reason, updated_at
+FROM provider_object_event_watermarks
+WHERE provider = 'stripe' AND requires_reconciliation = true
+ORDER BY updated_at;
+```
+
+`failed` 会指数退避重试；持续失败通常表示 Stripe 凭据、网络、Customer/Price 映射不完整。上线验收要求 `pending/processing/failed` 最终归零，且 `requires_reconciliation=true` 为 0。生产镜像应使用不可变的 `IMAGEFORGE_TAG`。数据库迁移已应用后，不能只回滚旧二进制；应保留新 schema，并使用兼容该 schema 的修复镜像。
 
 ### 反向代理
 

docs/email.md

@@ -360,7 +360,7 @@ Content-Type: application/json
 { "success": true, "data": { "message": "邮箱验证成功", "session_invalidated": false } }
 ```
 
-邮箱变更复用该确认入口，但申请变更必须先通过当前密码校验。新邮箱确认前不会替换 `users.email`，因此不能作为密码恢复地址；确认时会原子切换邮箱、提升 `token_version`、撤销未使用的密码重置链接，并向旧邮箱发送安全通知。邮箱变更响应的 `session_invalidated` 为 `true`，客户端应要求重新登录。
+邮箱变更复用该确认入口，但申请变更必须先通过当前密码校验。新邮箱确认前不会替换 `users.email`，因此不能作为密码恢复地址；确认时会原子切换邮箱、提升 `token_version`、撤销未使用的密码重置链接，并向旧邮箱发送安全通知。反向地，成功修改或重置密码也会在同一用户行锁事务内撤销所有未使用重置链接和待确认邮箱变更。邮箱变更响应的 `session_invalidated` 为 `true`，客户端应要求重新登录。
 
 ### 6.3 请求密码重置
 

docs/security.md

@@ -31,7 +31,7 @@
 ### 2.2 JWT 使用建议
 - 对外 API：支持 Bearer Token（适合 CLI/SDK）。
 - 网站（Vue3）：优先使用 HttpOnly Cookie 承载会话（降低 XSS 泄露风险），如使用 localStorage 必须配合严格 CSP。
-- JWT 包含用户 `token_version`；修改或重置密码会递增版本，使此前签发的 JWT 立即失效。
+- JWT 包含用户 `token_version`；修改或重置密码会递增版本，使此前签发的 JWT 立即失效。成功修改或重置密码时，服务端持有用户行锁并在同一事务中消费该用户全部未使用重置链接、撤销全部待确认邮箱变更，避免旧恢复凭据再次接管账号。
 
 ---
 

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';

migrations/018_stripe_watermark_reconciliation.sql

@@ -0,0 +1,57 @@
+ALTER TABLE provider_object_event_watermarks
+  ADD COLUMN IF NOT EXISTS requires_reconciliation BOOLEAN NOT NULL DEFAULT false,
+  ADD COLUMN IF NOT EXISTS reconciliation_reason TEXT,
+  ADD COLUMN IF NOT EXISTS last_snapshot_at TIMESTAMPTZ;
+
+CREATE TABLE IF NOT EXISTS stripe_subscription_reconciliations (
+  provider_subscription_id VARCHAR(200) PRIMARY KEY,
+  reason TEXT NOT NULL,
+  status VARCHAR(20) NOT NULL DEFAULT 'pending',
+  attempts INTEGER NOT NULL DEFAULT 0,
+  next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  last_error TEXT,
+  lease_owner UUID,
+  lease_until TIMESTAMPTZ,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  completed_at TIMESTAMPTZ,
+  CONSTRAINT stripe_subscription_reconciliations_status_check
+    CHECK (status IN ('pending', 'processing', 'completed', 'failed'))
+);
+
+WITH corrected AS (
+  UPDATE provider_object_event_watermarks
+  SET last_event_created = 0,
+      last_event_rank = 0,
+      last_event_id = 'reconcile:migration',
+      requires_reconciliation = true,
+      reconciliation_reason = 'migration_016_non_causal_seed',
+      updated_at = NOW()
+  WHERE provider = 'stripe'
+    AND object_type = 'subscription'
+    AND is_deleted = false
+    AND last_event_id LIKE 'migration:%'
+  RETURNING provider_object_id
+)
+INSERT INTO stripe_subscription_reconciliations (
+  provider_subscription_id, reason, status, next_attempt_at
+)
+SELECT
+  provider_object_id,
+  'migration_016_non_causal_seed',
+  'pending',
+  NOW()
+FROM corrected
+ON CONFLICT (provider_subscription_id) DO UPDATE
+SET reason = EXCLUDED.reason,
+    status = 'pending',
+    next_attempt_at = NOW(),
+    last_error = NULL,
+    lease_owner = NULL,
+    lease_until = NULL,
+    completed_at = NULL,
+    updated_at = NOW();
+
+CREATE INDEX IF NOT EXISTS idx_stripe_subscription_reconciliations_ready
+  ON stripe_subscription_reconciliations(next_attempt_at, updated_at)
+  WHERE status IN ('pending', 'failed');

src/api/auth.rs

@@ -856,14 +856,7 @@ async fn reset_password(
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "更新密码失败").with_source(err))?;
 
-    sqlx::query(
-        "UPDATE password_resets SET used_at = $2 WHERE token_hash = $1 AND used_at IS NULL",
-    )
-    .bind(token_hash)
-    .bind(now)
-    .execute(&mut *tx)
-    .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "更新重置记录失败").with_source(err))?;
+    credentials::invalidate_account_recovery(&mut tx, user_id, now).await?;
 
     tx.commit()
         .await

src/api/mod.rs

@@ -44,7 +44,7 @@ pub async fn run(state: AppState) -> Result<(), AppError> {
         .nest("/api/v1", v1)
         .fallback_service(static_service)
         .layer(axum::middleware::from_fn(request_context::middleware))
-        .with_state(state);
+        .with_state(state.clone());
 
     let listener = tokio::net::TcpListener::bind(&addr)
         .await
@@ -52,11 +52,14 @@ pub async fn run(state: AppState) -> Result<(), AppError> {
 
     tracing::info!(addr = %addr, "API server listening");
 
-    axum::serve(
+    let reconciliation_task = tokio::spawn(webhooks::reconciliation_loop(state.clone()));
+    let serve_result = axum::serve(
         listener,
         app.into_make_service_with_connect_info::<SocketAddr>(),
     )
-    .await
+    .await;
+    reconciliation_task.abort();
+    serve_result
         .map_err(|err| AppError::new(ErrorCode::Internal, "HTTP 服务异常退出").with_source(err))
 }
 

src/api/user.rs

@@ -474,9 +474,18 @@ async fn update_password(
         password_hash: String,
     }
 
-    let row = sqlx::query_as::<_, PasswordRow>("SELECT password_hash FROM users WHERE id = $1")
+    let new_hash = credentials::hash_password(&req.new_password).await?;
+    let now = Utc::now();
+    let mut tx = state
+        .db
+        .begin()
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
+    let row = sqlx::query_as::<_, PasswordRow>(
+        "SELECT password_hash FROM users WHERE id = $1 FOR UPDATE",
+    )
     .bind(user_id)
-        .fetch_one(&state.db)
+    .fetch_one(&mut *tx)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
 
@@ -484,16 +493,20 @@ async fn update_password(
         return Err(AppError::new(ErrorCode::Unauthorized, "密码错误"));
     }
 
-    let new_hash = credentials::hash_password(&req.new_password).await?;
     sqlx::query(
         "UPDATE users SET password_hash = $2, token_version = token_version + 1, updated_at = NOW() WHERE id = $1",
     )
         .bind(user_id)
         .bind(new_hash)
-        .execute(&state.db)
+        .execute(&mut *tx)
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "更新密码失败").with_source(err))?;
 
+    credentials::invalidate_account_recovery(&mut tx, user_id, now).await?;
+    tx.commit()
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "提交密码更新失败").with_source(err))?;
+
     Ok(Json(Envelope {
         success: true,
         data: MessageResponse {
@@ -1411,6 +1424,301 @@ mod tests {
                 .expect("query test admin");
         assert_eq!(admin, (admin_pending_email, "admin".to_string()));
 
+        let reset_user_id = Uuid::new_v4();
+        let reset_old_email = format!("reset-old-{marker}@example.test");
+        let reset_new_email = format!("reset-new-{marker}@example.test");
+        let reset_token_a = format!("reset-a-{marker}");
+        let reset_token_b = format!("reset-b-{marker}");
+        let reset_email_token = format!("reset-email-{marker}");
+        sqlx::query(
+            r#"
+            INSERT INTO users (id, email, username, password_hash, email_verified_at)
+            VALUES ($1, $2, $3, $4, NOW())
+            "#,
+        )
+        .bind(reset_user_id)
+        .bind(&reset_old_email)
+        .bind(format!("reset_{marker}"))
+        .bind(&password_hash)
+        .execute(&pool)
+        .await
+        .expect("insert multi-reset user");
+        sqlx::query(
+            r#"
+            INSERT INTO password_resets (user_id, token_hash, expires_at)
+            VALUES
+              ($1, $2, NOW() + INTERVAL '1 hour'),
+              ($1, $3, NOW() + INTERVAL '1 hour')
+            "#,
+        )
+        .bind(reset_user_id)
+        .bind(credentials::sha256_hex(&reset_token_a))
+        .bind(credentials::sha256_hex(&reset_token_b))
+        .execute(&pool)
+        .await
+        .expect("insert two password resets");
+        sqlx::query(
+            r#"
+            INSERT INTO email_change_requests (user_id, new_email, token_hash, expires_at)
+            VALUES ($1, $2, $3, NOW() + INTERVAL '1 hour')
+            "#,
+        )
+        .bind(reset_user_id)
+        .bind(&reset_new_email)
+        .bind(credentials::sha256_hex(&reset_email_token))
+        .execute(&pool)
+        .await
+        .expect("insert pending email change before reset");
+
+        let (status, response) = json_request(
+            &app,
+            Method::POST,
+            "/auth/reset-password",
+            None,
+            serde_json::json!({
+                "token": reset_token_a,
+                "new_password": "Replacement9!"
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{response}");
+        let (status, response) = json_request(
+            &app,
+            Method::POST,
+            "/auth/reset-password",
+            None,
+            serde_json::json!({
+                "token": reset_token_b,
+                "new_password": "SecondReplacement9!"
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::BAD_REQUEST, "{response}");
+        assert_eq!(response["error"]["code"], "INVALID_TOKEN");
+        let (status, response) = json_request(
+            &app,
+            Method::POST,
+            "/auth/verify-email",
+            None,
+            serde_json::json!({ "token": reset_email_token }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::BAD_REQUEST, "{response}");
+        assert_eq!(response["error"]["code"], "INVALID_TOKEN");
+        let recovery_state: (i64, i64) = sqlx::query_as(
+            r#"
+            SELECT
+              (SELECT COUNT(*) FROM password_resets WHERE user_id = $1 AND used_at IS NULL),
+              (SELECT COUNT(*) FROM email_change_requests
+               WHERE user_id = $1 AND confirmed_at IS NULL AND canceled_at IS NULL)
+            "#,
+        )
+        .bind(reset_user_id)
+        .fetch_one(&pool)
+        .await
+        .expect("query recovery invalidation state");
+        assert_eq!(recovery_state, (0, 0));
+
+        let password_user_id = Uuid::new_v4();
+        let password_email = format!("password-{marker}@example.test");
+        let password_reset_token = format!("password-reset-{marker}");
+        let password_email_token = format!("password-email-{marker}");
+        sqlx::query(
+            r#"
+            INSERT INTO users (id, email, username, password_hash, email_verified_at)
+            VALUES ($1, $2, $3, $4, NOW())
+            "#,
+        )
+        .bind(password_user_id)
+        .bind(&password_email)
+        .bind(format!("password_{marker}"))
+        .bind(&password_hash)
+        .execute(&pool)
+        .await
+        .expect("insert password-update user");
+        sqlx::query(
+            r#"
+            INSERT INTO password_resets (user_id, token_hash, expires_at)
+            VALUES ($1, $2, NOW() + INTERVAL '1 hour')
+            "#,
+        )
+        .bind(password_user_id)
+        .bind(credentials::sha256_hex(&password_reset_token))
+        .execute(&pool)
+        .await
+        .expect("insert reset before authenticated password update");
+        sqlx::query(
+            r#"
+            INSERT INTO email_change_requests (user_id, new_email, token_hash, expires_at)
+            VALUES ($1, $2, $3, NOW() + INTERVAL '1 hour')
+            "#,
+        )
+        .bind(password_user_id)
+        .bind(format!("password-new-{marker}@example.test"))
+        .bind(credentials::sha256_hex(&password_email_token))
+        .execute(&pool)
+        .await
+        .expect("insert email change before authenticated password update");
+        let (password_token, _) = auth::issue_jwt(
+            &state.config.jwt_secret,
+            state.config.jwt_expiry_hours,
+            password_user_id,
+            "user",
+            0,
+        )
+        .expect("issue password-update JWT");
+        let (status, response) = json_request(
+            &app,
+            Method::PUT,
+            "/user/password",
+            Some(&password_token),
+            serde_json::json!({
+                "current_password": password,
+                "new_password": "AuthenticatedReplacement9!"
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{response}");
+        let (status, response) = json_request(
+            &app,
+            Method::POST,
+            "/auth/reset-password",
+            None,
+            serde_json::json!({
+                "token": password_reset_token,
+                "new_password": "StaleReset9!"
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::BAD_REQUEST, "{response}");
+        assert_eq!(response["error"]["code"], "INVALID_TOKEN");
+        let (status, response) = json_request(
+            &app,
+            Method::POST,
+            "/auth/verify-email",
+            None,
+            serde_json::json!({ "token": password_email_token }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::BAD_REQUEST, "{response}");
+        assert_eq!(response["error"]["code"], "INVALID_TOKEN");
+
+        let reset_race_user_id = Uuid::new_v4();
+        let reset_race_old_email = format!("reset-race-old-{marker}@example.test");
+        let reset_race_new_email = format!("reset-race-new-{marker}@example.test");
+        let reset_race_token = format!("reset-race-{marker}");
+        let reset_race_email_token = format!("reset-race-email-{marker}");
+        sqlx::query(
+            r#"
+            INSERT INTO users (id, email, username, password_hash, email_verified_at)
+            VALUES ($1, $2, $3, $4, NOW())
+            "#,
+        )
+        .bind(reset_race_user_id)
+        .bind(&reset_race_old_email)
+        .bind(format!("reset_race_{marker}"))
+        .bind(&password_hash)
+        .execute(&pool)
+        .await
+        .expect("insert reset-email race user");
+        sqlx::query(
+            r#"
+            INSERT INTO password_resets (user_id, token_hash, expires_at)
+            VALUES ($1, $2, NOW() + INTERVAL '1 hour')
+            "#,
+        )
+        .bind(reset_race_user_id)
+        .bind(credentials::sha256_hex(&reset_race_token))
+        .execute(&pool)
+        .await
+        .expect("insert racing reset");
+        sqlx::query(
+            r#"
+            INSERT INTO email_change_requests (user_id, new_email, token_hash, expires_at)
+            VALUES ($1, $2, $3, NOW() + INTERVAL '1 hour')
+            "#,
+        )
+        .bind(reset_race_user_id)
+        .bind(&reset_race_new_email)
+        .bind(credentials::sha256_hex(&reset_race_email_token))
+        .execute(&pool)
+        .await
+        .expect("insert racing email change");
+        let mut blocker = pool.begin().await.expect("begin reset-email race blocker");
+        let _: Uuid = sqlx::query_scalar("SELECT id FROM users WHERE id = $1 FOR UPDATE")
+            .bind(reset_race_user_id)
+            .fetch_one(&mut *blocker)
+            .await
+            .expect("lock reset-email race user");
+        let reset_join = {
+            let app = app.clone();
+            let token = reset_race_token.clone();
+            tokio::spawn(async move {
+                json_request(
+                    &app,
+                    Method::POST,
+                    "/auth/reset-password",
+                    None,
+                    serde_json::json!({
+                        "token": token,
+                        "new_password": "RaceReplacement9!"
+                    }),
+                )
+                .await
+            })
+        };
+        let confirm_join = {
+            let app = app.clone();
+            let token = reset_race_email_token.clone();
+            tokio::spawn(async move {
+                json_request(
+                    &app,
+                    Method::POST,
+                    "/auth/verify-email",
+                    None,
+                    serde_json::json!({ "token": token }),
+                )
+                .await
+            })
+        };
+        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+        blocker
+            .commit()
+            .await
+            .expect("release reset-email race user");
+        let reset_result = reset_join.await.expect("join racing reset");
+        let confirm_result = confirm_join.await.expect("join racing confirmation");
+        let successes = [reset_result.0, confirm_result.0]
+            .into_iter()
+            .filter(|status| *status == StatusCode::OK)
+            .count();
+        assert_eq!(successes, 1, "reset and email confirmation both committed");
+        for (status, response) in [&reset_result, &confirm_result] {
+            if *status != StatusCode::OK {
+                assert_eq!(*status, StatusCode::BAD_REQUEST, "{response}");
+                assert_eq!(response["error"]["code"], "INVALID_TOKEN");
+            }
+        }
+        let (race_email, race_hash): (String, String) =
+            sqlx::query_as("SELECT email, password_hash FROM users WHERE id = $1")
+                .bind(reset_race_user_id)
+                .fetch_one(&pool)
+                .await
+                .expect("query reset-email race result");
+        if reset_result.0 == StatusCode::OK {
+            assert_eq!(race_email, reset_race_old_email);
+            assert!(
+                credentials::verify_password("RaceReplacement9!", &race_hash)
+                    .await
+                    .expect("verify racing reset password")
+            );
+        } else {
+            assert_eq!(race_email, reset_race_new_email);
+            assert!(credentials::verify_password(password, &race_hash)
+                .await
+                .expect("verify original password after email confirmation"));
+        }
+
         let race_user_id = Uuid::new_v4();
         let race_old_email = format!("race-old-{marker}@example.test");
         let race_new_email = format!("race-new-{marker}@example.test");
@@ -1502,10 +1810,13 @@ mod tests {
         .expect("query recovery race result");
         assert_eq!(race_result, (race_new_email, 0));
 
-        sqlx::query("DELETE FROM users WHERE id IN ($1, $2, $3)")
+        sqlx::query("DELETE FROM users WHERE id IN ($1, $2, $3, $4, $5, $6)")
             .bind(user_id)
             .bind(admin_id)
             .bind(race_user_id)
+            .bind(reset_user_id)
+            .bind(password_user_id)
+            .bind(reset_race_user_id)
             .execute(&pool)
             .await
             .expect("delete account recovery test users");

src/config.rs

@@ -25,6 +25,7 @@ pub struct Config {
 
     pub stripe_secret_key: Option<String>,
     pub stripe_webhook_secret: Option<String>,
+    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,

src/services/credentials.rs

@@ -2,8 +2,11 @@ use crate::error::{AppError, ErrorCode};
 
 use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier};
 use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
+use chrono::{DateTime, Utc};
 use rand::RngCore;
 use sha2::{Digest, Sha256};
+use sqlx::{Postgres, Transaction};
+use uuid::Uuid;
 
 pub fn validate_email(email: &str) -> Result<(), AppError> {
     if email.trim().is_empty() || !email.contains('@') {
@@ -80,6 +83,34 @@ pub async fn consume_dummy_password_work(password: &str) -> Result<(), AppError>
     .map_err(|err| AppError::new(ErrorCode::Internal, "密码校验失败").with_source(err))
 }
 
+pub async fn invalidate_account_recovery(
+    tx: &mut Transaction<'_, Postgres>,
+    user_id: Uuid,
+    invalidated_at: DateTime<Utc>,
+) -> Result<(), AppError> {
+    sqlx::query(
+        r#"
+        WITH consumed_resets AS (
+          UPDATE password_resets
+          SET used_at = $2
+          WHERE user_id = $1 AND used_at IS NULL
+          RETURNING id
+        )
+        UPDATE email_change_requests
+        SET canceled_at = $2
+        WHERE user_id = $1
+          AND confirmed_at IS NULL
+          AND canceled_at IS NULL
+        "#,
+    )
+    .bind(user_id)
+    .bind(invalidated_at)
+    .execute(&mut **tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "撤销账号恢复凭据失败").with_source(err))?;
+    Ok(())
+}
+
 pub fn generate_token() -> String {
     let mut bytes = [0u8; 32];
     rand::rngs::OsRng.fill_bytes(&mut bytes);

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<JsonValue>,
 }
 
-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,