fix: secure email change recovery flow

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 email = email.trim().to_lowercase();
-        credentials::validate_email(&email)?;
-        if email != user.email {
-            next_email = email;
-            email_changed = true;
+    let next_email = match req.email.as_ref() {
+        Some(email) => {
+            let email = email.trim().to_lowercase();
+            credentials::validate_email(&email)?;
+            email
         }
-    }
-
-    if let Some(username) = req.username.as_ref() {
-        let username = username.trim().to_string();
-        credentials::validate_username(&username)?;
-        if username != user.username {
-            next_username = username;
+        None => user.email.clone(),
+    };
+    let next_username = match req.username.as_ref() {
+        Some(username) => {
+            let username = username.trim().to_string();
+            credentials::validate_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)
-    .bind(&next_email)
-    .bind(&next_username)
-    .bind(email_verified_at)
-    .fetch_one(&mut *tx)
-    .await
-    .map_err(map_unique_violation)?;
+        let email_in_use: bool =
+            sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM users WHERE email = $1 AND id <> $2)")
+                .bind(&next_email)
+                .bind(user_id)
+                .fetch_one(&mut *tx)
+                .await
+                .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");
+    }
+}