fix: make Stripe subscription events monotonic

Cargo.lock

@@ -2215,6 +2215,7 @@ dependencies = [
  "time",
  "tokio",
  "tokio-util",
+ "tower",
  "tower-http",
  "tracing",
  "tracing-subscriber",

Cargo.toml

@@ -67,3 +67,6 @@ zip = "2"
 [target.'cfg(target_os = "linux")'.dependencies]
 image = { version = "0.25", default-features = false, features = ["avif-native"] }
 ravif = { version = "0.11", default-features = false, features = ["asm", "threading"] }
+
+[dev-dependencies]
+tower = { version = "0.5", features = ["util"] }

docs/api.md

@@ -214,9 +214,11 @@ Content-Type: application/json
 
 响应：
 ```json
-{ "success": true, "data": { "message": "邮箱验证成功" } }
+{ "success": true, "data": { "message": "邮箱验证成功", "session_invalidated": false } }
 ```
 
+同一路径也用于确认邮箱变更。邮箱变更成功时 `session_invalidated=true`，服务端会提升 `token_version`，客户端必须清除旧 JWT 并重新登录。
+
 ### 4.7 请求密码重置
 ```http
 POST /auth/forgot-password
@@ -447,7 +449,9 @@ Authorization: Bearer <token>
     "id": "550e8400-e29b-41d4-a716-446655440000",
     "email": "user@example.com",
     "username": "myusername",
-    "role": "user"
+    "role": "user",
+    "email_verified": true,
+    "pending_email": null
   }
 }
 ```
@@ -696,6 +700,17 @@ Authorization: Bearer <admin_token>
 Content-Type: application/json
 ```
 
+修改用户名只需提交 `username`。修改邮箱必须同时提交当前密码：
+
+```json
+{
+  "email": "new@example.com",
+  "current_password": "current-password"
+}
+```
+
+邮箱验证开启时，新邮箱只写入 `pending_email`，确认前主邮箱、登录邮箱和密码恢复地址均保持不变；确认后服务端原子切换主邮箱并撤销全部旧 JWT 与未使用的密码重置链接。邮箱验证关闭时，当前密码校验通过后立即切换邮箱，响应会返回替换当前会话使用的新 `token`。
+
 ### 11.6 S3 存储端点
 
 ```http

docs/billing.md

@@ -137,7 +137,8 @@ RETURNING used_units;
 要求：
 - **验签**：使用 `STRIPE_WEBHOOK_SECRET` 校验 `Stripe-Signature`。
 - **事件幂等**：按 `provider_event_id` 去重（落库 `webhook_events`）。
-- **乱序容忍**：以 Stripe 事件时间 + 当前 DB 状态做“只前进不回退”更新。
+- **乱序容忍**：订阅对象按 `(event.created, 事件优先级, event.id)` 保存独立水位；`deleted` 即使先到也会保留 tombstone，旧 `created/updated` 不得恢复已取消订阅。
+- **并发一致性**：`subscriptions(provider, provider_subscription_id)` 唯一，订阅业务写入与 `webhook_events=processed` 在同一事务提交。
 - **可重放**：保存原始 payload（脱敏）用于排查。
 
 建议首期处理的事件（示例）：

docs/email.md

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

frontend/src/pages/VerifyEmailPage.vue

@@ -25,7 +25,11 @@ onMounted(async () => {
   try {
     const resp = await verifyEmail(token.value)
     message.value = resp.message
+    if (resp.session_invalidated) {
+      auth.logout()
+    } else {
       auth.markEmailVerified()
+    }
   } catch (err) {
     if (err instanceof ApiError) {
       error.value = `[${err.code}] ${err.message}`

frontend/src/pages/dashboard/DashboardSettingsPage.vue

@@ -9,7 +9,7 @@ const auth = useAuthStore()
 
 const loading = ref(true)
 
-const profileForm = ref({ email: '', username: '' })
+const profileForm = ref({ email: '', username: '', currentPassword: '' })
 const profileBusy = ref(false)
 const profileMessage = ref<string | null>(null)
 const profileError = ref<string | null>(null)
@@ -24,9 +24,13 @@ const verificationMessage = ref<string | null>(null)
 const verificationError = ref<string | null>(null)
 
 const canResendVerification = computed(() => Boolean(auth.user && !auth.user.email_verified))
+const emailChangeRequested = computed(() => {
+  const email = profileForm.value.email.trim().toLowerCase()
+  return Boolean(email && auth.user && email !== auth.user.email)
+})
 
 function syncProfile(user: UserProfile) {
-  profileForm.value = { email: user.email ?? '', username: user.username ?? '' }
+  profileForm.value = { email: user.email ?? '', username: user.username ?? '', currentPassword: '' }
 }
 
 onMounted(async () => {
@@ -77,9 +81,16 @@ async function saveProfile() {
   try {
     const email = profileForm.value.email.trim().toLowerCase()
     const username = profileForm.value.username.trim()
-    const payload: { email?: string; username?: string } = {}
+    const payload: { email?: string; username?: string; current_password?: string } = {}
 
-    if (email && email !== auth.user.email) payload.email = email
+    if (email && email !== auth.user.email) {
+      if (!profileForm.value.currentPassword) {
+        profileError.value = '修改邮箱需要输入当前密码'
+        return
+      }
+      payload.email = email
+      payload.current_password = profileForm.value.currentPassword
+    }
     if (username && username !== auth.user.username) payload.username = username
 
     if (!payload.email && !payload.username) {
@@ -88,7 +99,11 @@ async function saveProfile() {
     }
 
     const resp = await updateProfile(auth.token, payload)
+    if (resp.token) {
+      auth.setAuth(resp.token, resp.user)
+    } else {
       auth.updateUser(resp.user)
+    }
     syncProfile(resp.user)
     profileMessage.value = resp.message
   } catch (err) {
@@ -178,6 +193,24 @@ async function changePassword() {
           </label>
         </div>
 
+        <label v-if="emailChangeRequested" class="mt-3 block max-w-md space-y-1">
+          <div class="text-xs font-medium text-slate-600">当前密码（修改邮箱必填）</div>
+          <input
+            v-model="profileForm.currentPassword"
+            type="password"
+            autocomplete="current-password"
+            class="w-full rounded-md border border-slate-200 bg-white px-3 py-2 text-sm text-slate-800"
+            placeholder="用于确认是本人操作"
+          />
+        </label>
+
+        <div
+          v-if="auth.user?.pending_email"
+          class="mt-3 rounded-lg border border-amber-200 bg-amber-50 p-3 text-sm text-amber-900"
+        >
+          待确认新邮箱：{{ auth.user.pending_email }}。确认前仍使用当前邮箱登录和找回密码。
+        </div>
+
         <div v-if="profileMessage" class="mt-4 rounded-lg border border-emerald-200 bg-emerald-50 p-3 text-sm text-emerald-900">
           {{ profileMessage }}
         </div>

frontend/src/services/api.ts

@@ -29,8 +29,14 @@ export async function sendVerification(token: string): Promise<{ message: string
   return apiJson<{ message: string }>('/api/v1/auth/send-verification', undefined, token, { method: 'POST' })
 }
 
-export async function verifyEmail(verificationToken: string): Promise<{ message: string }> {
-  return apiJson<{ message: string }>('/api/v1/auth/verify-email', { token: verificationToken }, null)
+export async function verifyEmail(
+  verificationToken: string,
+): Promise<{ message: string; session_invalidated: boolean }> {
+  return apiJson<{ message: string; session_invalidated: boolean }>(
+    '/api/v1/auth/verify-email',
+    { token: verificationToken },
+    null,
+  )
 }
 
 export async function forgotPassword(email: string): Promise<{ message: string }> {
@@ -49,9 +55,11 @@ export async function getProfile(token: string): Promise<UserProfile> {
 
 export async function updateProfile(
   token: string,
-  payload: { email?: string; username?: string },
-): Promise<{ user: UserProfile; message: string }> {
-  return apiJson<{ user: UserProfile; message: string }>('/api/v1/user/profile', payload, token, { method: 'PUT' })
+  payload: { email?: string; username?: string; current_password?: string },
+): Promise<{ user: UserProfile; message: string; token?: string }> {
+  return apiJson<{ user: UserProfile; message: string; token?: string }>('/api/v1/user/profile', payload, token, {
+    method: 'PUT',
+  })
 }
 
 export async function updatePassword(

frontend/src/stores/auth.ts

@@ -8,6 +8,7 @@ export interface User {
   username: string
   role: UserRole
   email_verified: boolean
+  pending_email?: string | null
 }
 
 interface StoredAuth {

migrations/014_email_change_requests.sql

@@ -0,0 +1,25 @@
+CREATE TABLE IF NOT EXISTS email_change_requests (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  new_email VARCHAR(255) NOT NULL,
+  token_hash VARCHAR(64) NOT NULL,
+  expires_at TIMESTAMPTZ NOT NULL,
+  confirmed_at TIMESTAMPTZ,
+  canceled_at TIMESTAMPTZ,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_email_change_requests_token
+  ON email_change_requests(token_hash);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_email_change_requests_pending_user
+  ON email_change_requests(user_id)
+  WHERE confirmed_at IS NULL AND canceled_at IS NULL;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_email_change_requests_pending_email
+  ON email_change_requests(new_email)
+  WHERE confirmed_at IS NULL AND canceled_at IS NULL;
+
+CREATE INDEX IF NOT EXISTS idx_email_change_requests_expires
+  ON email_change_requests(expires_at)
+  WHERE confirmed_at IS NULL AND canceled_at IS NULL;

migrations/015_worker_attempt_leases.sql

@@ -0,0 +1,17 @@
+ALTER TABLE tasks
+  ADD COLUMN IF NOT EXISTS processing_attempt BIGINT NOT NULL DEFAULT 0,
+  ADD COLUMN IF NOT EXISTS lease_owner UUID,
+  ADD COLUMN IF NOT EXISTS lease_until TIMESTAMPTZ;
+
+ALTER TABLE task_files
+  ADD COLUMN IF NOT EXISTS processing_attempt BIGINT NOT NULL DEFAULT 0,
+  ADD COLUMN IF NOT EXISTS lease_owner UUID,
+  ADD COLUMN IF NOT EXISTS lease_until TIMESTAMPTZ;
+
+CREATE INDEX IF NOT EXISTS idx_tasks_processing_lease
+  ON tasks(lease_until)
+  WHERE status = 'processing';
+
+CREATE INDEX IF NOT EXISTS idx_task_files_processing_lease
+  ON task_files(task_id, lease_until)
+  WHERE status = 'processing';

migrations/016_stripe_event_ordering.sql

@@ -0,0 +1,95 @@
+-- Repoint references before enforcing one row per provider subscription.
+WITH ranked AS (
+  SELECT
+    id,
+    FIRST_VALUE(id) OVER (
+      PARTITION BY provider, provider_subscription_id
+      ORDER BY updated_at DESC, created_at DESC, id DESC
+    ) AS keep_id,
+    ROW_NUMBER() OVER (
+      PARTITION BY provider, provider_subscription_id
+      ORDER BY updated_at DESC, created_at DESC, id DESC
+    ) AS row_number
+  FROM subscriptions
+  WHERE provider_subscription_id IS NOT NULL
+)
+UPDATE usage_periods AS usage
+SET subscription_id = ranked.keep_id
+FROM ranked
+WHERE ranked.row_number > 1
+  AND usage.subscription_id = ranked.id;
+
+WITH ranked AS (
+  SELECT
+    id,
+    FIRST_VALUE(id) OVER (
+      PARTITION BY provider, provider_subscription_id
+      ORDER BY updated_at DESC, created_at DESC, id DESC
+    ) AS keep_id,
+    ROW_NUMBER() OVER (
+      PARTITION BY provider, provider_subscription_id
+      ORDER BY updated_at DESC, created_at DESC, id DESC
+    ) AS row_number
+  FROM subscriptions
+  WHERE provider_subscription_id IS NOT NULL
+)
+UPDATE invoices AS invoice
+SET subscription_id = ranked.keep_id
+FROM ranked
+WHERE ranked.row_number > 1
+  AND invoice.subscription_id = ranked.id;
+
+WITH ranked AS (
+  SELECT
+    id,
+    ROW_NUMBER() OVER (
+      PARTITION BY provider, provider_subscription_id
+      ORDER BY updated_at DESC, created_at DESC, id DESC
+    ) AS row_number
+  FROM subscriptions
+  WHERE provider_subscription_id IS NOT NULL
+)
+DELETE FROM subscriptions AS subscription
+USING ranked
+WHERE ranked.row_number > 1
+  AND subscription.id = ranked.id;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_subscriptions_provider_object_unique
+  ON subscriptions(provider, provider_subscription_id);
+
+CREATE TABLE IF NOT EXISTS provider_object_event_watermarks (
+  provider VARCHAR(20) NOT NULL,
+  object_type VARCHAR(50) NOT NULL,
+  provider_object_id VARCHAR(200) NOT NULL,
+  last_event_created BIGINT NOT NULL,
+  last_event_rank SMALLINT NOT NULL DEFAULT 0,
+  last_event_id VARCHAR(200) NOT NULL,
+  is_deleted BOOLEAN NOT NULL DEFAULT false,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (provider, object_type, provider_object_id),
+  CONSTRAINT provider_object_event_watermarks_rank_check
+    CHECK (last_event_rank BETWEEN 0 AND 100)
+);
+
+CREATE INDEX IF NOT EXISTS idx_provider_object_event_watermarks_updated
+  ON provider_object_event_watermarks(updated_at);
+
+-- Seed rollout watermarks so a delayed pre-deployment event cannot revive an
+-- already-canceled Stripe subscription before the first new event arrives.
+INSERT INTO provider_object_event_watermarks (
+  provider, object_type, provider_object_id,
+  last_event_created, last_event_rank, last_event_id, is_deleted, updated_at
+)
+SELECT
+  'stripe',
+  'subscription',
+  provider_subscription_id,
+  EXTRACT(EPOCH FROM updated_at)::bigint,
+  CASE WHEN status = 'canceled' THEN 2 ELSE 1 END,
+  'migration:' || id::text,
+  status = 'canceled',
+  updated_at
+FROM subscriptions
+WHERE provider = 'stripe'
+  AND provider_subscription_id IS NOT NULL
+ON CONFLICT (provider, object_type, provider_object_id) DO NOTHING;

src/api/admin.rs

@@ -556,28 +556,14 @@ async fn cancel_task(
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
 
-    sqlx::query(
-        r#"
-        UPDATE task_files
-        SET status = 'failed',
-            error_message = '任务已取消',
-            completed_at = NOW()
-        WHERE task_id = $1 AND status IN ('pending', 'processing')
-        "#,
-    )
-    .bind(task_id)
-    .execute(&mut *tx)
-    .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "更新任务文件失败").with_source(err))?;
-
     let updated = sqlx::query(
         r#"
         UPDATE tasks
         SET status = 'cancelled',
             error_message = '管理员取消任务',
             completed_at = NOW(),
-            completed_files = (SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'completed'),
-            failed_files = (SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'failed')
+            lease_owner = NULL,
+            lease_until = NULL
         WHERE id = $1 AND status IN ('pending', 'processing')
         "#,
     )
@@ -590,6 +576,44 @@ async fn cancel_task(
         return Err(AppError::new(ErrorCode::InvalidRequest, "任务状态无法取消"));
     }
 
+    sqlx::query(
+        r#"
+        UPDATE task_files
+        SET status = 'failed',
+            error_message = '任务已取消',
+            completed_at = NOW(),
+            lease_owner = NULL,
+            lease_until = NULL
+        WHERE task_id = $1 AND status IN ('pending', 'processing')
+        "#,
+    )
+    .bind(task_id)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "更新任务文件失败").with_source(err))?;
+
+    sqlx::query(
+        r#"
+        UPDATE tasks
+        SET completed_files = (
+              SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'completed'
+            ),
+            failed_files = (
+              SELECT COUNT(*) FROM task_files WHERE task_id = $1 AND status = 'failed'
+            ),
+            total_compressed_size = COALESCE((
+              SELECT SUM(compressed_size)
+              FROM task_files
+              WHERE task_id = $1 AND status = 'completed'
+            ), 0)::bigint
+        WHERE id = $1 AND status = 'cancelled'
+        "#,
+    )
+    .bind(task_id)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "更新任务统计失败").with_source(err))?;
+
     tx.commit()
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;

src/api/auth.rs

@@ -415,12 +415,22 @@ struct VerifyEmailRequest {
     token: String,
 }
 
+#[derive(Debug, Serialize)]
+struct VerifyEmailResponse {
+    message: String,
+    session_invalidated: bool,
+}
+
+struct EmailChangeConfirmation {
+    old_email: String,
+}
+
 async fn verify_email(
     State(state): State<AppState>,
     ConnectInfo(addr): ConnectInfo<SocketAddr>,
     headers: HeaderMap,
     Json(req): Json<VerifyEmailRequest>,
-) -> Result<Json<Envelope<MessageResponse>>, AppError> {
+) -> Result<Json<Envelope<VerifyEmailResponse>>, AppError> {
     if req.token.trim().is_empty() {
         return Err(AppError::new(ErrorCode::InvalidRequest, "token 不能为空"));
     }
@@ -440,6 +450,19 @@ async fn verify_email(
     let token_hash = credentials::sha256_hex(&req.token);
     let now = Utc::now();
 
+    if let Some(confirmation) = confirm_email_change(&state, &token_hash, now).await? {
+        if let Err(err) = mail::send_email_change_notice(&state, &confirmation.old_email).await {
+            tracing::warn!(error = ?err, "email change notice delivery failed");
+        }
+        return Ok(Json(Envelope {
+            success: true,
+            data: VerifyEmailResponse {
+                message: "新邮箱确认成功，请重新登录".to_string(),
+                session_invalidated: true,
+            },
+        }));
+    }
+
     let updated = sqlx::query(
         r#"
         WITH v AS (
@@ -477,12 +500,149 @@ async fn verify_email(
 
     Ok(Json(Envelope {
         success: true,
-        data: MessageResponse {
+        data: VerifyEmailResponse {
             message: "邮箱验证成功".to_string(),
+            session_invalidated: false,
         },
     }))
 }
 
+async fn confirm_email_change(
+    state: &AppState,
+    token_hash: &str,
+    now: DateTime<Utc>,
+) -> Result<Option<EmailChangeConfirmation>, AppError> {
+    let user_id: Option<Uuid> = sqlx::query_scalar(
+        r#"
+        SELECT user_id
+        FROM email_change_requests
+        WHERE token_hash = $1
+          AND confirmed_at IS NULL
+          AND canceled_at IS NULL
+          AND expires_at > $2
+        "#,
+    )
+    .bind(token_hash)
+    .bind(now)
+    .fetch_optional(&state.db)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "查询邮箱变更请求失败").with_source(err))?;
+    let Some(user_id) = user_id else {
+        return Ok(None);
+    };
+
+    let mut tx = state
+        .db
+        .begin()
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
+    let user: Option<(String,)> =
+        sqlx::query_as("SELECT email 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))?;
+    let Some((old_email,)) = user else {
+        tx.rollback().await.ok();
+        return Ok(None);
+    };
+
+    let request: Option<(Uuid, String)> = sqlx::query_as(
+        r#"
+        SELECT id, new_email
+        FROM email_change_requests
+        WHERE token_hash = $1
+          AND user_id = $2
+          AND confirmed_at IS NULL
+          AND canceled_at IS NULL
+          AND expires_at > $3
+        FOR UPDATE
+        "#,
+    )
+    .bind(token_hash)
+    .bind(user_id)
+    .bind(now)
+    .fetch_optional(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "锁定邮箱变更请求失败").with_source(err))?;
+    let Some((request_id, new_email)) = request else {
+        tx.rollback().await.ok();
+        return Ok(None);
+    };
+
+    let changed = sqlx::query(
+        r#"
+        UPDATE users u
+        SET email = $2,
+            email_verified_at = $3,
+            token_version = token_version + 1,
+            updated_at = NOW()
+        WHERE u.id = $1
+          AND NOT EXISTS (
+            SELECT 1 FROM users other
+            WHERE other.email = $2 AND other.id <> u.id
+          )
+        "#,
+    )
+    .bind(user_id)
+    .bind(&new_email)
+    .bind(now)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| {
+        if matches!(&err, sqlx::Error::Database(db) if db.code().as_deref() == Some("23505")) {
+            AppError::new(ErrorCode::InvalidRequest, "新邮箱已被其他账号使用")
+        } else {
+            AppError::new(ErrorCode::Internal, "确认新邮箱失败").with_source(err)
+        }
+    })?;
+    if changed.rows_affected() == 0 {
+        return Err(AppError::new(
+            ErrorCode::InvalidRequest,
+            "新邮箱已被其他账号使用",
+        ));
+    }
+
+    sqlx::query(
+        "UPDATE email_change_requests SET confirmed_at = $2 WHERE id = $1 AND confirmed_at IS NULL AND canceled_at IS NULL",
+    )
+    .bind(request_id)
+    .bind(now)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "完成邮箱变更请求失败").with_source(err))?;
+    sqlx::query(
+        "UPDATE email_change_requests SET canceled_at = $2 WHERE user_id = $1 AND id <> $3 AND confirmed_at IS NULL AND canceled_at IS NULL",
+    )
+    .bind(user_id)
+    .bind(now)
+    .bind(request_id)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "撤销旧邮箱变更请求失败").with_source(err))?;
+    sqlx::query("UPDATE password_resets SET used_at = $2 WHERE user_id = $1 AND used_at IS NULL")
+        .bind(user_id)
+        .bind(now)
+        .execute(&mut *tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "撤销密码重置请求失败").with_source(err)
+        })?;
+    sqlx::query(
+        "UPDATE email_verifications SET verified_at = $2 WHERE user_id = $1 AND verified_at IS NULL",
+    )
+    .bind(user_id)
+    .bind(now)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "撤销旧邮箱验证请求失败").with_source(err))?;
+
+    tx.commit()
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "提交邮箱变更失败").with_source(err))?;
+    Ok(Some(EmailChangeConfirmation { old_email }))
+}
+
 #[derive(Debug, Deserialize)]
 struct ForgotPasswordRequest {
     email: String,
@@ -517,6 +677,12 @@ async fn forgot_password(
     )
     .await?;
 
+    let requested_email = req.email.to_lowercase();
+    let mut tx = state
+        .db
+        .begin()
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
     let user = sqlx::query_as::<_, UserRow>(
         r#"
         SELECT
@@ -530,19 +696,20 @@ async fn forgot_password(
           token_version
         FROM users
         WHERE email = $1
+        FOR UPDATE
         "#,
     )
-    .bind(req.email.to_lowercase())
-    .fetch_optional(&state.db)
+    .bind(&requested_email)
+    .fetch_optional(&mut *tx)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
 
-    if let Some(user) = user {
+    let delivery = if let Some(user) = user {
         let reset_token = credentials::generate_token();
         let token_hash = credentials::sha256_hex(&reset_token);
         let expires_at_db = Utc::now() + Duration::hours(1);
 
-        let _ = sqlx::query(
+        sqlx::query(
             r#"
             INSERT INTO password_resets (user_id, token_hash, expires_at)
             VALUES ($1, $2, $3)
@@ -551,16 +718,26 @@ async fn forgot_password(
         .bind(user.id)
         .bind(token_hash)
         .bind(expires_at_db)
-        .execute(&state.db)
-        .await;
+        .execute(&mut *tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "创建密码重置记录失败").with_source(err)
+        })?;
 
         let reset_url = format!(
             "{}/reset-password?token={}",
             state.config.public_base_url, reset_token
         );
+        Some((user.email, user.username, reset_url))
+    } else {
+        None
+    };
+    tx.commit().await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "提交密码找回事务失败").with_source(err)
+    })?;
 
-        let _ =
-            mail::send_password_reset_email(&state, &user.email, &user.username, &reset_url).await;
+    if let Some((email, username, reset_url)) = delivery {
+        let _ = mail::send_password_reset_email(&state, &email, &username, &reset_url).await;
     }
 
     Ok(Json(Envelope {
@@ -611,25 +788,23 @@ async fn reset_password(
 
     let token_hash = credentials::sha256_hex(&req.token);
     let now = Utc::now();
-    let token_exists: bool = sqlx::query_scalar(
+    let reset_user_id: Option<Uuid> = sqlx::query_scalar(
         r#"
-        SELECT EXISTS(
-          SELECT 1
+        SELECT user_id
         FROM password_resets
         WHERE token_hash = $1
           AND used_at IS NULL
           AND expires_at > $2
-        )
         "#,
     )
     .bind(&token_hash)
     .bind(now)
-    .fetch_one(&state.db)
+    .fetch_optional(&state.db)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "重置密码失败").with_source(err))?;
-    if !token_exists {
+    let Some(reset_user_id) = reset_user_id else {
         return Err(AppError::new(ErrorCode::InvalidToken, "Token 无效或已过期"));
-    }
+    };
 
     let password_hash = credentials::hash_password(&req.new_password).await?;
 
@@ -639,6 +814,16 @@ async fn reset_password(
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
 
+    let locked_user: Option<Uuid> =
+        sqlx::query_scalar("SELECT id FROM users WHERE id = $1 FOR UPDATE")
+            .bind(reset_user_id)
+            .fetch_optional(&mut *tx)
+            .await
+            .map_err(|err| AppError::new(ErrorCode::Internal, "锁定用户失败").with_source(err))?;
+    if locked_user.is_none() {
+        return Err(AppError::new(ErrorCode::InvalidToken, "Token 无效或已过期"));
+    }
+
     let user_id: Option<Uuid> = sqlx::query_scalar(
         r#"
         SELECT user_id
@@ -658,6 +843,9 @@ async fn reset_password(
     let Some(user_id) = user_id else {
         return Err(AppError::new(ErrorCode::InvalidToken, "Token 无效或已过期"));
     };
+    if user_id != reset_user_id {
+        return Err(AppError::new(ErrorCode::InvalidToken, "Token 无效或已过期"));
+    }
 
     sqlx::query(
         "UPDATE users SET password_hash = $1, token_version = token_version + 1, updated_at = NOW() WHERE id = $2",

src/api/user.rs

@@ -1,5 +1,6 @@
 use crate::api::context;
 use crate::api::envelope::Envelope;
+use crate::auth;
 use crate::error::{AppError, ErrorCode};
 use crate::services::billing;
 use crate::services::{credentials, mail, settings};
@@ -55,6 +56,7 @@ struct UserView {
     username: String,
     role: String,
     email_verified: bool,
+    pending_email: Option<String>,
 }
 
 #[derive(Debug, Serialize)]
@@ -83,13 +85,24 @@ async fn get_profile(
         username: String,
         role: String,
         email_verified_at: Option<DateTime<Utc>>,
+        pending_email: Option<String>,
     }
 
     let user = sqlx::query_as::<_, UserRow>(
         r#"
-        SELECT id, email, username, role::text AS role, email_verified_at
-        FROM users
-        WHERE id = $1
+        SELECT u.id, u.email, u.username, u.role::text AS role, u.email_verified_at,
+               (
+                 SELECT r.new_email
+                 FROM email_change_requests r
+                 WHERE r.user_id = u.id
+                   AND r.confirmed_at IS NULL
+                   AND r.canceled_at IS NULL
+                   AND r.expires_at > NOW()
+                 ORDER BY r.created_at DESC
+                 LIMIT 1
+               ) AS pending_email
+        FROM users u
+        WHERE u.id = $1
         "#,
     )
     .bind(user_id)
@@ -107,6 +120,7 @@ async fn get_profile(
             username: user.username,
             role: user.role,
             email_verified: user.email_verified_at.is_some() || !verification_required,
+            pending_email: user.pending_email,
         },
     }))
 }
@@ -115,12 +129,15 @@ async fn get_profile(
 struct UpdateProfileRequest {
     email: Option<String>,
     username: Option<String>,
+    current_password: Option<String>,
 }
 
 #[derive(Debug, Serialize)]
 struct UpdateProfileResponse {
     user: UserView,
     message: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    token: Option<String>,
 }
 
 async fn update_profile(
@@ -142,50 +159,71 @@ async fn update_profile(
         return Err(AppError::new(ErrorCode::InvalidRequest, "未提供可更新字段"));
     }
 
+    let verification_required = settings::email_verification_required(&state).await?;
+
+    let mut tx = state
+        .db
+        .begin()
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
+
     #[derive(Debug, FromRow)]
     struct UserRow {
         id: Uuid,
         email: String,
         username: String,
+        password_hash: String,
         role: String,
         email_verified_at: Option<DateTime<Utc>>,
+        token_version: i32,
+        pending_email: Option<String>,
     }
 
     let user = sqlx::query_as::<_, UserRow>(
         r#"
-        SELECT id, email, username, role::text AS role, email_verified_at
-        FROM users
-        WHERE id = $1
+        SELECT u.id, u.email, u.username, u.password_hash,
+               u.role::text AS role, u.email_verified_at, u.token_version,
+               (
+                 SELECT r.new_email
+                 FROM email_change_requests r
+                 WHERE r.user_id = u.id
+                   AND r.confirmed_at IS NULL
+                   AND r.canceled_at IS NULL
+                   AND r.expires_at > NOW()
+                 ORDER BY r.created_at DESC
+                 LIMIT 1
+               ) AS pending_email
+        FROM users u
+        WHERE u.id = $1
+        FOR UPDATE OF u
         "#,
     )
     .bind(user_id)
-    .fetch_one(&state.db)
+    .fetch_one(&mut *tx)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
 
-    let mut next_email = user.email.clone();
-    let mut next_username = user.username.clone();
-    let mut email_changed = false;
-    let verification_required = settings::email_verification_required(&state).await?;
-
-    if let Some(email) = req.email.as_ref() {
+    let next_email = match req.email.as_ref() {
+        Some(email) => {
             let email = email.trim().to_lowercase();
             credentials::validate_email(&email)?;
-        if email != user.email {
-            next_email = email;
-            email_changed = true;
+            email
         }
-    }
-
-    if let Some(username) = req.username.as_ref() {
+        None => user.email.clone(),
+    };
+    let next_username = match req.username.as_ref() {
+        Some(username) => {
             let username = username.trim().to_string();
             credentials::validate_username(&username)?;
-        if username != user.username {
-            next_username = username;
-        }
+            username
         }
+        None => user.username.clone(),
+    };
+    let email_changed = next_email != user.email;
+    let username_changed = next_username != user.username;
 
-    if next_email == user.email && next_username == user.username {
+    if !email_changed && !username_changed {
+        tx.rollback().await.ok();
         return Ok(Json(Envelope {
             success: true,
             data: UpdateProfileResponse {
@@ -195,46 +233,41 @@ async fn update_profile(
                     username: user.username,
                     role: user.role,
                     email_verified: user.email_verified_at.is_some() || !verification_required,
+                    pending_email: user.pending_email,
                 },
                 message: "暂无更新".to_string(),
+                token: None,
             },
         }));
     }
 
-    let mut tx = state
-        .db
-        .begin()
-        .await
-        .map_err(|err| AppError::new(ErrorCode::Internal, "开启事务失败").with_source(err))?;
+    if email_changed {
+        let current_password = req
+            .current_password
+            .as_deref()
+            .filter(|value| !value.is_empty())
+            .ok_or_else(|| AppError::new(ErrorCode::Unauthorized, "修改邮箱需要当前密码"))?;
+        if !credentials::verify_password(current_password, &user.password_hash).await? {
+            return Err(AppError::new(ErrorCode::Unauthorized, "当前密码不正确"));
+        }
 
-    let email_verified_at = if email_changed && verification_required {
-        None
-    } else if email_changed {
-        Some(Utc::now())
-    } else {
-        user.email_verified_at
-    };
-
-    let updated = sqlx::query_as::<_, UserRow>(
-        r#"
-        UPDATE users
-        SET email = $2,
-            username = $3,
-            email_verified_at = $4,
-            updated_at = NOW()
-        WHERE id = $1
-        RETURNING id, email, username, role::text AS role, email_verified_at
-        "#,
-    )
-    .bind(user_id)
+        let email_in_use: bool =
+            sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM users WHERE email = $1 AND id <> $2)")
                 .bind(&next_email)
-    .bind(&next_username)
-    .bind(email_verified_at)
+                .bind(user_id)
                 .fetch_one(&mut *tx)
                 .await
-    .map_err(map_unique_violation)?;
+                .map_err(|err| {
+                    AppError::new(ErrorCode::Internal, "检查邮箱失败").with_source(err)
+                })?;
+        if email_in_use {
+            return Err(AppError::new(ErrorCode::InvalidRequest, "邮箱已存在"));
+        }
+    }
 
     let mut verification_link: Option<String> = None;
+    let mut pending_email = user.pending_email.clone();
+    let updated: UserRow;
     if email_changed && verification_required {
         let token = credentials::generate_token();
         let token_hash = credentials::sha256_hex(&token);
@@ -242,43 +275,160 @@ async fn update_profile(
 
         sqlx::query(
             r#"
-            INSERT INTO email_verifications (user_id, token_hash, expires_at)
-            VALUES ($1, $2, $3)
+            UPDATE email_change_requests
+            SET canceled_at = NOW()
+            WHERE confirmed_at IS NULL
+              AND canceled_at IS NULL
+              AND (user_id = $1 OR (new_email = $2 AND expires_at <= NOW()))
             "#,
         )
         .bind(user_id)
+        .bind(&next_email)
+        .execute(&mut *tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "撤销旧邮箱变更请求失败").with_source(err)
+        })?;
+
+        sqlx::query(
+            r#"
+            INSERT INTO email_change_requests (user_id, new_email, token_hash, expires_at)
+            VALUES ($1, $2, $3, $4)
+            "#,
+        )
+        .bind(user_id)
+        .bind(&next_email)
         .bind(token_hash)
         .bind(expires_at)
         .execute(&mut *tx)
         .await
-        .map_err(|err| {
-            AppError::new(ErrorCode::Internal, "创建邮箱验证记录失败").with_source(err)
-        })?;
+        .map_err(map_unique_violation)?;
+
+        updated = sqlx::query_as::<_, UserRow>(
+            r#"
+            UPDATE users
+            SET username = $2, updated_at = NOW()
+            WHERE id = $1
+            RETURNING id, email, username, password_hash, role::text AS role,
+                      email_verified_at, token_version, $3::text AS pending_email
+            "#,
+        )
+        .bind(user_id)
+        .bind(&next_username)
+        .bind(&next_email)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(map_unique_violation)?;
+        pending_email = Some(next_email.clone());
 
         verification_link = Some(format!(
             "{}/verify-email?token={}",
             state.config.public_base_url, token
         ));
+    } else if email_changed {
+        updated = sqlx::query_as::<_, UserRow>(
+            r#"
+            UPDATE users
+            SET email = $2,
+                username = $3,
+                email_verified_at = NOW(),
+                token_version = token_version + 1,
+                updated_at = NOW()
+            WHERE id = $1
+            RETURNING id, email, username, password_hash, role::text AS role,
+                      email_verified_at, token_version, NULL::text AS pending_email
+            "#,
+        )
+        .bind(user_id)
+        .bind(&next_email)
+        .bind(&next_username)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(map_unique_violation)?;
+        sqlx::query(
+            "UPDATE email_change_requests SET canceled_at = NOW() WHERE user_id = $1 AND confirmed_at IS NULL AND canceled_at IS NULL",
+        )
+        .bind(user_id)
+        .execute(&mut *tx)
+        .await
+        .map_err(|err| AppError::new(ErrorCode::Internal, "撤销邮箱变更请求失败").with_source(err))?;
+        sqlx::query(
+            "UPDATE password_resets SET used_at = NOW() WHERE user_id = $1 AND used_at IS NULL",
+        )
+        .bind(user_id)
+        .execute(&mut *tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "撤销密码重置请求失败").with_source(err)
+        })?;
+        pending_email = None;
+    } else {
+        updated = sqlx::query_as::<_, UserRow>(
+            r#"
+            UPDATE users
+            SET username = $2, updated_at = NOW()
+            WHERE id = $1
+            RETURNING id, email, username, password_hash, role::text AS role,
+                      email_verified_at, token_version, $3::text AS pending_email
+            "#,
+        )
+        .bind(user_id)
+        .bind(&next_username)
+        .bind(pending_email.as_deref())
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(map_unique_violation)?;
     }
 
     tx.commit()
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "提交事务失败").with_source(err))?;
 
-    if let Some(link) = verification_link.as_deref() {
-        mail::send_verification_email(&state, &updated.email, &updated.username, link)
-            .await
-            .map_err(|err| {
-                AppError::new(ErrorCode::MailSendFailed, "验证邮件发送失败").with_source(err)
-            })?;
+    let verification_mail_sent = if let Some(link) = verification_link.as_deref() {
+        match mail::send_verification_email(&state, &next_email, &updated.username, link).await {
+            Ok(()) => true,
+            Err(err) => {
+                tracing::error!(user_id = %user_id, error = ?err, "email change verification delivery failed");
+                false
+            }
+        }
+    } else {
+        true
+    };
+
+    if email_changed && !verification_required {
+        if let Err(err) = mail::send_email_change_notice(&state, &user.email).await {
+            tracing::warn!(user_id = %user_id, error = ?err, "email change notice delivery failed");
+        }
     }
 
     let message = if email_changed && verification_required {
-        "资料已更新，请验证新邮箱".to_string()
+        if verification_mail_sent {
+            "资料已更新，请验证新邮箱；确认前仍使用原邮箱登录和找回密码".to_string()
+        } else {
+            "新邮箱已进入待确认状态，但验证邮件发送失败，请重新提交邮箱变更".to_string()
+        }
+    } else if email_changed {
+        "资料已更新，其他登录状态已失效".to_string()
     } else {
         "资料已更新".to_string()
     };
 
+    let token = if email_changed && !verification_required {
+        Some(
+            auth::issue_jwt(
+                &state.config.jwt_secret,
+                state.config.jwt_expiry_hours,
+                updated.id,
+                &updated.role,
+                updated.token_version,
+            )?
+            .0,
+        )
+    } else {
+        None
+    };
+
     Ok(Json(Envelope {
         success: true,
         data: UpdateProfileResponse {
@@ -288,8 +438,10 @@ async fn update_profile(
                 username: updated.username,
                 role: updated.role,
                 email_verified: updated.email_verified_at.is_some() || !verification_required,
+                pending_email,
             },
             message,
+            token,
         },
     }))
 }
@@ -909,3 +1061,453 @@ fn map_unique_violation(err: sqlx::Error) -> AppError {
     }
     AppError::new(ErrorCode::Internal, "数据库操作失败").with_source(err)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::Config;
+    use crate::services::mail::Mailer;
+    use axum::body::{to_bytes, Body};
+    use axum::http::{Method, Request, StatusCode};
+    use sqlx::postgres::PgPoolOptions;
+    use std::sync::Arc;
+    use tokio::sync::Semaphore;
+    use tower::ServiceExt;
+
+    async fn json_request(
+        app: &axum::Router,
+        method: Method,
+        uri: &str,
+        token: Option<&str>,
+        payload: serde_json::Value,
+    ) -> (StatusCode, serde_json::Value) {
+        let mut builder = Request::builder()
+            .method(method)
+            .uri(uri)
+            .header(axum::http::header::CONTENT_TYPE, "application/json");
+        if let Some(token) = token {
+            builder = builder.header(axum::http::header::AUTHORIZATION, format!("Bearer {token}"));
+        }
+        let mut request = builder
+            .body(Body::from(payload.to_string()))
+            .expect("build test request");
+        request.extensions_mut().insert(ConnectInfo(
+            "127.0.0.1:41000"
+                .parse::<SocketAddr>()
+                .expect("parse test address"),
+        ));
+        let response = app
+            .clone()
+            .oneshot(request)
+            .await
+            .expect("execute test request");
+        let status = response.status();
+        let body = to_bytes(response.into_body(), usize::MAX)
+            .await
+            .expect("read test response");
+        let json = serde_json::from_slice(&body)
+            .unwrap_or_else(|_| panic!("response is not JSON: {}", String::from_utf8_lossy(&body)));
+        (status, json)
+    }
+
+    async fn replace_pending_token(pool: &sqlx::PgPool, user_id: Uuid, token: &str) {
+        let updated = sqlx::query(
+            r#"
+            UPDATE email_change_requests
+            SET token_hash = $2
+            WHERE user_id = $1
+              AND confirmed_at IS NULL
+              AND canceled_at IS NULL
+            "#,
+        )
+        .bind(user_id)
+        .bind(credentials::sha256_hex(token))
+        .execute(pool)
+        .await
+        .expect("replace pending email token");
+        assert_eq!(updated.rows_affected(), 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[ignore = "requires isolated IMAGEFORGE_TEST_DATABASE_URL and IMAGEFORGE_TEST_REDIS_URL"]
+    async fn pending_email_cannot_take_over_user_or_admin_recovery() {
+        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");
+        sqlx::query(
+            r#"
+            INSERT INTO system_config (key, value, description)
+            VALUES (
+              'auth_config',
+              '{"email_verification_required":true}'::jsonb,
+              'security integration test'
+            )
+            ON CONFLICT (key) DO UPDATE
+            SET value = EXCLUDED.value, updated_at = NOW()
+            "#,
+        )
+        .execute(&pool)
+        .await
+        .expect("enable email verification");
+
+        let mut config = Config::from_env().expect("load test config");
+        config.database_url = database_url;
+        config.redis_url = redis_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");
+        let state = 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.clone(),
+            redis,
+        };
+        let app = axum::Router::new()
+            .nest("/auth", crate::api::auth::router())
+            .merge(router())
+            .with_state(state.clone());
+
+        let marker = Uuid::new_v4().simple().to_string();
+        let password = "Original9!";
+        let password_hash = credentials::hash_password(password)
+            .await
+            .expect("hash test password");
+        let user_id = Uuid::new_v4();
+        let old_email = format!("old-{marker}@example.test");
+        let pending_email = format!("pending-{marker}@example.test");
+        sqlx::query(
+            r#"
+            INSERT INTO users (
+              id, email, username, password_hash, email_verified_at
+            ) VALUES ($1, $2, $3, $4, NOW())
+            "#,
+        )
+        .bind(user_id)
+        .bind(&old_email)
+        .bind(format!("user_{marker}"))
+        .bind(&password_hash)
+        .execute(&pool)
+        .await
+        .expect("insert test user");
+        let (old_token, _) = auth::issue_jwt(
+            &state.config.jwt_secret,
+            state.config.jwt_expiry_hours,
+            user_id,
+            "user",
+            0,
+        )
+        .expect("issue test jwt");
+
+        let (status, _) = json_request(
+            &app,
+            Method::PUT,
+            "/user/profile",
+            Some(&old_token),
+            serde_json::json!({ "email": pending_email }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::UNAUTHORIZED, "JWT alone changed email");
+        let (status, _) = json_request(
+            &app,
+            Method::PUT,
+            "/user/profile",
+            Some(&old_token),
+            serde_json::json!({
+                "email": pending_email,
+                "current_password": "WrongPassword9!"
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::UNAUTHORIZED);
+        let (status, response) = json_request(
+            &app,
+            Method::PUT,
+            "/user/profile",
+            Some(&old_token),
+            serde_json::json!({
+                "email": pending_email,
+                "current_password": password
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{response}");
+        assert_eq!(response["data"]["user"]["email"], old_email);
+        assert_eq!(response["data"]["user"]["pending_email"], pending_email);
+        let persisted_email: String = sqlx::query_scalar("SELECT email FROM users WHERE id = $1")
+            .bind(user_id)
+            .fetch_one(&pool)
+            .await
+            .expect("query persisted email");
+        assert_eq!(persisted_email, old_email);
+
+        let (status, _) = json_request(
+            &app,
+            Method::POST,
+            "/auth/forgot-password",
+            None,
+            serde_json::json!({ "email": pending_email }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK);
+        let active_resets: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM password_resets WHERE user_id = $1 AND used_at IS NULL",
+        )
+        .bind(user_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count pending-email resets");
+        assert_eq!(active_resets, 0, "pending email became a recovery address");
+
+        let (status, _) = json_request(
+            &app,
+            Method::POST,
+            "/auth/forgot-password",
+            None,
+            serde_json::json!({ "email": old_email }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK);
+        let active_resets: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM password_resets WHERE user_id = $1 AND used_at IS NULL",
+        )
+        .bind(user_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count old-email resets");
+        assert_eq!(
+            active_resets, 1,
+            "old email stopped being the recovery address early"
+        );
+
+        let confirm_token = format!("confirm-{marker}");
+        replace_pending_token(&pool, user_id, &confirm_token).await;
+        let (status, response) = json_request(
+            &app,
+            Method::POST,
+            "/auth/verify-email",
+            None,
+            serde_json::json!({ "token": confirm_token }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{response}");
+        assert_eq!(response["data"]["session_invalidated"], true);
+        let active_resets: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM password_resets WHERE user_id = $1 AND used_at IS NULL",
+        )
+        .bind(user_id)
+        .fetch_one(&pool)
+        .await
+        .expect("count invalidated resets");
+        assert_eq!(active_resets, 0);
+        let (status, _) = json_request(
+            &app,
+            Method::GET,
+            "/user/profile",
+            Some(&old_token),
+            serde_json::json!({}),
+        )
+        .await;
+        assert_eq!(status, StatusCode::UNAUTHORIZED, "old JWT remained valid");
+        let (status, response) = json_request(
+            &app,
+            Method::POST,
+            "/auth/login",
+            None,
+            serde_json::json!({ "email": pending_email, "password": password }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{response}");
+
+        let admin_id = Uuid::new_v4();
+        let admin_old_email = format!("admin-old-{marker}@example.test");
+        let admin_pending_email = format!("admin-pending-{marker}@example.test");
+        sqlx::query(
+            r#"
+            INSERT INTO users (
+              id, email, username, password_hash, role, email_verified_at
+            ) VALUES ($1, $2, $3, $4, 'admin', NOW())
+            "#,
+        )
+        .bind(admin_id)
+        .bind(&admin_old_email)
+        .bind(format!("admin_{marker}"))
+        .bind(&password_hash)
+        .execute(&pool)
+        .await
+        .expect("insert test admin");
+        let (admin_token, _) = auth::issue_jwt(
+            &state.config.jwt_secret,
+            state.config.jwt_expiry_hours,
+            admin_id,
+            "admin",
+            0,
+        )
+        .expect("issue admin jwt");
+        let (status, _) = json_request(
+            &app,
+            Method::PUT,
+            "/user/profile",
+            Some(&admin_token),
+            serde_json::json!({ "email": admin_pending_email }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::UNAUTHORIZED);
+        let (status, response) = json_request(
+            &app,
+            Method::PUT,
+            "/user/profile",
+            Some(&admin_token),
+            serde_json::json!({
+                "email": admin_pending_email,
+                "current_password": password
+            }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{response}");
+        assert_eq!(response["data"]["user"]["email"], admin_old_email);
+        let admin_confirm_token = format!("admin-confirm-{marker}");
+        replace_pending_token(&pool, admin_id, &admin_confirm_token).await;
+        let (status, response) = json_request(
+            &app,
+            Method::POST,
+            "/auth/verify-email",
+            None,
+            serde_json::json!({ "token": admin_confirm_token }),
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{response}");
+        let (status, _) = json_request(
+            &app,
+            Method::GET,
+            "/user/profile",
+            Some(&admin_token),
+            serde_json::json!({}),
+        )
+        .await;
+        assert_eq!(status, StatusCode::UNAUTHORIZED);
+        let admin: (String, String) =
+            sqlx::query_as("SELECT email, role::text FROM users WHERE id = $1")
+                .bind(admin_id)
+                .fetch_one(&pool)
+                .await
+                .expect("query test admin");
+        assert_eq!(admin, (admin_pending_email, "admin".to_string()));
+
+        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");
+        let race_confirm_token = format!("race-confirm-{marker}");
+        sqlx::query(
+            r#"
+            INSERT INTO users (
+              id, email, username, password_hash, email_verified_at
+            ) VALUES ($1, $2, $3, $4, NOW())
+            "#,
+        )
+        .bind(race_user_id)
+        .bind(&race_old_email)
+        .bind(format!("race_{marker}"))
+        .bind(&password_hash)
+        .execute(&pool)
+        .await
+        .expect("insert recovery race user");
+        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(race_user_id)
+        .bind(&race_new_email)
+        .bind(credentials::sha256_hex(&race_confirm_token))
+        .execute(&pool)
+        .await
+        .expect("insert recovery race email change");
+
+        let mut blocker = pool.begin().await.expect("begin recovery race blocker");
+        let _: Uuid = sqlx::query_scalar("SELECT id FROM users WHERE id = $1 FOR UPDATE")
+            .bind(race_user_id)
+            .fetch_one(&mut *blocker)
+            .await
+            .expect("lock recovery race user");
+        let verify_join = {
+            let app = app.clone();
+            let token = race_confirm_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(100)).await;
+        let forgot_join = {
+            let app = app.clone();
+            let email = race_old_email.clone();
+            tokio::spawn(async move {
+                json_request(
+                    &app,
+                    Method::POST,
+                    "/auth/forgot-password",
+                    None,
+                    serde_json::json!({ "email": email }),
+                )
+                .await
+            })
+        };
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        blocker.commit().await.expect("release recovery race user");
+        let (verify_status, verify_response) =
+            verify_join.await.expect("join racing email confirmation");
+        let (forgot_status, forgot_response) =
+            forgot_join.await.expect("join racing password recovery");
+        assert_eq!(verify_status, StatusCode::OK, "{verify_response}");
+        assert_eq!(forgot_status, StatusCode::OK, "{forgot_response}");
+        let race_result: (String, i64) = sqlx::query_as(
+            r#"
+            SELECT u.email,
+                   COUNT(r.id) FILTER (WHERE r.used_at IS NULL)
+            FROM users u
+            LEFT JOIN password_resets r ON r.user_id = u.id
+            WHERE u.id = $1
+            GROUP BY u.email
+            "#,
+        )
+        .bind(race_user_id)
+        .fetch_one(&pool)
+        .await
+        .expect("query recovery race result");
+        assert_eq!(race_result, (race_new_email, 0));
+
+        sqlx::query("DELETE FROM users WHERE id IN ($1, $2, $3)")
+            .bind(user_id)
+            .bind(admin_id)
+            .bind(race_user_id)
+            .execute(&pool)
+            .await
+            .expect("delete account recovery test users");
+    }
+}

src/api/webhooks.rs

@@ -12,20 +12,22 @@ use chrono::{TimeZone, Utc};
 use hmac::{Hmac, Mac};
 use serde::Deserialize;
 use sha2::Sha256;
+use sqlx::{Postgres, Transaction};
 
 pub fn router() -> Router<AppState> {
     Router::new().route("/webhooks/stripe", post(stripe_webhook))
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 struct StripeEvent {
     id: String,
+    created: i64,
     #[serde(rename = "type")]
     type_: String,
     data: StripeEventData,
 }
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 struct StripeEventData {
     object: serde_json::Value,
 }
@@ -99,7 +101,7 @@ async fn stripe_webhook(
         ));
     }
 
-    if let Err(err) = process_stripe_event(&state, &event).await {
+    if let Err(err) = process_claimed_stripe_event(&state, &event).await {
         let _ = sqlx::query(
             "UPDATE webhook_events SET status = 'failed', error_message = $2, processed_at = NULL WHERE provider = 'stripe' AND provider_event_id = $1 AND status = 'processing'",
         )
@@ -111,19 +113,64 @@ async fn stripe_webhook(
         return Err(err);
     }
 
-    let _ = sqlx::query(
-        "UPDATE webhook_events SET status = 'processed', processed_at = NOW() WHERE provider = 'stripe' AND provider_event_id = $1",
-    )
-    .bind(&event.id)
-    .execute(&state.db)
-    .await;
-
     Ok(Json(Envelope {
         success: true,
         data: serde_json::json!({ "status": "ok" }),
     }))
 }
 
+async fn process_claimed_stripe_event(
+    state: &AppState,
+    event: &StripeEvent,
+) -> Result<(), AppError> {
+    let mut tx = state.db.begin().await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "开启 Webhook 事务失败").with_source(err)
+    })?;
+    let status: Option<String> = sqlx::query_scalar(
+        r#"
+        SELECT status
+        FROM webhook_events
+        WHERE provider = 'stripe' AND provider_event_id = $1
+        FOR UPDATE
+        "#,
+    )
+    .bind(&event.id)
+    .fetch_optional(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "锁定 Webhook 失败").with_source(err))?;
+    if status.as_deref() != Some("processing") {
+        return Err(AppError::new(
+            ErrorCode::IdempotencyConflict,
+            "Webhook 事件未处于可处理状态",
+        ));
+    }
+
+    process_stripe_event(&mut tx, event).await?;
+    let updated = sqlx::query(
+        r#"
+        UPDATE webhook_events
+        SET status = 'processed', processed_at = NOW(), error_message = NULL
+        WHERE provider = 'stripe'
+          AND provider_event_id = $1
+          AND status = 'processing'
+        "#,
+    )
+    .bind(&event.id)
+    .execute(&mut *tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "更新 Webhook 状态失败").with_source(err))?;
+    if updated.rows_affected() != 1 {
+        return Err(AppError::new(
+            ErrorCode::IdempotencyConflict,
+            "Webhook 处理租约已失效",
+        ));
+    }
+    tx.commit().await.map_err(|err| {
+        AppError::new(ErrorCode::Internal, "提交 Webhook 事务失败").with_source(err)
+    })?;
+    Ok(())
+}
+
 fn verify_stripe_signature(payload: &[u8], sig_header: &str, secret: &str) -> Result<(), AppError> {
     let mut timestamp: Option<i64> = None;
     let mut signatures = Vec::<String>::new();
@@ -186,24 +233,25 @@ fn secure_eq(a: &str, b: &str) -> bool {
     out == 0
 }
 
-async fn process_stripe_event(state: &AppState, event: &StripeEvent) -> Result<(), AppError> {
+async fn process_stripe_event(
+    tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
+) -> Result<(), AppError> {
     match event.type_.as_str() {
         "checkout.session.completed" => {
-            map_checkout_session_completed(state, &event.data.object).await
+            map_checkout_session_completed(tx, &event.data.object).await
         }
         "customer.subscription.created" | "customer.subscription.updated" => {
-            upsert_subscription(state, &event.data.object).await
-        }
-        "customer.subscription.deleted" => cancel_subscription(state, &event.data.object).await,
-        "invoice.paid" | "invoice.payment_failed" => {
-            upsert_invoice(state, &event.data.object).await
+            upsert_subscription(tx, event, &event.data.object).await
         }
+        "customer.subscription.deleted" => cancel_subscription(tx, event, &event.data.object).await,
+        "invoice.paid" | "invoice.payment_failed" => upsert_invoice(tx, &event.data.object).await,
         _ => Ok(()),
     }
 }
 
 async fn map_checkout_session_completed(
-    state: &AppState,
+    tx: &mut Transaction<'_, Postgres>,
     object: &serde_json::Value,
 ) -> Result<(), AppError> {
     let customer_id = object
@@ -242,7 +290,7 @@ async fn map_checkout_session_completed(
     )
     .bind(user_id)
     .bind(customer_id)
-    .execute(&state.db)
+    .execute(&mut **tx)
     .await
     .map_err(|err| {
         AppError::new(ErrorCode::Internal, "更新 Stripe Customer 映射失败").with_source(err)
@@ -253,7 +301,7 @@ async fn map_checkout_session_completed(
             "SELECT billing_customer_id FROM users WHERE id = $1",
         )
         .bind(user_id)
-        .fetch_optional(&state.db)
+        .fetch_optional(&mut **tx)
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?
         .flatten();
@@ -275,11 +323,72 @@ async fn map_checkout_session_completed(
     Ok(())
 }
 
-async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> {
+fn subscription_event_rank(event_type: &str) -> i16 {
+    match event_type {
+        "customer.subscription.deleted" => 2,
+        "customer.subscription.updated" => 1,
+        _ => 0,
+    }
+}
+
+async fn claim_subscription_event(
+    tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
+    provider_subscription_id: &str,
+    is_deleted: bool,
+) -> Result<bool, AppError> {
+    let claimed: Option<String> = sqlx::query_scalar(
+        r#"
+        INSERT INTO provider_object_event_watermarks (
+          provider, object_type, provider_object_id,
+          last_event_created, last_event_rank, last_event_id, is_deleted
+        ) VALUES (
+          'stripe', 'subscription', $1,
+          $2, $3, $4, $5
+        )
+        ON CONFLICT (provider, object_type, provider_object_id) DO UPDATE
+        SET last_event_created = EXCLUDED.last_event_created,
+            last_event_rank = EXCLUDED.last_event_rank,
+            last_event_id = EXCLUDED.last_event_id,
+            is_deleted = EXCLUDED.is_deleted,
+            updated_at = NOW()
+        WHERE (
+          EXCLUDED.last_event_created,
+          EXCLUDED.last_event_rank,
+          EXCLUDED.last_event_id
+        ) > (
+          provider_object_event_watermarks.last_event_created,
+          provider_object_event_watermarks.last_event_rank,
+          provider_object_event_watermarks.last_event_id
+        )
+        RETURNING provider_object_id
+        "#,
+    )
+    .bind(provider_subscription_id)
+    .bind(event.created)
+    .bind(subscription_event_rank(&event.type_))
+    .bind(&event.id)
+    .bind(is_deleted)
+    .fetch_optional(&mut **tx)
+    .await
+    .map_err(|err| {
+        AppError::new(ErrorCode::Internal, "更新 Stripe 订阅事件水位失败").with_source(err)
+    })?;
+    Ok(claimed.is_some())
+}
+
+async fn upsert_subscription(
+    tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
+    object: &serde_json::Value,
+) -> Result<(), AppError> {
     let provider_subscription_id = object
         .get("id")
         .and_then(|v| v.as_str())
         .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?;
+    if !claim_subscription_event(tx, event, provider_subscription_id, false).await? {
+        return Ok(());
+    }
     let provider_customer_id = object
         .get("customer")
         .and_then(|v| v.as_str())
@@ -320,7 +429,7 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re
     let user_id: Option<uuid::Uuid> =
         sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1")
             .bind(provider_customer_id)
-            .fetch_optional(&state.db)
+            .fetch_optional(&mut **tx)
             .await
             .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
 
@@ -332,7 +441,7 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re
     let plan_id: Option<uuid::Uuid> =
         sqlx::query_scalar("SELECT id FROM plans WHERE stripe_price_id = $1 LIMIT 1")
             .bind(price_id)
-            .fetch_optional(&state.db)
+            .fetch_optional(&mut **tx)
             .await
             .map_err(|err| AppError::new(ErrorCode::Internal, "查询套餐失败").with_source(err))?;
 
@@ -341,48 +450,34 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re
         return Ok(());
     };
 
-    let updated: Option<uuid::Uuid> = sqlx::query_scalar(
-        r#"
-        UPDATE subscriptions
-        SET user_id = $1,
-            plan_id = $2,
-            status = $3::subscription_status,
-            current_period_start = $4,
-            current_period_end = $5,
-            cancel_at_period_end = $6,
-            provider = 'stripe',
-            provider_customer_id = $7,
-            updated_at = NOW()
-        WHERE provider = 'stripe' AND provider_subscription_id = $8
-        RETURNING id
-        "#,
-    )
-    .bind(user_id)
-    .bind(plan_id)
-    .bind(mapped_status)
-    .bind(current_period_start)
-    .bind(current_period_end)
-    .bind(cancel_at_period_end)
-    .bind(provider_customer_id)
-    .bind(provider_subscription_id)
-    .fetch_optional(&state.db)
-    .await
-    .map_err(|err| AppError::new(ErrorCode::Internal, "更新订阅失败").with_source(err))?;
-
-    if updated.is_none() {
-        let _ = sqlx::query(
+    sqlx::query(
         r#"
         INSERT INTO subscriptions (
           user_id, plan_id, status,
           current_period_start, current_period_end,
-              cancel_at_period_end,
+          cancel_at_period_end, canceled_at,
           provider, provider_customer_id, provider_subscription_id
         ) VALUES (
           $1, $2, $3::subscription_status,
           $4, $5,
           $6,
+          CASE WHEN $3 = 'canceled' THEN NOW() ELSE NULL END,
           'stripe', $7, $8
         )
+        ON CONFLICT (provider, provider_subscription_id) DO UPDATE
+        SET user_id = EXCLUDED.user_id,
+            plan_id = EXCLUDED.plan_id,
+            status = EXCLUDED.status,
+            current_period_start = EXCLUDED.current_period_start,
+            current_period_end = EXCLUDED.current_period_end,
+            cancel_at_period_end = EXCLUDED.cancel_at_period_end,
+            canceled_at = CASE
+              WHEN EXCLUDED.status = 'canceled'::subscription_status
+                THEN COALESCE(subscriptions.canceled_at, NOW())
+              ELSE NULL
+            END,
+            provider_customer_id = EXCLUDED.provider_customer_id,
+            updated_at = NOW()
         "#,
     )
     .bind(user_id)
@@ -393,38 +488,141 @@ async fn upsert_subscription(state: &AppState, object: &serde_json::Value) -> Re
     .bind(cancel_at_period_end)
     .bind(provider_customer_id)
     .bind(provider_subscription_id)
-        .execute(&state.db)
+    .execute(&mut **tx)
     .await
-        .map_err(|err| AppError::new(ErrorCode::Internal, "创建订阅失败").with_source(err))?;
-    }
+    .map_err(|err| AppError::new(ErrorCode::Internal, "写入订阅失败").with_source(err))?;
 
     Ok(())
 }
 
-async fn cancel_subscription(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> {
+async fn cancel_subscription(
+    tx: &mut Transaction<'_, Postgres>,
+    event: &StripeEvent,
+    object: &serde_json::Value,
+) -> Result<(), AppError> {
     let provider_subscription_id = object
         .get("id")
         .and_then(|v| v.as_str())
         .ok_or_else(|| AppError::new(ErrorCode::InvalidRequest, "subscription.id 缺失"))?;
+    if !claim_subscription_event(tx, event, provider_subscription_id, true).await? {
+        return Ok(());
+    }
+    let canceled_at = object
+        .get("canceled_at")
+        .and_then(|v| v.as_i64())
+        .unwrap_or(event.created);
+    let canceled_at = Utc
+        .timestamp_opt(canceled_at, 0)
+        .single()
+        .unwrap_or_else(Utc::now);
 
-    let _ = sqlx::query(
+    let updated = sqlx::query(
         r#"
         UPDATE subscriptions
         SET status = 'canceled',
             cancel_at_period_end = false,
-            canceled_at = NOW(),
+            canceled_at = $2,
             updated_at = NOW()
         WHERE provider = 'stripe' AND provider_subscription_id = $1
         "#,
     )
     .bind(provider_subscription_id)
-    .execute(&state.db)
-    .await;
+    .bind(canceled_at)
+    .execute(&mut **tx)
+    .await
+    .map_err(|err| AppError::new(ErrorCode::Internal, "取消订阅失败").with_source(err))?;
+
+    if updated.rows_affected() == 0 {
+        let Some(provider_customer_id) = object
+            .get("customer")
+            .and_then(|value| value.as_str())
+            .filter(|value| !value.trim().is_empty())
+        else {
+            tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription has no customer mapping; watermark retained");
+            return Ok(());
+        };
+        let Some(price_id) = object
+            .pointer("/items/data/0/price/id")
+            .and_then(|value| value.as_str())
+            .or_else(|| {
+                object
+                    .pointer("/items/data/0/plan/id")
+                    .and_then(|value| value.as_str())
+            })
+        else {
+            tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription has no price mapping; watermark retained");
+            return Ok(());
+        };
+        let user_id: Option<uuid::Uuid> =
+            sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1")
+                .bind(provider_customer_id)
+                .fetch_optional(&mut **tx)
+                .await
+                .map_err(|err| {
+                    AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err)
+                })?;
+        let plan_id: Option<uuid::Uuid> =
+            sqlx::query_scalar("SELECT id FROM plans WHERE stripe_price_id = $1 LIMIT 1")
+                .bind(price_id)
+                .fetch_optional(&mut **tx)
+                .await
+                .map_err(|err| {
+                    AppError::new(ErrorCode::Internal, "查询套餐失败").with_source(err)
+                })?;
+        let (Some(user_id), Some(plan_id)) = (user_id, plan_id) else {
+            tracing::warn!(subscription = %provider_subscription_id, "deleted stripe subscription is not mapped locally; watermark retained");
+            return Ok(());
+        };
+        let period_start = object
+            .get("current_period_start")
+            .and_then(|value| value.as_i64())
+            .and_then(|value| Utc.timestamp_opt(value, 0).single())
+            .unwrap_or(canceled_at);
+        let period_end = object
+            .get("current_period_end")
+            .and_then(|value| value.as_i64())
+            .and_then(|value| Utc.timestamp_opt(value, 0).single())
+            .unwrap_or(canceled_at);
+
+        sqlx::query(
+            r#"
+            INSERT INTO subscriptions (
+              user_id, plan_id, status,
+              current_period_start, current_period_end,
+              cancel_at_period_end, canceled_at,
+              provider, provider_customer_id, provider_subscription_id
+            ) VALUES (
+              $1, $2, 'canceled', $3, $4, false, $5,
+              'stripe', $6, $7
+            )
+            ON CONFLICT (provider, provider_subscription_id) DO UPDATE
+            SET status = 'canceled',
+                cancel_at_period_end = false,
+                canceled_at = EXCLUDED.canceled_at,
+                updated_at = NOW()
+            "#,
+        )
+        .bind(user_id)
+        .bind(plan_id)
+        .bind(period_start)
+        .bind(period_end)
+        .bind(canceled_at)
+        .bind(provider_customer_id)
+        .bind(provider_subscription_id)
+        .execute(&mut **tx)
+        .await
+        .map_err(|err| {
+            AppError::new(ErrorCode::Internal, "创建已取消订阅 tombstone 失败").with_source(err)
+        })?;
+    }
 
     Ok(())
 }
 
-async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<(), AppError> {
+async fn upsert_invoice(
+    tx: &mut Transaction<'_, Postgres>,
+    object: &serde_json::Value,
+) -> Result<(), AppError> {
     let provider_invoice_id = object
         .get("id")
         .and_then(|v| v.as_str())
@@ -437,7 +635,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<
     let user_id: Option<uuid::Uuid> =
         sqlx::query_scalar("SELECT id FROM users WHERE billing_customer_id = $1 LIMIT 1")
             .bind(provider_customer_id)
-            .fetch_optional(&state.db)
+            .fetch_optional(&mut **tx)
             .await
             .map_err(|err| AppError::new(ErrorCode::Internal, "查询用户失败").with_source(err))?;
 
@@ -511,7 +709,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<
     .bind(period_end)
     .bind(paid_at)
     .bind(provider_invoice_id)
-    .execute(&state.db)
+    .execute(&mut **tx)
     .await
     .map_err(|err| AppError::new(ErrorCode::Internal, "更新发票失败").with_source(err))?;
 
@@ -543,7 +741,7 @@ async fn upsert_invoice(state: &AppState, object: &serde_json::Value) -> Result<
         .bind(hosted_invoice_url.as_deref())
         .bind(pdf_url.as_deref())
         .bind(paid_at)
-        .execute(&state.db)
+        .execute(&mut **tx)
         .await
         .map_err(|err| AppError::new(ErrorCode::Internal, "创建发票失败").with_source(err))?;
     }
@@ -585,6 +783,10 @@ fn map_invoice_status(status: &str) -> &'static str {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use sqlx::postgres::PgPoolOptions;
+    use std::sync::Arc;
+    use tokio::sync::Barrier;
+    use uuid::Uuid;
 
     #[test]
     fn truncate_preserves_utf8_boundaries() {
@@ -592,4 +794,250 @@ mod tests {
         assert_eq!(truncate("abc中文".to_string(), 5), "abc");
         assert_eq!(truncate("short".to_string(), 20), "short");
     }
+
+    fn subscription_event(
+        event_id: &str,
+        event_type: &str,
+        created: i64,
+        subscription_id: &str,
+        customer_id: &str,
+        price_id: &str,
+    ) -> StripeEvent {
+        let status = if event_type == "customer.subscription.deleted" {
+            "canceled"
+        } else {
+            "active"
+        };
+        StripeEvent {
+            id: event_id.to_string(),
+            created,
+            type_: event_type.to_string(),
+            data: StripeEventData {
+                object: serde_json::json!({
+                    "id": subscription_id,
+                    "customer": customer_id,
+                    "status": status,
+                    "current_period_start": 1_700_000_000_i64,
+                    "current_period_end": 1_702_592_000_i64,
+                    "cancel_at_period_end": false,
+                    "canceled_at": if status == "canceled" { Some(created) } else { None },
+                    "items": { "data": [{ "price": { "id": price_id } }] }
+                }),
+            },
+        }
+    }
+
+    async fn apply_test_event(pool: &sqlx::PgPool, event: &StripeEvent) {
+        let mut tx = pool.begin().await.expect("begin event transaction");
+        process_stripe_event(&mut tx, event)
+            .await
+            .expect("apply stripe event");
+        tx.commit().await.expect("commit stripe event");
+    }
+
+    async fn assert_canceled_once(pool: &sqlx::PgPool, subscription_id: &str) {
+        let rows: Vec<(String,)> = sqlx::query_as(
+            "SELECT status::text FROM subscriptions WHERE provider = 'stripe' AND provider_subscription_id = $1",
+        )
+        .bind(subscription_id)
+        .fetch_all(pool)
+        .await
+        .expect("query subscription");
+        assert_eq!(rows, vec![("canceled".to_string(),)]);
+
+        let watermark: (bool,) = sqlx::query_as(
+            "SELECT is_deleted FROM provider_object_event_watermarks WHERE provider = 'stripe' AND object_type = 'subscription' AND provider_object_id = $1",
+        )
+        .bind(subscription_id)
+        .fetch_one(pool)
+        .await
+        .expect("query subscription watermark");
+        assert!(watermark.0);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[ignore = "requires an isolated IMAGEFORGE_TEST_DATABASE_URL containing 'test'"]
+    async fn subscription_events_are_monotonic_for_all_orders_and_concurrency() {
+        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 pool = PgPoolOptions::new()
+            .max_connections(16)
+            .connect(&database_url)
+            .await
+            .expect("connect test database");
+        sqlx::migrate!().run(&pool).await.expect("run migrations");
+
+        let marker = Uuid::new_v4().simple().to_string();
+        let customer_id = format!("cus_test_{marker}");
+        let price_id = format!("price_test_{marker}");
+        let user_id = Uuid::new_v4();
+        let plan_id = Uuid::new_v4();
+        sqlx::query(
+            r#"
+            INSERT INTO users (id, email, username, password_hash, billing_customer_id)
+            VALUES ($1, $2, $3, 'test-only', $4)
+            "#,
+        )
+        .bind(user_id)
+        .bind(format!("stripe-{marker}@example.test"))
+        .bind(format!("stripe_{marker}"))
+        .bind(&customer_id)
+        .execute(&pool)
+        .await
+        .expect("insert test user");
+        sqlx::query(
+            r#"
+            INSERT INTO plans (
+              id, code, name, stripe_price_id,
+              included_units_per_period, max_file_size_mb,
+              max_files_per_batch, concurrency_limit, retention_days
+            ) VALUES ($1, $2, 'Stripe ordering test', $3, 10, 10, 10, 1, 1)
+            "#,
+        )
+        .bind(plan_id)
+        .bind(format!("stripe_test_{marker}"))
+        .bind(&price_id)
+        .execute(&pool)
+        .await
+        .expect("insert test plan");
+
+        let permutations = [
+            [0, 1, 2],
+            [0, 2, 1],
+            [1, 0, 2],
+            [1, 2, 0],
+            [2, 0, 1],
+            [2, 1, 0],
+        ];
+        for (case, order) in permutations.into_iter().enumerate() {
+            let subscription_id = format!("sub_{marker}_perm_{case}");
+            let events = [
+                subscription_event(
+                    &format!("evt_{marker}_{case}_created"),
+                    "customer.subscription.created",
+                    1_700_000_100,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+                subscription_event(
+                    &format!("evt_{marker}_{case}_updated"),
+                    "customer.subscription.updated",
+                    1_700_000_200,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+                subscription_event(
+                    &format!("evt_{marker}_{case}_deleted"),
+                    "customer.subscription.deleted",
+                    1_700_000_300,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+            ];
+            for index in order {
+                apply_test_event(&pool, &events[index]).await;
+            }
+            assert_canceled_once(&pool, &subscription_id).await;
+        }
+
+        let same_second_id = format!("sub_{marker}_same_second");
+        for event in [
+            subscription_event(
+                &format!("evt_{marker}_same_deleted"),
+                "customer.subscription.deleted",
+                1_700_000_400,
+                &same_second_id,
+                &customer_id,
+                &price_id,
+            ),
+            subscription_event(
+                &format!("evt_{marker}_same_updated"),
+                "customer.subscription.updated",
+                1_700_000_400,
+                &same_second_id,
+                &customer_id,
+                &price_id,
+            ),
+            subscription_event(
+                &format!("evt_{marker}_same_created"),
+                "customer.subscription.created",
+                1_700_000_400,
+                &same_second_id,
+                &customer_id,
+                &price_id,
+            ),
+        ] {
+            apply_test_event(&pool, &event).await;
+        }
+        assert_canceled_once(&pool, &same_second_id).await;
+
+        for case in 0..12 {
+            let subscription_id = format!("sub_{marker}_concurrent_{case}");
+            let events = [
+                subscription_event(
+                    &format!("evt_{marker}_concurrent_{case}_created"),
+                    "customer.subscription.created",
+                    1_700_001_100,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+                subscription_event(
+                    &format!("evt_{marker}_concurrent_{case}_updated"),
+                    "customer.subscription.updated",
+                    1_700_001_200,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+                subscription_event(
+                    &format!("evt_{marker}_concurrent_{case}_deleted"),
+                    "customer.subscription.deleted",
+                    1_700_001_300,
+                    &subscription_id,
+                    &customer_id,
+                    &price_id,
+                ),
+            ];
+            let barrier = Arc::new(Barrier::new(events.len()));
+            let mut joins = Vec::new();
+            for event in events {
+                let pool = pool.clone();
+                let barrier = barrier.clone();
+                joins.push(tokio::spawn(async move {
+                    barrier.wait().await;
+                    apply_test_event(&pool, &event).await;
+                }));
+            }
+            for join in joins {
+                join.await.expect("concurrent event task");
+            }
+            assert_canceled_once(&pool, &subscription_id).await;
+        }
+
+        sqlx::query(
+            "DELETE FROM provider_object_event_watermarks WHERE provider_object_id LIKE $1",
+        )
+        .bind(format!("sub_{marker}%"))
+        .execute(&pool)
+        .await
+        .expect("delete test watermarks");
+        sqlx::query("DELETE FROM users WHERE id = $1")
+            .bind(user_id)
+            .execute(&pool)
+            .await
+            .expect("delete test user");
+        sqlx::query("DELETE FROM plans WHERE id = $1")
+            .bind(plan_id)
+            .execute(&pool)
+            .await
+            .expect("delete test plan");
+    }
 }

src/services/mail.rs

@@ -183,6 +183,13 @@ impl Mailer {
             .await
     }
 
+    pub async fn send_email_change_notice(&self, to: &str) -> Result<(), AppError> {
+        let text = "您的 ImageForge 账号邮箱已完成修改。若非本人操作，请立即联系管理员并重置密码。";
+        let html = "<h2>账号邮箱已修改</h2><p>您的 ImageForge 账号邮箱已完成修改。</p><p>若非本人操作，请立即联系管理员并重置密码。</p>";
+        self.send_email(to, "ImageForge 账号邮箱变更通知", text, html)
+            .await
+    }
+
     async fn send_email(
         &self,
         to: &str,
@@ -336,6 +343,11 @@ pub async fn send_password_reset_email(
         .await
 }
 
+pub async fn send_email_change_notice(state: &AppState, to: &str) -> Result<(), AppError> {
+    let mailer = resolve_mailer(state).await?;
+    mailer.send_email_change_notice(to).await
+}
+
 pub async fn send_test_email(state: &AppState, to: &str) -> Result<(), AppError> {
     let mailer = resolve_mailer(state).await?;
     let year = chrono::Utc::now().year().to_string();

src/services/storage.rs

@@ -291,6 +291,26 @@ pub fn result_key(retention_hours: i64, task_id: Uuid, file_id: Uuid, extension:
     )
 }
 
+pub fn result_attempt_key(
+    retention_hours: i64,
+    task_id: Uuid,
+    file_id: Uuid,
+    task_attempt: i64,
+    file_attempt: i64,
+    extension: &str,
+) -> String {
+    let now = Utc::now();
+    format!(
+        "results/{}/{:04}/{:02}/{task_id}/{file_id}-t{}-f{}.{}",
+        retention_prefix(retention_hours),
+        now.year(),
+        now.month(),
+        task_attempt.max(1),
+        file_attempt.max(1),
+        extension.trim_start_matches('.')
+    )
+}
+
 pub fn archive_key(retention_hours: i64, task_id: Uuid) -> String {
     let now = Utc::now();
     format!(
@@ -964,6 +984,8 @@ mod tests {
         assert!(key.starts_with("results/7d/"));
         assert!(key.ends_with("/00000000-0000-0000-0000-000000000001.webp"));
         assert!(archive_key(360, task_id).starts_with("archives/15d/"));
+        let attempt_key = result_attempt_key(24, task_id, file_id, 2, 3, "avif");
+        assert!(attempt_key.contains("-t2-f3.avif"));
     }
 
     #[test]