eric / rustwarden

//! Two-step email change service (RFC 0041).
//!
//! Nothing is ever written to `users.email` before possession of the candidate
//! address is proven. A request lands in `pending_email_changes` (hashed
//! tokens + expiration), the account keeps its current — verified — address,
//! and the previous mailbox is informed with a report link.
//!
//! The only path that writes an address without that proof is
//! [`EmailChangeService::admin_force_change`](crate::services::email_change_service::EmailChangeService::admin_force_change),
//! restricted to the `admin` role
//! and traced in `audit_logs`.

use crate::config::email::EmailConfig;
use crate::error::AppError;
use crate::hooks::{AuditEvent, RustwardenHooks};
use crate::i18n::service::I18nService;
use crate::models::{audit_log, pending_email_change, user};
use crate::services::auth_service::AuthService;
use crate::services::email_queue_service::EmailQueueService;
use crate::services::email_sanitization::{mask_email, sanitize_email_content};
use crate::services::password_change_service::PasswordChangeService;
use crate::services::rate_limit_service::RateLimitService;
use crate::services::refresh_token_service::RefreshTokenService;
use crate::services::user_preferences_service::UserPreferencesService;
use crate::services::user_service::UserService;
use crate::templates::email_templates::{
    EmailChangeByAdminTemplate, EmailChangeNoticeTemplate, EmailChangeReportedAdminTemplate,
    EmailChangeValidationTemplate,
};
use chrono::{Datelike, Duration, Utc};
use sea_orm::{
    ActiveModelTrait, ColumnTrait, Condition, DatabaseConnection, EntityTrait, NotSet, QueryFilter,
    Set,
};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::OnceLock;
use std::time::Duration as StdDuration;
use uuid::Uuid;

/// Default: 3 email-change requests per user per hour.
const EMAIL_CHANGE_RATE_LIMIT_MAX: u32 = 3;
/// Sliding window backing [`EMAIL_CHANGE_RATE_LIMIT_MAX`].
const EMAIL_CHANGE_RATE_LIMIT_WINDOW: StdDuration = StdDuration::from_secs(3600);

/// Retention of `pending_email_changes` rows, in days after expiration.
///
/// Compile-time constant on purpose: RFC 0041 adds **no** environment
/// variable. Past this delay a row carries no operational value — the durable
/// trail lives in `audit_logs` with masked addresses — while `new_email` is
/// personal data kept in clear text.
pub const PENDING_EMAIL_CHANGE_RETENTION_DAYS: i64 = 30;

/// Error surfaced when the candidate address was claimed by another account
/// between the request and its confirmation.
const EMAIL_NO_LONGER_AVAILABLE: &str =
    "This email address is no longer available. Request the change again with a different address.";

/// Process-wide sliding-window limiter for email-change requests.
///
/// Distinct from the password-reset and email-validation limiters so the three
/// flows keep separate budgets.
static EMAIL_CHANGE_LIMITER: OnceLock<RateLimitService> = OnceLock::new();

/// Return the process-wide email-change limiter, initializing it once.
fn shared_email_change_limiter() -> RateLimitService {
    EMAIL_CHANGE_LIMITER
        .get_or_init(|| {
            let limiter = RateLimitService::new();
            if let Ok(handle) = tokio::runtime::Handle::try_current() {
                handle.spawn(
                    limiter
                        .clone()
                        .cleanup_loop(StdDuration::from_secs(300), StdDuration::from_secs(3600)),
                );
            } else {
                tracing::warn!("email-change limiter: no tokio runtime, cleanup_loop not started");
            }
            limiter
        })
        .clone()
}

/// Read-only state of a link token, as reported to an interstitial page.
///
/// Returned by [`EmailChangeService::inspect_confirmation_token`] and
/// [`EmailChangeService::inspect_report_token`] so a `GET` can display a
/// useful message **without any side effect**: a consumer replacing the
/// built-in pages with its own reuses the same read-only check.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EmailChangeLinkState {
    /// The token matches a pending request: the form may be displayed.
    Usable,
    /// No request carries this token.
    Unknown,
    /// The request exists but its expiration has passed.
    Expired,
    /// The request exists but has already been applied.
    Used,
}

/// Two-step email change service.
///
/// Nothing is written to `users.email` until the confirmation link mailed to
/// the candidate address is followed. A typo therefore expires harmlessly and
/// the account never loses access.
///
/// ```rust,ignore
/// use rustwarden_core::config::email::EmailConfig;
/// use rustwarden_core::services::email_change_service::EmailChangeService;
///
/// let email_config = EmailConfig::from_env().map_err(AppError::Config)?;
/// EmailChangeService::new(email_config)
///     .request_change(&db, user_id, "new@example.com", current_password)
///     .await?;
/// // `users.email` is unchanged; two emails are queued.
/// ```
pub struct EmailChangeService {
    /// SMTP / queue configuration used to build the outgoing messages.
    email_config: EmailConfig,
    /// Sliding-window rate limiter (shared abstraction, see [`RateLimitService`]).
    rate_limiter: RateLimitService,
    /// Optional translation backend. When `None`, emails fall back to the
    /// strings embedded in this module, so the service stays usable headless
    /// without any i18n plumbing.
    i18n: Option<Arc<I18nService>>,
}

impl EmailChangeService {
    /// Create a service using the process-wide rate limiter.
    ///
    /// ```rust,ignore
    /// use rustwarden_core::services::email_change_service::EmailChangeService;
    ///
    /// let service = EmailChangeService::new(EmailConfig::from_env().map_err(AppError::Config)?);
    /// ```
    pub fn new(email_config: EmailConfig) -> Self {
        Self {
            email_config,
            rate_limiter: shared_email_change_limiter(),
            i18n: None,
        }
    }

    /// Attach the translation service so emails honour the recipient locale.
    ///
    /// The recipient locale is read from `user_preferences.language` and falls
    /// back to [`config::tasks::default_locale`](crate::config::tasks::default_locale).
    /// Missing keys fall back to the strings embedded in this module.
    ///
    /// ```rust,ignore
    /// use std::sync::Arc;
    /// use rustwarden_core::services::email_change_service::EmailChangeService;
    ///
    /// let service = EmailChangeService::new(email_config).with_i18n(Arc::clone(&i18n_service));
    /// ```
    pub fn with_i18n(mut self, i18n: Arc<I18nService>) -> Self {
        self.i18n = Some(i18n);
        self
    }

    /// Build a service around an injected limiter (unit tests; avoids the
    /// process [`OnceLock`]).
    #[cfg(test)]
    pub(crate) fn new_with_limiter(
        email_config: EmailConfig,
        rate_limiter: RateLimitService,
    ) -> Self {
        Self {
            email_config,
            rate_limiter,
            i18n: None,
        }
    }

    /// Open an email-change request for an authenticated user.
    ///
    /// Re-authentication is mandatory: a stolen session alone cannot move the
    /// account's identity. `current_password` is checked through
    /// [`AuthService::verify_credentials`](crate::services::auth_service::AuthService::verify_credentials),
    /// which runs bcrypt in constant time.
    ///
    /// On success two emails are queued:
    /// - a confirmation link to `new_email`;
    /// - a notice plus a report link to the account's **current** address.
    ///
    /// `users.email`, `email_verified` and `email_verified_at` are left
    /// untouched. Unlike the v1 flow, queueing is **not** best-effort: without
    /// the emails the request is unusable, so the error is propagated. This is
    /// safe precisely because nothing was written to `users`.
    ///
    /// ```rust,ignore
    /// use rustwarden_core::services::email_change_service::EmailChangeService;
    ///
    /// EmailChangeService::new(email_config)
    ///     .request_change(&db, user_id, "new@example.com", "current-password")
    ///     .await?;
    /// ```
    ///
    /// # Errors
    ///
    /// - [`AppError::TooManyRequests`] — more than 3 requests for this user
    ///   within the hour.
    /// - [`AppError::InvalidCredentials`] — wrong `current_password`.
    /// - [`AppError::Validation`] — malformed address, address identical to the
    ///   current one, or address already taken by another account.
    /// - [`AppError::NotFound`] — unknown `user_id`.
    pub async fn request_change(
        &self,
        db: &DatabaseConnection,
        user_id: Uuid,
        new_email: &str,
        current_password: &str,
    ) -> Result<(), AppError> {
        // 1. Per-user budget: the abuse to bound is "one account spamming
        //    third-party mailboxes", hence the user id as key.
        self.rate_limiter.check_sliding_window(
            &format!("email_change:{user_id}"),
            EMAIL_CHANGE_RATE_LIMIT_MAX,
            EMAIL_CHANGE_RATE_LIMIT_WINDOW,
        )?;

        // 2. Target account.
        let user = UserService::get_user_by_id(db, user_id)
            .await?
            .ok_or_else(|| AppError::NotFound("User not found".to_string()))?;

        // 3. Re-authentication. The current address always contains '@', so the
        //    email branch of `verify_credentials` is taken deterministically,
        //    even for legacy usernames containing '@'. `verify_credentials` —
        //    not `authenticate_user` — is used on purpose: the RFC 0040
        //    `validation_email_required` policy does not apply here.
        AuthService::verify_credentials(db, &user.email, current_password).await?;

        // 4. Normalisation and checks.
        let trimmed = new_email.trim();
        if !trimmed.contains('@') || !trimmed.contains('.') {
            return Err(AppError::Validation("Invalid email format".to_string()));
        }
        if trimmed == user.email.as_str() {
            return Err(AppError::Validation(
                "New email address is identical to the current one".to_string(),
            ));
        }
        if let Some(existing) = UserService::get_user_by_email(db, trimmed).await? {
            if existing.id != user_id {
                return Err(AppError::Validation(format!(
                    "Email '{trimmed}' already exists"
                )));
            }
        }

        // 5. One active request at a time, plus an opportunistic retention
        //    purge so a consumer that never starts the cleanup processor still
        //    honours the retention window for active accounts.
        self.invalidate_previous_requests(db, user_id).await?;
        Self::purge_stale_requests_for_user(db, user_id).await?;

        // 6. Two independent secrets: confirming and reporting are two
        //    different capabilities held by two different mailboxes.
        let token = Uuid::new_v4().to_string();
        let report_token = Uuid::new_v4().to_string();
        let now = Utc::now();
        let expires_at =
            now + Duration::hours(crate::config::tasks::email_validation_expiration_hours());

        // 7. Persist the request.
        pending_email_change::ActiveModel {
            id: Set(Uuid::new_v4()),
            user_id: Set(user_id),
            new_email: Set(trimmed.to_string()),
            token_hash: Set(Self::hash_token(&token)),
            report_token_hash: Set(Self::hash_token(&report_token)),
            expires_at: Set(expires_at),
            used_at: NotSet,
            reported_at: NotSet,
            created_at: Set(now),
        }
        .insert(db)
        .await?;

        // 8. Both notifications. Errors propagate: a request nobody can
        //    confirm is worthless, and the orphan row expires harmlessly.
        let locale = Self::locale_for(db, user_id).await;
        let confirm_url = format!(
            "{}/api/v1/auth/email-change/confirm/{}",
            self.email_config.email_base_url, token
        );
        let report_url = format!(
            "{}/api/v1/auth/email-change/report/{}",
            self.email_config.email_base_url, report_token
        );
        let hours = crate::config::tasks::email_validation_expiration_hours();

        // Confirmation to the CANDIDATE address.
        let (subject, body_html, body_text) = self
            .build_validation_email(&locale, &user.username, &confirm_url, hours)
            .await;
        EmailQueueService::enqueue(
            db,
            trimmed.to_string(),
            Some(user.username.clone()),
            self.email_config.smtp_from.clone(),
            self.email_config.smtp_from_name.clone(),
            subject,
            body_html,
            body_text,
            "email_change_validation".to_string(),
        )
        .await?;

        // Notice + report link to the CURRENT address. It never carries the
        // confirmation link, and the candidate address is masked.
        let (subject, body_html, body_text) = self
            .build_notice_email(&locale, &user.username, &report_url, trimmed)
            .await;
        EmailQueueService::enqueue(
            db,
            user.email.clone(),
            Some(user.username.clone()),
            self.email_config.smtp_from.clone(),
            self.email_config.smtp_from_name.clone(),
            subject,
            body_html,
            body_text,
            "email_change_notice".to_string(),
        )
        .await?;

        tracing::info!(
            user_id = %user_id,
            candidate = %mask_email(trimmed),
            "Email-change request opened; users.email left untouched"
        );

        Ok(())
    }

    /// Apply a pending change from its confirmation token.
    ///
    /// Confirming from the page the emailed link opens **is** the proof of
    /// possession, so the address lands already verified: `users.email =
    /// new_email`, `email_verified = true`, `email_verified_at = now()`.
    /// Uniqueness is re-checked here because another account may have claimed
    /// the address in the meantime.
    ///
    /// All refresh tokens are revoked (same contract as
    /// [`UserService::change_password_direct`](crate::services::user_service::UserService::change_password_direct)),
    /// and any pending password-change request is dropped through
    /// [`PasswordChangeService::invalidate_on_email_change`](crate::services::password_change_service::PasswordChangeService::invalidate_on_email_change):
    /// a confirmation link mailed to the previous address must never be usable
    /// once that address no longer belongs to the account.
    ///
    /// Hooks fired here, best-effort:
    /// [`RustwardenHooks::on_email_verified`] (the click *is* the proof of
    /// possession) and [`RustwardenHooks::on_audit_event`] with kind
    /// `user.email_changed`.
    ///
    /// Returns the user id whose address changed.
    ///
    /// ```rust,ignore
    /// use rustwarden_core::services::email_change_service::EmailChangeService;
    ///
    /// let user_id = EmailChangeService::new(email_config)
    ///     .confirm_change(&db, token, hooks.as_ref())
    ///     .await?;
    /// ```
    ///
    /// # Errors
    ///
    /// - [`AppError::Validation`] — unknown, expired or already-used token, or
    ///   the candidate address is no longer available.
    /// - [`AppError::NotFound`] — the user was deleted since the request.
    pub async fn confirm_change(
        &self,
        db: &DatabaseConnection,
        token: &str,
        hooks: Option<&Arc<dyn RustwardenHooks>>,
    ) -> Result<Uuid, AppError> {
        // 1. Lookup by hash — the plain token is never stored.
        let token_hash = Self::hash_token(token);
        let row = pending_email_change::Entity::find()
            .filter(pending_email_change::Column::TokenHash.eq(&token_hash))
            .one(db)
            .await?
            .ok_or_else(|| AppError::Validation("Invalid email-change token".to_string()))?;

        // 2 & 3. Single use, then expiration.
        if row.is_used() {
            return Err(AppError::Validation(
                "This email-change link has already been used".to_string(),
            ));
        }
        if row.is_expired() {
            return Err(AppError::Validation(
                "This email-change link has expired".to_string(),
            ));
        }

        // 4. `reported_at` is deliberately NOT a gate: a report alerts, it
        //    never refuses (see the RFC's dedicated section).

        // 5. Uniqueness may have been lost since the request was opened. The
        //    row is NOT consumed, so the link keeps working if the address is
        //    released before expiration.
        if let Some(existing) = UserService::get_user_by_email(db, &row.new_email).await? {
            if existing.id != row.user_id {
                return Err(AppError::Validation(EMAIL_NO_LONGER_AVAILABLE.to_string()));
            }
        }

        let user = UserService::get_user_by_id(db, row.user_id)
            .await?
            .ok_or_else(|| AppError::NotFound("User not found".to_string()))?;
        let previous_email = user.email.clone();
        let username = user.username.clone();
        let now = Utc::now();

        // 6. Write the proven address, already verified.
        let mut user_active: user::ActiveModel = user.into();
        user_active.email = Set(row.new_email.clone());
        user_active.email_verified = Set(true);
        user_active.email_verified_at = Set(Some(now));
        user_active.updated_at = Set(now);
        user_active
            .update(db)
            .await
            .map_err(Self::map_email_write_error)?;

        // 7. Consume the request.
        let mut row_active: pending_email_change::ActiveModel = row.clone().into();
        row_active.used_at = Set(Some(now));
        row_active.update(db).await?;

        // 8. The login identity just moved: every session re-authenticates.
        RefreshTokenService::revoke_all_tokens_for_user(db, row.user_id).await?;

        // 9. A password-change link mailed to the previous address must not
        //    survive the address it was sent to. Logged, never fatal: the
        //    address change is already written.
        if let Err(e) = PasswordChangeService::new()
            .invalidate_on_email_change(db, row.user_id)
            .await
        {
            tracing::warn!(
                user_id = %row.user_id,
                "Failed to invalidate pending password-change requests after email change: {}",
                e
            );
        }

        // 10. Durable audit trail, masked addresses only.
        let message = format!(
            "user={} old={} new={}",
            username,
            mask_email(&previous_email),
            mask_email(&row.new_email)
        );
        Self::write_audit_log(
            db,
            "email_changed",
            "INFO",
            Some(row.user_id),
            None,
            message,
        )
        .await;

        // 11. Hooks, dispatched from the service so headless consumers mounting
        //     their own confirmation route get the same events.
        if let Some(hooks) = hooks {
            if let Err(e) = hooks.on_email_verified(db, row.user_id).await {
                tracing::warn!("RustwardenHooks::on_email_verified failed: {}", e);
            }
            let metadata = serde_json::json!({
                "previous_email": mask_email(&previous_email),
                "new_email": mask_email(&row.new_email),
            });
            let event = AuditEvent {
                // Actor and target are the same account here — this is a
                // self-service change — but the target is still named, so a
                // consumer routing events on `target` sees the self-service
                // path alongside `user.email_changed_by_admin`.
                kind: "user.email_changed",
                actor_id: Some(row.user_id),
                target: Some(("user", row.user_id)),
                metadata: &metadata,
            };
            if let Err(e) = hooks.on_audit_event(db, event).await {
                tracing::warn!("RustwardenHooks::on_audit_event failed: {}", e);
            }
        }

        Ok(row.user_id)
    }

    /// Record that the current address owner flagged a request as suspicious.
    ///
    /// Writes an `audit_logs` row, emits
    /// [`RustwardenHooks::on_audit_event`] with kind `user.email_change_reported`
    /// and queues an alert to every administrator.
    ///
    /// **This never blocks the change.** A report is a signal for operators,
    /// not an authorisation gate: turning it into one would hand any reader of
    /// the old mailbox a denial-of-service lever over the account owner.
    ///
    /// Idempotent: a request already reported returns `Ok` without re-notifying.
    ///
    /// ```rust,ignore
    /// use rustwarden_core::services::email_change_service::EmailChangeService;
    ///
    /// let user_id = EmailChangeService::new(email_config)
    ///     .report_suspicious(&db, report_token, hooks.as_ref())
    ///     .await?;
    /// ```
    ///
    /// # Errors
    ///
    /// - [`AppError::Validation`] — unknown report token.
    pub async fn report_suspicious(
        &self,
        db: &DatabaseConnection,
        report_token: &str,
        hooks: Option<&Arc<dyn RustwardenHooks>>,
    ) -> Result<Uuid, AppError> {
        // 1. Lookup by the *report* hash: a different secret than the one
        //    mailed to the candidate address.
        let report_token_hash = Self::hash_token(report_token);
        let row = pending_email_change::Entity::find()
            .filter(pending_email_change::Column::ReportTokenHash.eq(&report_token_hash))
            .one(db)
            .await?
            .ok_or_else(|| AppError::Validation("Invalid report token".to_string()))?;

        // 2. Idempotence — double click, page refresh or browser form resend.
        if row.is_reported() {
            return Ok(row.user_id);
        }

        // 3. Flag it, even when expired or already consumed: the signal keeps
        //    its value for the operator.
        let now = Utc::now();
        let user_id = row.user_id;
        let candidate = row.new_email.clone();
        let mut row_active: pending_email_change::ActiveModel = row.into();
        row_active.reported_at = Set(Some(now));
        row_active.update(db).await?;

        // 4. Audit trail.
        let message = format!("user_id={} candidate={}", user_id, mask_email(&candidate));
        Self::write_audit_log(
            db,
            "email_change_reported",
            "WARN",
            None,
            Some(user_id),
            message,
        )
        .await;

        // 5. Hook fan-out.
        if let Some(hooks) = hooks {
            let metadata = serde_json::json!({
                "user_id": user_id.to_string(),
                "candidate_email": mask_email(&candidate),
            });
            let event = AuditEvent {
                kind: "user.email_change_reported",
                actor_id: None,
                target: Some(("user", user_id)),
                metadata: &metadata,
            };
            if let Err(e) = hooks.on_audit_event(db, event).await {
                tracing::warn!("RustwardenHooks::on_audit_event failed: {}", e);
            }
        }

        // 6. Alert every administrator. Best-effort: the report is already
        //    recorded, a mail failure must not turn it into an error.
        self.notify_admins_of_report(db, user_id, &candidate).await;

        Ok(user_id)
    }

    /// Read-only state of a **confirmation** token.
    ///
    /// Performs a single indexed `SELECT` and writes nothing: this is what
    /// makes `GET /api/v1/auth/email-change/confirm/{token}` immune to the
    /// link prefetching done by anti-phishing gateways and mail clients.
    ///
    /// ```rust,ignore
    /// use rustwarden_core::services::email_change_service::{
    ///     EmailChangeLinkState, EmailChangeService,
    /// };
    ///
    /// match EmailChangeService::inspect_confirmation_token(&db, &token).await? {
    ///     EmailChangeLinkState::Usable => { /* render the confirm form */ }
    ///     _ => { /* render an explanation, no form */ }
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`AppError::Database`] if the lookup fails.
    pub async fn inspect_confirmation_token(
        db: &DatabaseConnection,
        token: &str,
    ) -> Result<EmailChangeLinkState, AppError> {
        let token_hash = Self::hash_token(token);
        let row = pending_email_change::Entity::find()
            .filter(pending_email_change::Column::TokenHash.eq(&token_hash))
            .one(db)
            .await?;
        Ok(Self::state_of(row.as_ref()))
    }

    /// Read-only state of a **report** token. Same no-side-effect contract as
    /// [`EmailChangeService::inspect_confirmation_token`].
    ///
    /// ```rust,ignore
    /// let state = EmailChangeService::inspect_report_token(&db, &report_token).await?;
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`AppError::Database`] if the lookup fails.
    pub async fn inspect_report_token(
        db: &DatabaseConnection,
        report_token: &str,
    ) -> Result<EmailChangeLinkState, AppError> {
        let token_hash = Self::hash_token(report_token);
        let row = pending_email_change::Entity::find()
            .filter(pending_email_change::Column::ReportTokenHash.eq(&token_hash))
            .one(db)
            .await?;
        Ok(Self::state_of(row.as_ref()))
    }

    /// Map a looked-up row to its link state. `Used` wins over `Expired`:
    /// "already done" is the more actionable message of the two.
    fn state_of(row: Option<&pending_email_change::Model>) -> EmailChangeLinkState {
        match row {
            None => EmailChangeLinkState::Unknown,
            Some(row) if row.is_used() => EmailChangeLinkState::Used,
            Some(row) if row.is_expired() => EmailChangeLinkState::Expired,
            Some(_) => EmailChangeLinkState::Usable,
        }
    }

    /// Administrative override: write `new_email` without confirmation.
    ///
    /// Operational recourse for the one dead end the two-step flow cannot
    /// solve on its own: a user who lost **both** the password and the access
    /// to the account's current mailbox. The v1 capability — an admin fixing
    /// an address through `PUT /admin/users/{id}` — is preserved, but it
    /// becomes an **explicit, audited operation** instead of a side effect of
    /// [`UserService::update_profile`](crate::services::user_service::UserService::update_profile).
    ///
    /// Contract:
    /// - `actor_id` **must** carry the `admin` role
    ///   ([`UserService::get_user_roles`](crate::services::user_service::UserService::get_user_roles));
    ///   otherwise [`AppError::Forbidden`] and nothing is written.
    /// - `users.email = new_email`, `email_verified = true`,
    ///   `email_verified_at = now()` — the operator vouches for the address
    ///   after an out-of-band identity check.
    /// - Every pending request of the target user is consumed, so a stale
    ///   confirmation link cannot resurrect the previous candidate address.
    /// - All refresh tokens of the **target** account are revoked, exactly as
    ///   in [`Self::confirm_change`]; the acting administrator's own sessions
    ///   are untouched.
    /// - An `audit_logs` row is written (`action = "email_changed_by_admin"`,
    ///   `level = "WARN"`, `user_id = actor_id`, `target_user_id = user_id`,
    ///   masked addresses) and [`RustwardenHooks::on_audit_event`] is emitted
    ///   with kind `user.email_changed_by_admin`.
    /// - [`RustwardenHooks::on_email_verified`] is **not** emitted: there was
    ///   no proof of possession by the address owner.
    /// - Both mailboxes are notified. Neither mail carries an action link —
    ///   there is nothing left to confirm.
    ///
    /// No rate limit and no password check: authorisation is the admin role,
    /// enforced by the caller's route middleware **and** re-checked here.
    /// Echoing the target's current address is a silent no-op, because admin
    /// forms resubmit every field.
    ///
    /// ```rust,ignore
    /// use rustwarden_core::services::email_change_service::EmailChangeService;
    ///
    /// // `admin_id` must hold the `admin` role, or the call fails with Forbidden.
    /// EmailChangeService::new(email_config)
    ///     .admin_force_change(&db, admin_id, user_id, "fixed@example.com", None)
    ///     .await?;
    /// ```
    ///
    /// # Errors
    ///
    /// - [`AppError::Forbidden`] — `actor_id` is not an administrator.
    /// - [`AppError::Validation`] — malformed address, or address already
    ///   taken by another account.
    /// - [`AppError::NotFound`] — unknown `actor_id` or unknown `user_id`.
    pub async fn admin_force_change(
        &self,
        db: &DatabaseConnection,
        actor_id: Uuid,
        user_id: Uuid,
        new_email: &str,
        hooks: Option<&Arc<dyn RustwardenHooks>>,
    ) -> Result<(), AppError> {
        // 1. Authorisation is re-checked in the service: a headless call
        //    straight to the service must not bypass the route middleware.
        let actor = UserService::get_user_by_id(db, actor_id)
            .await?
            .ok_or_else(|| AppError::NotFound("Actor not found".to_string()))?;
        let roles = UserService::get_user_roles(db, actor_id).await?;
        if !roles.iter().any(|role| role == "admin") {
            return Err(AppError::Forbidden);
        }

        // 2. Self-targeting is allowed: an admin changing their own address is
        //    still an admin acting as an admin, and the trail shows it.
        let target = UserService::get_user_by_id(db, user_id)
            .await?
            .ok_or_else(|| AppError::NotFound("User not found".to_string()))?;

        // 3. Same checks as `request_change`, minus the "identical address"
        //    error which becomes a silent no-op here.
        let trimmed = new_email.trim();
        if !trimmed.contains('@') || !trimmed.contains('.') {
            return Err(AppError::Validation("Invalid email format".to_string()));
        }
        if trimmed == target.email.as_str() {
            return Ok(());
        }
        if let Some(existing) = UserService::get_user_by_email(db, trimmed).await? {
            if existing.id != user_id {
                return Err(AppError::Validation(format!(
                    "Email '{trimmed}' already exists"
                )));
            }
        }

        // 4. A still-valid confirmation link must not overwrite the address
        //    the operator just set.
        self.invalidate_previous_requests(db, user_id).await?;

        let previous_email = target.email.clone();
        let username = target.username.clone();
        let now = Utc::now();

        // 5. The verified flag is set because the operator vouches out of band,
        //    exactly like `admin_reset_password` does for the password.
        let mut target_active: user::ActiveModel = target.into();
        target_active.email = Set(trimmed.to_string());
        target_active.email_verified = Set(true);
        target_active.email_verified_at = Set(Some(now));
        target_active.updated_at = Set(now);
        target_active
            .update(db)
            .await
            .map_err(Self::map_email_write_error)?;

        // 6. Target sessions only — the acting admin keeps theirs.
        RefreshTokenService::revoke_all_tokens_for_user(db, user_id).await?;

        // 7. Same reason as in `confirm_change`.
        if let Err(e) = PasswordChangeService::new()
            .invalidate_on_email_change(db, user_id)
            .await
        {
            tracing::warn!(
                user_id = %user_id,
                "Failed to invalidate pending password-change requests after admin email change: {}",
                e
            );
        }

        // 8. WARN, not INFO: an identity write without proof of possession
        //    deserves to surface.
        let message = format!(
            "actor={} user={} old={} new={}",
            actor.username,
            username,
            mask_email(&previous_email),
            mask_email(trimmed)
        );
        Self::write_audit_log(
            db,
            "email_changed_by_admin",
            "WARN",
            Some(actor_id),
            Some(user_id),
            message,
        )
        .await;

        // 9. `on_email_verified` is deliberately NOT emitted here.
        if let Some(hooks) = hooks {
            let metadata = serde_json::json!({
                "previous_email": mask_email(&previous_email),
                "new_email": mask_email(trimmed),
            });
            let event = AuditEvent {
                kind: "user.email_changed_by_admin",
                actor_id: Some(actor_id),
                target: Some(("user", user_id)),
                metadata: &metadata,
            };
            if let Err(e) = hooks.on_audit_event(db, event).await {
                tracing::warn!("RustwardenHooks::on_audit_event failed: {}", e);
            }
        }

        // 10. Both mailboxes, no action link. Best-effort: the write is done
        //     and the operator already has a synchronous return value.
        self.notify_admin_forced_change(db, user_id, &username, &previous_email, trimmed)
            .await;

        Ok(())
    }

    /// Hard-delete rows expired for longer than
    /// [`PENDING_EMAIL_CHANGE_RETENTION_DAYS`] (30 days, a compile-time
    /// constant — no environment variable).
    ///
    /// The predicate is `expires_at < now() - retention`, whatever the row
    /// state (consumed, reported, or simply stale): past that delay the row has
    /// no operational value left, the durable trail being in `audit_logs`.
    ///
    /// Returns the number of deleted rows. Called by
    /// [`PendingEmailChangeCleanupProcessor`](crate::tasks::pending_email_change_cleanup::PendingEmailChangeCleanupProcessor),
    /// and public so a consumer can drive it from its own cron instead. **The
    /// framework starts nothing by itself**: the consumer must call `.start()`
    /// on the processor.
    ///
    /// ```rust,ignore
    /// use rustwarden_core::services::email_change_service::EmailChangeService;
    ///
    /// let deleted = EmailChangeService::cleanup_expired(&db).await?;
    /// println!("{deleted} stale email-change request(s) purged");
    /// ```
    ///
    /// # Errors
    ///
    /// - [`AppError::Database`] — the delete statement failed.
    pub async fn cleanup_expired(db: &DatabaseConnection) -> Result<u64, AppError> {
        let cutoff = Utc::now() - Duration::days(PENDING_EMAIL_CHANGE_RETENTION_DAYS);
        let result = pending_email_change::Entity::delete_many()
            .filter(pending_email_change::Column::ExpiresAt.lt(cutoff))
            .exec(db)
            .await?;
        Ok(result.rows_affected)
    }

    // ── Internals ───────────────────────────────────────────────────────────

    /// Hash a token using SHA-256 (hex, 64 chars).
    fn hash_token(token: &str) -> String {
        let mut hasher = Sha256::new();
        hasher.update(token.as_bytes());
        format!("{:x}", hasher.finalize())
    }

    /// Mark every pending request of this user as consumed.
    ///
    /// Functional replacement for the partial unique index `sea-query` cannot
    /// express portably — same approach as
    /// `EmailValidationService::invalidate_previous_tokens`.
    async fn invalidate_previous_requests(
        &self,
        db: &DatabaseConnection,
        user_id: Uuid,
    ) -> Result<(), AppError> {
        let rows = pending_email_change::Entity::find()
            .filter(pending_email_change::Column::UserId.eq(user_id))
            .filter(pending_email_change::Column::UsedAt.is_null())
            .all(db)
            .await?;

        let now = Utc::now();
        for row in rows {
            let mut active: pending_email_change::ActiveModel = row.into();
            active.used_at = Set(Some(now));
            active.update(db).await?;
        }

        Ok(())
    }

    /// Opportunistic retention purge, scoped to a single user.
    ///
    /// Belt and braces for consumers that never start
    /// [`PendingEmailChangeCleanupProcessor`](crate::tasks::pending_email_change_cleanup::PendingEmailChangeCleanupProcessor):
    /// rows of **this** user older than the retention window disappear on the
    /// next request. Rows of other users are untouched.
    async fn purge_stale_requests_for_user(
        db: &DatabaseConnection,
        user_id: Uuid,
    ) -> Result<u64, AppError> {
        let cutoff = Utc::now() - Duration::days(PENDING_EMAIL_CHANGE_RETENTION_DAYS);
        let result = pending_email_change::Entity::delete_many()
            .filter(
                Condition::all()
                    .add(pending_email_change::Column::UserId.eq(user_id))
                    .add(pending_email_change::Column::ExpiresAt.lt(cutoff)),
            )
            .exec(db)
            .await?;
        Ok(result.rows_affected)
    }

    /// Translate a concurrent `users.email` unique violation into the same
    /// user-facing `Validation` as the explicit uniqueness pre-check.
    ///
    /// The `UNIQUE` constraint on `users.email` closes the residual race
    /// between the `SELECT` and the `UPDATE`; surfacing it as a 500 would be
    /// both wrong and unhelpful.
    fn map_email_write_error(err: sea_orm::DbErr) -> AppError {
        let text = err.to_string().to_lowercase();
        if text.contains("duplicate key")
            || text.contains("unique constraint")
            || text.contains("unique violation")
        {
            AppError::Validation(EMAIL_NO_LONGER_AVAILABLE.to_string())
        } else {
            AppError::Database(err)
        }
    }

    /// Insert an `audit_logs` row. Best-effort: a logging failure must never
    /// undo an identity change that is already committed.
    async fn write_audit_log(
        db: &DatabaseConnection,
        action: &str,
        level: &str,
        user_id: Option<Uuid>,
        target_user_id: Option<Uuid>,
        message: String,
    ) {
        let log = audit_log::ActiveModel {
            id: Set(Uuid::new_v4()),
            user_id: Set(user_id),
            action: Set(action.to_string()),
            target_user_id: Set(target_user_id),
            level: Set(level.to_string()),
            message: Set(message),
            created_at: Set(Utc::now()),
            actor_remote_identity_id: Set(None),
            target_remote_identity_id: Set(None),
        };
        if let Err(e) = log.insert(db).await {
            tracing::warn!("Failed to write email-change audit log ({}): {}", action, e);
        }
    }

    /// Recipient locale: `user_preferences.language`, falling back to
    /// [`config::tasks::default_locale`](crate::config::tasks::default_locale).
    async fn locale_for(db: &DatabaseConnection, user_id: Uuid) -> String {
        match UserPreferencesService::get_preferences(db, user_id).await {
            Ok(Some(prefs)) => prefs.language,
            Ok(None) => crate::config::tasks::default_locale(),
            Err(e) => {
                tracing::warn!(
                    user_id = %user_id,
                    "Failed to read user preferences for email locale: {}",
                    e
                );
                crate::config::tasks::default_locale()
            }
        }
    }

    /// Translate `key` for `locale`, falling back to `fallback` when no
    /// [`I18nService`] is attached or the key is missing.
    async fn tr(&self, key: &str, locale: &str, fallback: &str) -> String {
        self.tr_vars(key, locale, fallback, &[]).await
    }

    /// Translate `key` for `locale` with `{placeholder}` substitution.
    ///
    /// Placeholders are applied to the translation **and** to the fallback, so
    /// a headless deployment without locale files renders the same sentences.
    async fn tr_vars(
        &self,
        key: &str,
        locale: &str,
        fallback: &str,
        vars: &[(&str, &str)],
    ) -> String {
        match &self.i18n {
            Some(i18n) => {
                let map: HashMap<String, String> = vars
                    .iter()
                    .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
                    .collect();
                let value = i18n.translate_with_vars(key, locale, &map).await;
                // `I18nService::translate` echoes the key when it is unknown.
                if value == key {
                    Self::interpolate(fallback, vars)
                } else {
                    value
                }
            }
            None => Self::interpolate(fallback, vars),
        }
    }

    /// Replace every `{name}` occurrence of `vars` in `template`.
    fn interpolate(template: &str, vars: &[(&str, &str)]) -> String {
        let mut out = template.to_string();
        for (name, value) in vars {
            out = out.replace(&format!("{{{name}}}"), value);
        }
        out
    }

    /// Last-resort HTML body when an Askama render fails.
    ///
    /// Keeps the flow usable instead of dropping the email: the text body is
    /// already complete, it is simply wrapped in a minimal document.
    fn html_fallback(text: &str) -> String {
        format!(
            "<!DOCTYPE html><html><body><pre>{}</pre></body></html>",
            sanitize_email_content(text)
        )
    }

    /// Render an email template, falling back to the plain-text body wrapped
    /// in minimal HTML if rendering fails.
    fn render_or_fallback(
        rendered: Result<String, askama::Error>,
        text: &str,
        which: &str,
    ) -> String {
        match rendered {
            Ok(html) => html,
            Err(e) => {
                tracing::warn!("Failed to render {} email template: {}", which, e);
                Self::html_fallback(text)
            }
        }
    }

    /// Build `(subject, html, text)` of the confirmation email sent to the
    /// candidate address.
    async fn build_validation_email(
        &self,
        locale: &str,
        username: &str,
        confirm_url: &str,
        hours: i64,
    ) -> (String, String, String) {
        let app_name = crate::config::app::default_app_name();
        let hours = hours.to_string();
        let subject = self
            .tr(
                "email_change.validation_subject",
                locale,
                "Confirm your new email address",
            )
            .await;
        let template = EmailChangeValidationTemplate {
            locale: locale.to_string(),
            app_name: app_name.clone(),
            title: self
                .tr(
                    "email_change.validation_title",
                    locale,
                    "Confirm your new email address",
                )
                .await,
            greeting: self
                .tr_vars(
                    "email_change.greeting",
                    locale,
                    "Hello {username},",
                    &[("username", username)],
                )
                .await,
            intro: self
                .tr_vars(
                    "email_change.validation_intro",
                    locale,
                    "An email-change request was opened for your {app_name} account. Open the \
                     confirmation page below and confirm the change there to complete it.",
                    &[("app_name", &app_name)],
                )
                .await,
            expires: self
                .tr_vars(
                    "email_change.validation_expires",
                    locale,
                    "This link is valid for {hours} hour(s).",
                    &[("hours", &hours)],
                )
                .await,
            ignore_notice: self
                .tr(
                    "email_change.validation_ignore",
                    locale,
                    "If you did not request this change, simply ignore this message: nothing has \
                     been modified and the link will expire on its own.",
                )
                .await,
            button_label: self
                .tr(
                    "email_change.validation_button",
                    locale,
                    "Open the confirmation page",
                )
                .await,
            confirm_url: confirm_url.to_string(),
            fallback_label: self
                .tr(
                    "email_change.fallback_label",
                    locale,
                    "Button not working? Copy this link into your browser:",
                )
                .await,
            footer: self
                .tr(
                    "email_change.footer",
                    locale,
                    "This email was sent automatically, please do not reply.",
                )
                .await,
            current_year: Utc::now().year(),
        };
        let text = template.render_text();
        let html =
            Self::render_or_fallback(template.render_html(), &text, "email-change validation");
        (subject, html, text)
    }

    /// Build `(subject, html, text)` of the notice sent to the current
    /// address. The candidate address is masked, and no confirmation link is
    /// ever included.
    async fn build_notice_email(
        &self,
        locale: &str,
        username: &str,
        report_url: &str,
        candidate_email: &str,
    ) -> (String, String, String) {
        let app_name = crate::config::app::default_app_name();
        let masked = mask_email(candidate_email);
        let subject = self
            .tr(
                "email_change.notice_subject",
                locale,
                "An email-change request was opened on your account",
            )
            .await;
        let template = EmailChangeNoticeTemplate {
            locale: locale.to_string(),
            app_name: app_name.clone(),
            title: self
                .tr("email_change.notice_title", locale, "Email-change request")
                .await,
            greeting: self
                .tr_vars(
                    "email_change.greeting",
                    locale,
                    "Hello {username},",
                    &[("username", username)],
                )
                .await,
            intro: self
                .tr_vars(
                    "email_change.notice_intro",
                    locale,
                    "A request to move your account to {email} has just been opened.",
                    &[("email", &masked)],
                )
                .await,
            unchanged: self
                .tr(
                    "email_change.notice_unchanged",
                    locale,
                    "Your account has NOT changed: it still uses this address, which stays \
                     verified until the new one is confirmed.",
                )
                .await,
            report_intro: self
                .tr(
                    "email_change.notice_report_intro",
                    locale,
                    "If you are not the origin of this request, open the report page below to \
                     alert the administrators.",
                )
                .await,
            button_label: self
                .tr("email_change.notice_button", locale, "Open the report page")
                .await,
            report_url: report_url.to_string(),
            fallback_label: self
                .tr(
                    "email_change.fallback_label",
                    locale,
                    "Button not working? Copy this link into your browser:",
                )
                .await,
            footer: self
                .tr(
                    "email_change.footer",
                    locale,
                    "This email was sent automatically, please do not reply.",
                )
                .await,
            current_year: Utc::now().year(),
        };
        let text = template.render_text();
        let html = Self::render_or_fallback(template.render_html(), &text, "email-change notice");
        (subject, html, text)
    }

    /// Build `(subject, html, text)` of the alert sent to administrators when
    /// a request is reported. Both addresses are masked.
    async fn build_reported_admin_email(
        &self,
        locale: &str,
        target_username: &str,
        current_email: &str,
        candidate_email: &str,
        reported_at: &str,
    ) -> (String, String, String) {
        let app_name = crate::config::app::default_app_name();
        let masked_old = mask_email(current_email);
        let masked_new = mask_email(candidate_email);
        let subject = self
            .tr(
                "email_change.reported_admin_subject",
                locale,
                "Email change reported by a user",
            )
            .await;
        let template = EmailChangeReportedAdminTemplate {
            locale: locale.to_string(),
            app_name: app_name.clone(),
            title: self
                .tr(
                    "email_change.reported_admin_title",
                    locale,
                    "Email change reported",
                )
                .await,
            intro: self
                .tr_vars(
                    "email_change.reported_admin_intro",
                    locale,
                    "User {username} reported an email-change request they state they did not \
                     initiate.",
                    &[("username", target_username)],
                )
                .await,
            details: self
                .tr_vars(
                    "email_change.reported_admin_details",
                    locale,
                    "Current address: {old_email} — requested address: {new_email} — report \
                     received on {reported_at}.",
                    &[
                        ("old_email", &masked_old),
                        ("new_email", &masked_new),
                        ("reported_at", reported_at),
                    ],
                )
                .await,
            not_blocked: self
                .tr(
                    "email_change.reported_admin_not_blocked",
                    locale,
                    "The change is NOT blocked: if the confirmation link is followed, the address \
                     will be updated. Step in if the situation warrants it.",
                )
                .await,
            footer: self
                .tr(
                    "email_change.footer",
                    locale,
                    "This email was sent automatically, please do not reply.",
                )
                .await,
            current_year: Utc::now().year(),
        };
        let text = template.render_text();
        let html =
            Self::render_or_fallback(template.render_html(), &text, "email-change report alert");
        (subject, html, text)
    }

    /// Build `(subject, html, text)` of the link-less notice sent to both
    /// addresses after an operator-driven change.
    async fn build_by_admin_email(
        &self,
        locale: &str,
        username: &str,
        previous_email: &str,
        new_email: &str,
    ) -> (String, String, String) {
        let app_name = crate::config::app::default_app_name();
        let masked_old = mask_email(previous_email);
        let masked_new = mask_email(new_email);
        let subject = self
            .tr(
                "email_change.by_admin_subject",
                locale,
                "Your email address was changed by an administrator",
            )
            .await;
        let template = EmailChangeByAdminTemplate {
            locale: locale.to_string(),
            app_name: app_name.clone(),
            title: self
                .tr(
                    "email_change.by_admin_title",
                    locale,
                    "Email address changed by an administrator",
                )
                .await,
            greeting: self
                .tr_vars(
                    "email_change.greeting",
                    locale,
                    "Hello {username},",
                    &[("username", username)],
                )
                .await,
            intro: self
                .tr_vars(
                    "email_change.by_admin_intro",
                    locale,
                    "The email address of your {app_name} account was changed by an administrator.",
                    &[("app_name", &app_name)],
                )
                .await,
            details: self
                .tr_vars(
                    "email_change.by_admin_details",
                    locale,
                    "Previous address: {old_email} — new address: {new_email}.",
                    &[("old_email", &masked_old), ("new_email", &masked_new)],
                )
                .await,
            support: self
                .tr(
                    "email_change.by_admin_support",
                    locale,
                    "No action is required on your side and there is nothing to confirm. If in \
                     doubt, contact support.",
                )
                .await,
            footer: self
                .tr(
                    "email_change.footer",
                    locale,
                    "This email was sent automatically, please do not reply.",
                )
                .await,
            current_year: Utc::now().year(),
        };
        let text = template.render_text();
        let html =
            Self::render_or_fallback(template.render_html(), &text, "admin-forced email change");
        (subject, html, text)
    }

    /// Queue one alert per administrator after a report. Best-effort.
    async fn notify_admins_of_report(
        &self,
        db: &DatabaseConnection,
        user_id: Uuid,
        candidate_email: &str,
    ) {
        let admins = match UserService::get_all_admin_users(db).await {
            Ok(admins) => admins,
            Err(e) => {
                tracing::warn!(
                    "Failed to list administrators for email-change report: {}",
                    e
                );
                return;
            }
        };

        let target = UserService::get_user_by_id(db, user_id)
            .await
            .ok()
            .flatten();
        let target_username = target
            .as_ref()
            .map(|u| u.username.clone())
            .unwrap_or_else(|| user_id.to_string());
        // Masked in the template; an operator mailbox does not need the
        // clear-text address, the durable trail lives in `audit_logs`.
        let target_email = target.map(|u| u.email).unwrap_or_default();
        let reported_at = Utc::now().format("%Y-%m-%d %H:%M:%S UTC").to_string();

        for admin in admins {
            let locale = Self::locale_for(db, admin.id).await;
            let (subject, body_html, body_text) = self
                .build_reported_admin_email(
                    &locale,
                    &target_username,
                    &target_email,
                    candidate_email,
                    &reported_at,
                )
                .await;
            if let Err(e) = EmailQueueService::enqueue(
                db,
                admin.email.clone(),
                Some(admin.username.clone()),
                self.email_config.smtp_from.clone(),
                self.email_config.smtp_from_name.clone(),
                subject,
                body_html,
                body_text,
                "email_change_reported".to_string(),
            )
            .await
            {
                tracing::warn!(
                    admin_id = %admin.id,
                    "Failed to queue email-change report alert: {}",
                    e
                );
            }
        }
    }

    /// Queue the two link-less notices after an operator-driven change.
    /// Best-effort: the write is already committed.
    async fn notify_admin_forced_change(
        &self,
        db: &DatabaseConnection,
        user_id: Uuid,
        username: &str,
        previous_email: &str,
        new_email: &str,
    ) {
        let locale = Self::locale_for(db, user_id).await;
        let (subject, body_html, body_text) = self
            .build_by_admin_email(&locale, username, previous_email, new_email)
            .await;

        for recipient in [previous_email, new_email] {
            if let Err(e) = EmailQueueService::enqueue(
                db,
                recipient.to_string(),
                Some(username.to_string()),
                self.email_config.smtp_from.clone(),
                self.email_config.smtp_from_name.clone(),
                subject.clone(),
                body_html.clone(),
                body_text.clone(),
                "email_change_by_admin".to_string(),
            )
            .await
            {
                tracing::warn!(
                    user_id = %user_id,
                    "Failed to queue admin email-change notice: {}",
                    e
                );
            }
        }
    }
}

/// Serialises every test that plants or purges rows past the retention
/// window.
///
/// [`EmailChangeService::cleanup_expired`] deletes **every** such row, whoever
/// owns it, so two retention tests running concurrently would purge each
/// other's fixtures. Shared with `tasks::pending_email_change_cleanup`, which
/// drives the same predicate from its processor.
#[cfg(test)]
pub(crate) static RETENTION_GUARD: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

#[cfg(test)]
mod tests {
    use super::*;
    use crate::models::{email_queue, refresh_token};
    use crate::services::role_service::RoleService;
    use async_trait::async_trait;
    use sea_orm::{PaginatorTrait, QueryOrder};
    use std::sync::Mutex;

    /// Base URL of the links built by the service; the tests parse the token
    /// back out of the queued bodies, exactly like a human clicking them.
    const TEST_BASE_URL: &str = "https://rw.test";

    /// One captured `on_audit_event` call, owned so the assertions can outlive
    /// the borrowed [`AuditEvent`].
    #[derive(Clone, Debug)]
    struct CapturedEvent {
        kind: String,
        actor_id: Option<Uuid>,
        target: Option<(String, Uuid)>,
        metadata: serde_json::Value,
    }

    type AuditSink = Arc<Mutex<Vec<CapturedEvent>>>;
    type VerifiedSink = Arc<Mutex<Vec<Uuid>>>;

    /// Hooks impl forwarding every audit event to an externally-owned sink —
    /// same shape as `auth_logging::tests::CapturingHooks`, so no downcast is
    /// needed once the value is erased behind `Arc<dyn RustwardenHooks>`.
    struct CapturingHooks {
        audit: AuditSink,
        verified: VerifiedSink,
    }

    #[async_trait]
    impl RustwardenHooks for CapturingHooks {
        async fn on_audit_event(
            &self,
            _db: &DatabaseConnection,
            event: AuditEvent<'_>,
        ) -> anyhow::Result<()> {
            self.audit.lock().expect("audit sink").push(CapturedEvent {
                kind: event.kind.to_string(),
                actor_id: event.actor_id,
                target: event.target.map(|(kind, id)| (kind.to_string(), id)),
                metadata: event.metadata.clone(),
            });
            Ok(())
        }

        async fn on_email_verified(
            &self,
            _db: &DatabaseConnection,
            user_id: Uuid,
        ) -> anyhow::Result<()> {
            self.verified.lock().expect("verified sink").push(user_id);
            Ok(())
        }
    }

    fn capturing_hooks() -> (Arc<dyn RustwardenHooks>, AuditSink, VerifiedSink) {
        let audit = Arc::new(Mutex::new(Vec::new()));
        let verified = Arc::new(Mutex::new(Vec::new()));
        let hooks: Arc<dyn RustwardenHooks> = Arc::new(CapturingHooks {
            audit: audit.clone(),
            verified: verified.clone(),
        });
        (hooks, audit, verified)
    }

    /// Per-test connection — see `personal_token_service::tests` for rationale.
    async fn test_db() -> DatabaseConnection {
        dotenvy::dotenv().ok();
        let url = std::env::var("DATABASE_URL")
            .expect("DATABASE_URL required for email_change_service tests");
        sea_orm::Database::connect(&url)
            .await
            .expect("connect to the test database")
    }

    fn dummy_email_config() -> EmailConfig {
        EmailConfig {
            smtp_host: "localhost".into(),
            smtp_port: 587,
            smtp_user: "u".into(),
            smtp_password: "p".into(),
            smtp_from: "from@example.test".into(),
            smtp_from_name: Some("Rustwarden Test".into()),
            smtp_use_tls: false,
            smtp_use_starttls: false,
            smtp_accept_invalid_certs: true,
            smtp_ca_cert_path: None,
            queue_batch_size: 10,
            queue_retry_attempts: 5,
            queue_retry_delay_seconds: 300,
            email_base_url: TEST_BASE_URL.into(),
            request_dsn: false,
            imap_host: None,
            imap_port: 993,
            imap_user: None,
            imap_password: None,
            imap_use_tls: true,
            imap_use_starttls: false,
            imap_accept_invalid_certs: false,
            imap_ca_cert_path: None,
            imap_mailbox: "INBOX".into(),
            imap_poll_interval_seconds: 300,
        }
    }

    /// Service backed by a **private** limiter: the process-wide `OnceLock`
    /// would otherwise leak budgets between tests running in the same binary.
    fn service() -> EmailChangeService {
        EmailChangeService::new_with_limiter(dummy_email_config(), RateLimitService::new())
    }

    const PASSWORD: &str = "TestPassword1!";

    /// Create a **verified** account, the realistic starting point: RFC 0041
    /// protects an identity that has already been proven.
    async fn verified_user(db: &DatabaseConnection) -> user::Model {
        let unique = Uuid::new_v4().simple().to_string();
        let hash = crate::services::password_service::PasswordService::hash_password(PASSWORD)
            .await
            .expect("hash password");
        let created = UserService::create_user(
            db,
            format!("ec-{}", &unique[..10]),
            format!("ec-{}@example.test", &unique[..10]),
            hash,
            false,
        )
        .await
        .expect("create user");

        let mut active: user::ActiveModel = created.into();
        active.email_verified = Set(true);
        active.email_verified_at = Set(Some(Utc::now()));
        active.update(db).await.expect("mark verified")
    }

    /// Address nobody owns yet, unique per call.
    fn fresh_email() -> String {
        format!("cand-{}@example.test", Uuid::new_v4().simple())
    }

    /// Latest queued mail of `email_type` addressed to `to_email`.
    async fn last_queued(
        db: &DatabaseConnection,
        to_email: &str,
        email_type: &str,
    ) -> Option<email_queue::Model> {
        email_queue::Entity::find()
            .filter(email_queue::Column::ToEmail.eq(to_email))
            .filter(email_queue::Column::EmailType.eq(email_type))
            .order_by_desc(email_queue::Column::CreatedAt)
            .one(db)
            .await
            .expect("query email_queue")
    }

    async fn queued_count(db: &DatabaseConnection, to_email: &str, email_type: &str) -> u64 {
        email_queue::Entity::find()
            .filter(email_queue::Column::ToEmail.eq(to_email))
            .filter(email_queue::Column::EmailType.eq(email_type))
            .count(db)
            .await
            .expect("count email_queue")
    }

    /// Pull the token back out of a queued body — the tests follow the same
    /// path as the human: read the mail, take the link.
    fn token_in(body: &str, kind: &str) -> String {
        let marker = format!("{TEST_BASE_URL}/api/v1/auth/email-change/{kind}/");
        let start = body
            .find(&marker)
            .unwrap_or_else(|| panic!("no {kind} link in body: {body}"))
            + marker.len();
        body[start..]
            .chars()
            .take_while(|c| c.is_ascii_alphanumeric() || *c == '-')
            .collect()
    }

    async fn confirmation_token(db: &DatabaseConnection, candidate: &str) -> String {
        let mail = last_queued(db, candidate, "email_change_validation")
            .await
            .expect("validation mail queued");
        token_in(&mail.body_text, "confirm")
    }

    async fn report_token(db: &DatabaseConnection, current: &str) -> String {
        let mail = last_queued(db, current, "email_change_notice")
            .await
            .expect("notice mail queued");
        token_in(&mail.body_text, "report")
    }

    async fn pending_rows(
        db: &DatabaseConnection,
        user_id: Uuid,
    ) -> Vec<pending_email_change::Model> {
        pending_email_change::Entity::find()
            .filter(pending_email_change::Column::UserId.eq(user_id))
            .order_by_asc(pending_email_change::Column::CreatedAt)
            .all(db)
            .await
            .expect("query pending_email_changes")
    }

    async fn audit_rows(
        db: &DatabaseConnection,
        user_id: Uuid,
        action: &str,
    ) -> Vec<audit_log::Model> {
        audit_log::Entity::find()
            .filter(
                Condition::any()
                    .add(audit_log::Column::UserId.eq(user_id))
                    .add(audit_log::Column::TargetUserId.eq(user_id)),
            )
            .filter(audit_log::Column::Action.eq(action))
            .all(db)
            .await
            .expect("query audit_logs")
    }

    async fn reload(db: &DatabaseConnection, user_id: Uuid) -> user::Model {
        UserService::get_user_by_id(db, user_id)
            .await
            .expect("reload user")
            .expect("user still exists")
    }

    /// Give `user_id` the `admin` role. Returns `false` when the RBAC seed is
    /// absent, so the admin tests skip instead of failing on a bare schema.
    async fn promote_to_admin(db: &DatabaseConnection, user_id: Uuid) -> bool {
        match RoleService::get_role_by_name(db, "admin").await {
            Ok(Some(role)) => {
                UserService::assign_role_to_user(db, user_id, role.id)
                    .await
                    .expect("assign admin role");
                true
            }
            _ => false,
        }
    }

    /// Force a pending row into the past without going through the service.
    async fn expire(db: &DatabaseConnection, row: pending_email_change::Model) {
        let mut active: pending_email_change::ActiveModel = row.into();
        active.expires_at = Set(Utc::now() - Duration::hours(1));
        active.update(db).await.expect("expire row");
    }

    // ── Unit: token hashing ─────────────────────────────────────────────────

    #[test]
    fn hash_token_is_deterministic_lowercase_sha256_hex() {
        let hash = EmailChangeService::hash_token("abc");
        assert_eq!(hash.len(), 64, "SHA-256 hex is 64 characters");
        assert!(
            hash.chars()
                .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()),
            "hash must be lowercase hex, got {hash}"
        );
        assert_eq!(
            hash,
            EmailChangeService::hash_token("abc"),
            "hashing must be deterministic"
        );
    }

    #[test]
    fn hash_token_differs_for_two_inputs() {
        assert_ne!(
            EmailChangeService::hash_token("token-a"),
            EmailChangeService::hash_token("token-b")
        );
    }

    #[test]
    fn hash_token_never_returns_the_plain_token() {
        let plain = Uuid::new_v4().to_string();
        assert_ne!(EmailChangeService::hash_token(&plain), plain);
    }

    // ── Unit: rate limiter wiring ───────────────────────────────────────────

    #[test]
    fn new_with_limiter_uses_injected_instance() {
        let limiter = RateLimitService::new();
        let svc = EmailChangeService::new_with_limiter(dummy_email_config(), limiter.clone());
        let key = format!("email_change:{}", Uuid::new_v4());
        for _ in 0..EMAIL_CHANGE_RATE_LIMIT_MAX {
            limiter
                .check_sliding_window(
                    &key,
                    EMAIL_CHANGE_RATE_LIMIT_MAX,
                    EMAIL_CHANGE_RATE_LIMIT_WINDOW,
                )
                .expect("budget not exhausted yet");
        }
        assert!(
            svc.rate_limiter
                .check_sliding_window(
                    &key,
                    EMAIL_CHANGE_RATE_LIMIT_MAX,
                    EMAIL_CHANGE_RATE_LIMIT_WINDOW
                )
                .is_err(),
            "the service must see the budget consumed through the injected limiter"
        );
    }

    #[test]
    fn new_shares_the_process_wide_limiter() {
        let a = EmailChangeService::new(dummy_email_config());
        let b = EmailChangeService::new(dummy_email_config());
        let key = format!("email_change:{}", Uuid::new_v4());
        for _ in 0..EMAIL_CHANGE_RATE_LIMIT_MAX {
            a.rate_limiter
                .check_sliding_window(
                    &key,
                    EMAIL_CHANGE_RATE_LIMIT_MAX,
                    EMAIL_CHANGE_RATE_LIMIT_WINDOW,
                )
                .expect("budget not exhausted yet");
        }
        assert!(
            b.rate_limiter
                .check_sliding_window(
                    &key,
                    EMAIL_CHANGE_RATE_LIMIT_MAX,
                    EMAIL_CHANGE_RATE_LIMIT_WINDOW
                )
                .is_err(),
            "two `new()` services must share one process-wide budget"
        );
    }

    #[test]
    fn email_change_budget_is_distinct_from_verify_and_reset() {
        let limiter = RateLimitService::new();
        let id = Uuid::new_v4();
        let window = EMAIL_CHANGE_RATE_LIMIT_WINDOW;
        for _ in 0..EMAIL_CHANGE_RATE_LIMIT_MAX {
            limiter
                .check_sliding_window(
                    &format!("email_change:{id}"),
                    EMAIL_CHANGE_RATE_LIMIT_MAX,
                    window,
                )
                .expect("email_change budget");
        }
        // Same subject, other prefixes: untouched. A user who exhausted their
        // change budget must still be able to reset their password.
        assert!(
            limiter
                .check_sliding_window(
                    &format!("email_verify:{id}"),
                    EMAIL_CHANGE_RATE_LIMIT_MAX,
                    window
                )
                .is_ok(),
            "email_verify budget must not share the email_change key"
        );
        assert!(
            limiter
                .check_sliding_window(
                    &format!("pwd_reset:{id}"),
                    EMAIL_CHANGE_RATE_LIMIT_MAX,
                    window
                )
                .is_ok(),
            "pwd_reset budget must not share the email_change key"
        );
    }

    /// Count the sessions still usable by `user_id`.
    async fn live_sessions(db: &DatabaseConnection, user_id: Uuid) -> u64 {
        refresh_token::Entity::find()
            .filter(refresh_token::Column::UserId.eq(user_id))
            .filter(refresh_token::Column::Revoked.eq(false))
            .count(db)
            .await
            .expect("count refresh tokens")
    }

    // ── request_change ──────────────────────────────────────────────────────

    /// RFC 0041 test 1 — the whole point of the RFC: a request changes nothing
    /// on the account.
    #[tokio::test]
    async fn request_change_writes_a_pending_row_and_leaves_the_account_untouched() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();

        service()
            .request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request must succeed");

        let rows = pending_rows(&db, user.id).await;
        assert_eq!(rows.len(), 1, "exactly one pending row");
        assert_eq!(rows[0].new_email, candidate);
        assert!(rows[0].used_at.is_none(), "a fresh request is not consumed");
        assert!(rows[0].reported_at.is_none());
        assert!(rows[0].is_pending(), "the row must be usable");
        assert_ne!(
            rows[0].token_hash, rows[0].report_token_hash,
            "confirming and reporting are two distinct capabilities"
        );

        let reloaded = reload(&db, user.id).await;
        assert_eq!(
            reloaded.email, user.email,
            "users.email must NOT move before confirmation"
        );
        assert!(
            reloaded.email_verified,
            "a pending request must not un-verify the account"
        );

        assert_eq!(
            queued_count(&db, &candidate, "email_change_validation").await,
            1,
            "one confirmation mail to the candidate address"
        );
        assert_eq!(
            queued_count(&db, &user.email, "email_change_notice").await,
            1,
            "one notice mail to the current address"
        );
    }

    /// The notice must never carry the confirmation link, and must not spell
    /// out the candidate address: the old mailbox may already be hostile.
    #[tokio::test]
    async fn the_notice_carries_the_report_link_only_and_masks_the_candidate() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();

        service()
            .request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request must succeed");

        let notice = last_queued(&db, &user.email, "email_change_notice")
            .await
            .expect("notice queued");
        let confirm_token = confirmation_token(&db, &candidate).await;

        for body in [&notice.body_text, &notice.body_html] {
            assert!(
                !body.contains(&confirm_token),
                "the notice must never carry the confirmation token"
            );
            assert!(
                !body.contains("/email-change/confirm/"),
                "the notice must never carry a confirmation link"
            );
            assert!(
                !body.contains(&candidate),
                "the candidate address must be masked in the notice"
            );
            assert!(
                body.contains("/email-change/report/"),
                "the notice must carry the report link"
            );
        }
    }

    /// RFC 0041 test 3 — a stolen access token alone must not open a change.
    #[tokio::test]
    async fn request_change_with_a_wrong_password_writes_nothing() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();

        let result = service()
            .request_change(&db, user.id, &candidate, "WrongPassword1!")
            .await;

        assert!(
            matches!(result, Err(AppError::InvalidCredentials)),
            "expected InvalidCredentials, got {result:?}"
        );
        assert!(
            pending_rows(&db, user.id).await.is_empty(),
            "no row may exist after a refused request"
        );
        assert_eq!(
            queued_count(&db, &candidate, "email_change_validation").await,
            0,
            "no mail may reach the candidate address"
        );
    }

    /// RFC 0041 test 4 — the format is checked before anything is written.
    #[tokio::test]
    async fn request_change_with_an_invalid_format_writes_nothing() {
        let db = test_db().await;
        let user = verified_user(&db).await;

        let result = service()
            .request_change(&db, user.id, "not-an-email", PASSWORD)
            .await;

        assert!(
            matches!(result, Err(AppError::Validation(_))),
            "expected Validation, got {result:?}"
        );
        assert!(pending_rows(&db, user.id).await.is_empty());
        assert_eq!(reload(&db, user.id).await.email, user.email);
    }

    /// RFC 0041 test 5 — an address already owned by another account.
    #[tokio::test]
    async fn request_change_on_a_taken_address_is_refused() {
        let db = test_db().await;
        let alice = verified_user(&db).await;
        let bob = verified_user(&db).await;

        let result = service()
            .request_change(&db, alice.id, &bob.email, PASSWORD)
            .await;

        assert!(
            matches!(result, Err(AppError::Validation(_))),
            "expected Validation, got {result:?}"
        );
        assert!(pending_rows(&db, alice.id).await.is_empty());
    }

    /// RFC 0041 test 6 — echoing the current address is a refusal here, not a
    /// silent no-op: `request_change` expresses an explicit intent, unlike a
    /// profile form that resubmits every field.
    #[tokio::test]
    async fn request_change_on_the_current_address_is_refused() {
        let db = test_db().await;
        let user = verified_user(&db).await;

        let result = service()
            .request_change(&db, user.id, &user.email, PASSWORD)
            .await;

        assert!(
            matches!(result, Err(AppError::Validation(_))),
            "expected Validation, got {result:?}"
        );
        assert!(pending_rows(&db, user.id).await.is_empty());
        assert_eq!(
            queued_count(&db, &user.email, "email_change_notice").await,
            0,
            "a refused request must not mail anybody"
        );
    }

    /// RFC 0041 test 7 — the per-user budget bounds "one account spamming
    /// third-party mailboxes".
    #[tokio::test]
    async fn the_fourth_request_within_the_hour_is_rate_limited() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let svc = service();

        for i in 1..=EMAIL_CHANGE_RATE_LIMIT_MAX {
            svc.request_change(&db, user.id, &fresh_email(), PASSWORD)
                .await
                .unwrap_or_else(|e| panic!("request {i} must succeed: {e:?}"));
        }

        let candidate = fresh_email();
        let result = svc.request_change(&db, user.id, &candidate, PASSWORD).await;
        match result {
            Err(AppError::TooManyRequests { retry_after_secs }) => assert!(
                retry_after_secs > 0,
                "a 429 must carry an actionable Retry-After"
            ),
            other => panic!("expected TooManyRequests, got {other:?}"),
        }
        assert_eq!(
            queued_count(&db, &candidate, "email_change_validation").await,
            0,
            "a throttled request must not mail the candidate address"
        );
    }

    /// RFC 0041 test 10 — a second request consumes the first: two live links
    /// would mean two ways to move the account's identity.
    #[tokio::test]
    async fn a_new_request_consumes_the_previous_one() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let svc = service();

        let first_candidate = fresh_email();
        svc.request_change(&db, user.id, &first_candidate, PASSWORD)
            .await
            .expect("first request");
        let first_token = confirmation_token(&db, &first_candidate).await;

        let second_candidate = fresh_email();
        svc.request_change(&db, user.id, &second_candidate, PASSWORD)
            .await
            .expect("second request");

        assert_eq!(
            EmailChangeService::inspect_confirmation_token(&db, &first_token)
                .await
                .expect("inspect"),
            EmailChangeLinkState::Used,
            "the superseded link must be dead"
        );
        assert!(
            svc.confirm_change(&db, &first_token, None).await.is_err(),
            "the superseded token must not confirm"
        );
        assert_eq!(
            reload(&db, user.id).await.email,
            user.email,
            "a dead token must not move the address"
        );
    }

    // ── confirm_change ──────────────────────────────────────────────────────

    /// RFC 0041 test 2 — the happy path, including the session closure the
    /// flow promises.
    #[tokio::test]
    async fn confirm_change_applies_the_address_and_closes_every_session() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();
        let svc = service();

        RefreshTokenService::create_refresh_token(&db, user.id, 30)
            .await
            .expect("seed a live session");

        svc.request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request");
        let token = confirmation_token(&db, &candidate).await;

        let changed_id = svc
            .confirm_change(&db, &token, None)
            .await
            .expect("confirmation must succeed");
        assert_eq!(changed_id, user.id);

        let reloaded = reload(&db, user.id).await;
        assert_eq!(
            reloaded.email, candidate,
            "the new address is now the account's"
        );
        assert!(
            reloaded.email_verified,
            "clicking the link IS the proof of possession"
        );
        assert!(reloaded.email_verified_at.is_some());

        let rows = pending_rows(&db, user.id).await;
        assert_eq!(rows.len(), 1);
        assert!(rows[0].used_at.is_some(), "the row must be consumed");

        assert_eq!(
            live_sessions(&db, user.id).await,
            0,
            "the login identity moved: every session must re-authenticate"
        );
        assert_eq!(
            audit_rows(&db, user.id, "email_changed").await.len(),
            1,
            "exactly one audit row"
        );
    }

    /// RFC 0041 test 8 — the three refusal shapes, each leaving the account
    /// untouched.
    #[tokio::test]
    async fn confirm_change_refuses_unknown_expired_and_used_tokens() {
        let db = test_db().await;
        let svc = service();

        // Unknown.
        let unknown = Uuid::new_v4().to_string();
        assert!(matches!(
            svc.confirm_change(&db, &unknown, None).await,
            Err(AppError::Validation(_))
        ));
        assert_eq!(
            EmailChangeService::inspect_confirmation_token(&db, &unknown)
                .await
                .expect("inspect"),
            EmailChangeLinkState::Unknown
        );

        // Expired.
        let expired_user = verified_user(&db).await;
        let expired_candidate = fresh_email();
        svc.request_change(&db, expired_user.id, &expired_candidate, PASSWORD)
            .await
            .expect("request");
        let expired_token = confirmation_token(&db, &expired_candidate).await;
        expire(&db, pending_rows(&db, expired_user.id).await.remove(0)).await;
        assert!(matches!(
            svc.confirm_change(&db, &expired_token, None).await,
            Err(AppError::Validation(_))
        ));
        assert_eq!(
            EmailChangeService::inspect_confirmation_token(&db, &expired_token)
                .await
                .expect("inspect"),
            EmailChangeLinkState::Expired
        );
        assert_eq!(
            reload(&db, expired_user.id).await.email,
            expired_user.email,
            "an expired link must not move the address"
        );

        // Already used.
        let used_user = verified_user(&db).await;
        let used_candidate = fresh_email();
        svc.request_change(&db, used_user.id, &used_candidate, PASSWORD)
            .await
            .expect("request");
        let used_token = confirmation_token(&db, &used_candidate).await;
        svc.confirm_change(&db, &used_token, None)
            .await
            .expect("first confirmation");
        assert!(matches!(
            svc.confirm_change(&db, &used_token, None).await,
            Err(AppError::Validation(_))
        ));
        assert_eq!(
            EmailChangeService::inspect_confirmation_token(&db, &used_token)
                .await
                .expect("inspect"),
            EmailChangeLinkState::Used
        );
        assert_eq!(
            reload(&db, used_user.id).await.email,
            used_candidate,
            "replaying a used link must not change the address a second time"
        );
    }

    /// RFC 0041 test 9 — two accounts racing for the same address. The loser
    /// gets a neutral message and keeps a **reusable** row.
    #[tokio::test]
    async fn the_loser_of_a_race_keeps_a_reusable_row() {
        let db = test_db().await;
        let alice = verified_user(&db).await;
        let bob = verified_user(&db).await;
        let shared = fresh_email();
        let svc = service();

        svc.request_change(&db, alice.id, &shared, PASSWORD)
            .await
            .expect("alice requests");
        let alice_token = confirmation_token(&db, &shared).await;

        svc.request_change(&db, bob.id, &shared, PASSWORD)
            .await
            .expect("bob requests the same address");
        let bob_token = confirmation_token(&db, &shared).await;
        assert_ne!(alice_token, bob_token, "two requests, two secrets");

        svc.confirm_change(&db, &alice_token, None)
            .await
            .expect("alice confirms first");

        match svc.confirm_change(&db, &bob_token, None).await {
            Err(AppError::Validation(msg)) => assert!(
                !msg.contains(&alice.username) && !msg.contains(&alice.email),
                "the refusal must not name the winning account: {msg}"
            ),
            other => panic!("expected Validation, got {other:?}"),
        }
        assert_eq!(
            reload(&db, bob.id).await.email,
            bob.email,
            "bob keeps his address"
        );
        let bob_rows = pending_rows(&db, bob.id).await;
        assert_eq!(bob_rows.len(), 1);
        assert!(
            bob_rows[0].used_at.is_none(),
            "losing the race must not burn bob's link: it works again if the address frees up"
        );
    }

    /// RFC 0041 test 17 — a password-change link mailed to the previous
    /// address must die with that address.
    #[tokio::test]
    async fn confirm_change_drops_pending_password_change_requests() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();
        let password_service = PasswordChangeService::new();

        password_service
            .create_pending_request(&db, user.id, "AnotherPassword1!")
            .await
            .expect("seed a pending password change");
        assert!(
            password_service
                .find_pending_by_user_id(&db, user.id)
                .await
                .expect("lookup")
                .is_some(),
            "precondition: a pending password change exists"
        );

        let svc = service();
        svc.request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request");
        let token = confirmation_token(&db, &candidate).await;
        svc.confirm_change(&db, &token, None)
            .await
            .expect("confirmation");

        assert!(
            password_service
                .find_pending_by_user_id(&db, user.id)
                .await
                .expect("lookup")
                .is_none(),
            "the pending password change must be gone"
        );
    }

    /// RFC 0041 test 16 — hooks dispatched on confirmation, addresses masked.
    #[tokio::test]
    async fn confirm_change_dispatches_audit_and_email_verified_hooks() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();
        let (hooks, audit, verified) = capturing_hooks();
        let svc = service();

        svc.request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request");
        let token = confirmation_token(&db, &candidate).await;
        svc.confirm_change(&db, &token, Some(&hooks))
            .await
            .expect("confirmation");

        let events = audit.lock().expect("audit sink").clone();
        let changed = events
            .iter()
            .find(|event| event.kind == "user.email_changed")
            .unwrap_or_else(|| panic!("user.email_changed must be dispatched, got {events:?}"));
        assert_eq!(
            changed.target,
            Some(("user".to_string(), user.id)),
            "the event must name the account whose address moved"
        );
        let metadata = changed.metadata.to_string();
        assert!(
            !metadata.contains(&candidate) && !metadata.contains(&user.email),
            "hook metadata must carry masked addresses only: {metadata}"
        );
        assert!(
            verified.lock().expect("verified sink").contains(&user.id),
            "the click is the proof of possession: on_email_verified must fire"
        );
    }

    /// The two secrets are not interchangeable.
    #[tokio::test]
    async fn a_report_token_cannot_confirm_and_a_confirmation_token_cannot_report() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();
        let svc = service();

        svc.request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request");
        let confirm = confirmation_token(&db, &candidate).await;
        let report = report_token(&db, &user.email).await;

        assert!(
            svc.confirm_change(&db, &report, None).await.is_err(),
            "the report token must not confirm the change"
        );
        assert!(
            svc.report_suspicious(&db, &confirm, None).await.is_err(),
            "the confirmation token must not file a report"
        );
        assert_eq!(reload(&db, user.id).await.email, user.email);
        let rows = pending_rows(&db, user.id).await;
        // The admin fan-out sits behind the `reported_at` write, so an
        // unflagged row is also a row that woke nobody. Counting admin mails
        // directly would race with the other tests sharing this database.
        assert!(rows[0].used_at.is_none() && rows[0].reported_at.is_none());
    }

    /// RFC 0041 test 15 — deleting the account takes its pending requests with
    /// it (FK `ON DELETE CASCADE`).
    #[tokio::test]
    async fn deleting_a_user_cascades_to_their_pending_requests() {
        let db = test_db().await;
        let user = verified_user(&db).await;

        service()
            .request_change(&db, user.id, &fresh_email(), PASSWORD)
            .await
            .expect("request");
        assert_eq!(pending_rows(&db, user.id).await.len(), 1);

        UserService::delete_user(&db, user.id, None)
            .await
            .expect("delete user");

        assert!(
            pending_rows(&db, user.id).await.is_empty(),
            "pending rows must not survive their owner"
        );
    }

    // ── report_suspicious ───────────────────────────────────────────────────

    /// RFC 0041 tests 11-13 — the report alerts, traces, and does **not**
    /// block: the mailbox receiving this link may alert, never decide.
    #[tokio::test]
    async fn report_suspicious_alerts_and_traces_without_blocking_the_change() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();
        let (hooks, audit, _) = capturing_hooks();
        let svc = service();

        svc.request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request");
        let report = report_token(&db, &user.email).await;
        let confirm = confirmation_token(&db, &candidate).await;

        let reported_id = svc
            .report_suspicious(&db, &report, Some(&hooks))
            .await
            .expect("report must succeed");
        assert_eq!(reported_id, user.id);

        let rows = pending_rows(&db, user.id).await;
        assert!(rows[0].reported_at.is_some(), "the row must be flagged");
        assert!(
            rows[0].used_at.is_none(),
            "a report must not consume the request"
        );

        let audits = audit_rows(&db, user.id, "email_change_reported").await;
        assert_eq!(audits.len(), 1, "exactly one audit row");
        assert_eq!(audits[0].level, "WARN", "a report is a warning-level event");
        assert!(
            !audits[0].message.contains(&candidate),
            "the audit message must carry masked addresses only"
        );

        let events = audit.lock().expect("audit sink").clone();
        assert!(
            events
                .iter()
                .any(|event| event.kind == "user.email_change_reported"),
            "the audit hook must be dispatched, got {events:?}"
        );

        // The decisive assertion: a report does not veto the change.
        svc.confirm_change(&db, &confirm, None)
            .await
            .expect("a reported request must still be confirmable");
        assert_eq!(reload(&db, user.id).await.email, candidate);
    }

    /// Double click, page refresh, browser form resend: the second report is a
    /// no-op, not a second wake-up call for every administrator.
    #[tokio::test]
    async fn replaying_a_report_is_idempotent() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let svc = service();

        svc.request_change(&db, user.id, &fresh_email(), PASSWORD)
            .await
            .expect("request");
        let report = report_token(&db, &user.email).await;

        let first_id = svc
            .report_suspicious(&db, &report, None)
            .await
            .expect("first report");
        let first_at = pending_rows(&db, user.id).await[0]
            .reported_at
            .expect("the row must be flagged");

        let second_id = svc
            .report_suspicious(&db, &report, None)
            .await
            .expect("a replay must not fail");

        assert_eq!(second_id, first_id);
        assert_eq!(
            pending_rows(&db, user.id).await[0].reported_at,
            Some(first_at),
            "the first report timestamp stands"
        );
        assert_eq!(
            audit_rows(&db, user.id, "email_change_reported")
                .await
                .len(),
            1,
            "replaying must not wake every administrator twice"
        );
    }

    /// RFC 0041 test 14 — reporting an already-consumed request stays useful:
    /// the operator signal keeps its value after the fact.
    #[tokio::test]
    async fn a_consumed_request_can_still_be_reported() {
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();
        let svc = service();

        svc.request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request");
        let report = report_token(&db, &user.email).await;
        let confirm = confirmation_token(&db, &candidate).await;
        svc.confirm_change(&db, &confirm, None)
            .await
            .expect("confirmation");

        svc.report_suspicious(&db, &report, None)
            .await
            .expect("reporting after the fact must still work");

        let rows = pending_rows(&db, user.id).await;
        assert!(rows[0].used_at.is_some() && rows[0].reported_at.is_some());
        assert_eq!(
            audit_rows(&db, user.id, "email_change_reported")
                .await
                .len(),
            1,
            "the operator is alerted even though the change already landed"
        );
        assert_eq!(
            reload(&db, user.id).await.email,
            candidate,
            "a late report does not roll the change back"
        );
    }

    // ── admin_force_change ──────────────────────────────────────────────────

    /// RFC 0041 tests 23-25 — the operator recourse: a write without proof of
    /// possession, traced as a warning and never silent.
    #[tokio::test]
    async fn admin_force_change_writes_without_proof_and_traces_it_as_a_warning() {
        let db = test_db().await;
        let admin = verified_user(&db).await;
        if !promote_to_admin(&db, admin.id).await {
            eprintln!("skipped: no `admin` role in this database (RBAC seed absent)");
            return;
        }
        let target = verified_user(&db).await;
        let fixed = fresh_email();
        let (hooks, audit, verified) = capturing_hooks();

        RefreshTokenService::create_refresh_token(&db, target.id, 30)
            .await
            .expect("seed a live session");

        service()
            .admin_force_change(&db, admin.id, target.id, &fixed, Some(&hooks))
            .await
            .expect("the override must succeed");

        let reloaded = reload(&db, target.id).await;
        assert_eq!(reloaded.email, fixed);
        assert!(
            reloaded.email_verified,
            "the operator vouches for the address out of band"
        );
        assert_eq!(
            live_sessions(&db, target.id).await,
            0,
            "the target's sessions must close, like any identity move"
        );

        let audits = audit_rows(&db, target.id, "email_changed_by_admin").await;
        assert_eq!(audits.len(), 1, "exactly one audit row");
        assert_eq!(
            audits[0].level, "WARN",
            "an identity write without proof must surface"
        );
        assert_eq!(
            audits[0].user_id,
            Some(admin.id),
            "the acting operator must be named"
        );
        assert!(
            !audits[0].message.contains(&fixed) && !audits[0].message.contains(&target.email),
            "the audit message must carry masked addresses only"
        );

        let events = audit.lock().expect("audit sink").clone();
        let forced = events
            .iter()
            .find(|event| event.kind == "user.email_changed_by_admin")
            .unwrap_or_else(|| {
                panic!("user.email_changed_by_admin must be dispatched, got {events:?}")
            });
        assert_eq!(
            forced.actor_id,
            Some(admin.id),
            "the event must name the acting operator"
        );
        assert_eq!(
            forced.target,
            Some(("user".to_string(), target.id)),
            "the event must name the account that was rewritten"
        );
        assert!(
            !verified.lock().expect("verified sink").contains(&target.id),
            "nobody proved possession: on_email_verified must NOT fire"
        );

        // Both mailboxes informed; there is nothing left to confirm, so
        // neither mail carries an action link.
        for mailbox in [target.email.as_str(), fixed.as_str()] {
            let mail = last_queued(&db, mailbox, "email_change_by_admin")
                .await
                .unwrap_or_else(|| panic!("no notice queued for {mailbox}"));
            for body in [&mail.body_text, &mail.body_html] {
                assert!(
                    !body.contains("/email-change/confirm/")
                        && !body.contains("/email-change/report/"),
                    "an operator-driven change leaves nothing to click"
                );
            }
        }
    }

    /// RFC 0041 test 26 — the route middleware is not the only gate: a
    /// headless call straight to the service is refused too.
    #[tokio::test]
    async fn admin_force_change_refuses_a_non_admin_actor() {
        let db = test_db().await;
        let actor = verified_user(&db).await;
        let target = verified_user(&db).await;
        let candidate = fresh_email();

        let result = service()
            .admin_force_change(&db, actor.id, target.id, &candidate, None)
            .await;

        assert!(
            matches!(result, Err(AppError::Forbidden)),
            "expected Forbidden, got {result:?}"
        );
        assert_eq!(
            reload(&db, target.id).await.email,
            target.email,
            "nothing may be written"
        );
        assert!(
            audit_rows(&db, target.id, "email_changed_by_admin")
                .await
                .is_empty(),
            "a refused override leaves no trail to read"
        );
    }

    /// RFC 0041 test 27 — a still-valid confirmation link must not overwrite
    /// the address the operator just set.
    #[tokio::test]
    async fn admin_force_change_kills_a_live_confirmation_link() {
        let db = test_db().await;
        let admin = verified_user(&db).await;
        if !promote_to_admin(&db, admin.id).await {
            eprintln!("skipped: no `admin` role in this database (RBAC seed absent)");
            return;
        }
        let target = verified_user(&db).await;
        let candidate = fresh_email();
        let fixed = fresh_email();
        let svc = service();

        svc.request_change(&db, target.id, &candidate, PASSWORD)
            .await
            .expect("the user opens a request");
        let token = confirmation_token(&db, &candidate).await;

        svc.admin_force_change(&db, admin.id, target.id, &fixed, None)
            .await
            .expect("the operator overrides it");

        assert_eq!(
            EmailChangeService::inspect_confirmation_token(&db, &token)
                .await
                .expect("inspect"),
            EmailChangeLinkState::Used,
            "the pending request must be consumed by the override"
        );
        assert!(
            svc.confirm_change(&db, &token, None).await.is_err(),
            "a stale link must not resurrect the previous candidate address"
        );
        assert_eq!(reload(&db, target.id).await.email, fixed);
    }

    /// Admin forms resubmit every field: echoing the current address must not
    /// pollute the trail with a change that did not happen.
    #[tokio::test]
    async fn admin_force_change_on_the_current_address_is_a_silent_noop() {
        let db = test_db().await;
        let admin = verified_user(&db).await;
        if !promote_to_admin(&db, admin.id).await {
            eprintln!("skipped: no `admin` role in this database (RBAC seed absent)");
            return;
        }
        let target = verified_user(&db).await;

        service()
            .admin_force_change(&db, admin.id, target.id, &target.email, None)
            .await
            .expect("echoing the current address must not fail");

        assert_eq!(reload(&db, target.id).await.email, target.email);
        assert!(
            audit_rows(&db, target.id, "email_changed_by_admin")
                .await
                .is_empty(),
            "a no-op must not be traced as a change"
        );
        assert_eq!(
            queued_count(&db, &target.email, "email_change_by_admin").await,
            0,
            "a no-op must not mail anybody"
        );
    }

    /// RFC 0041 test 26 (suite) — an unknown actor is a `NotFound`, not a
    /// silent `Forbidden`: the two say different things to an operator.
    #[tokio::test]
    async fn admin_force_change_with_an_unknown_actor_is_not_found() {
        let db = test_db().await;
        let target = verified_user(&db).await;

        let result = service()
            .admin_force_change(&db, Uuid::new_v4(), target.id, &fresh_email(), None)
            .await;

        assert!(
            matches!(result, Err(AppError::NotFound(_))),
            "expected NotFound, got {result:?}"
        );
        assert_eq!(reload(&db, target.id).await.email, target.email);
    }

    /// RFC 0041 test 27 — the operator recourse is not a licence to duplicate
    /// an address: uniqueness still holds.
    #[tokio::test]
    async fn admin_force_change_on_a_taken_address_is_refused() {
        let db = test_db().await;
        let admin = verified_user(&db).await;
        if !promote_to_admin(&db, admin.id).await {
            eprintln!("skipped: no `admin` role in this database (RBAC seed absent)");
            return;
        }
        let target = verified_user(&db).await;
        let neighbour = verified_user(&db).await;

        let result = service()
            .admin_force_change(&db, admin.id, target.id, &neighbour.email, None)
            .await;

        assert!(
            matches!(result, Err(AppError::Validation(_))),
            "expected Validation, got {result:?}"
        );
        assert_eq!(reload(&db, target.id).await.email, target.email);
        assert_eq!(reload(&db, neighbour.id).await.email, neighbour.email);
    }

    /// RFC 0041 test 17, admin side — same reason as on the user path: a
    /// password-change link mailed to the previous address must not survive it.
    #[tokio::test]
    async fn admin_force_change_drops_pending_password_change_requests() {
        let db = test_db().await;
        let admin = verified_user(&db).await;
        if !promote_to_admin(&db, admin.id).await {
            eprintln!("skipped: no `admin` role in this database (RBAC seed absent)");
            return;
        }
        let target = verified_user(&db).await;
        let password_service = PasswordChangeService::new();

        password_service
            .create_pending_request(&db, target.id, "AnotherPassword1!")
            .await
            .expect("seed a pending password change");

        service()
            .admin_force_change(&db, admin.id, target.id, &fresh_email(), None)
            .await
            .expect("override");

        assert!(
            password_service
                .find_pending_by_user_id(&db, target.id)
                .await
                .expect("lookup")
                .is_none(),
            "the pending password change must be gone"
        );
    }

    /// The operator keeps working: only the target account is logged out.
    #[tokio::test]
    async fn admin_force_change_leaves_the_operators_own_sessions_alone() {
        let db = test_db().await;
        let admin = verified_user(&db).await;
        if !promote_to_admin(&db, admin.id).await {
            eprintln!("skipped: no `admin` role in this database (RBAC seed absent)");
            return;
        }
        let target = verified_user(&db).await;

        RefreshTokenService::create_refresh_token(&db, admin.id, 30)
            .await
            .expect("seed the operator's session");
        RefreshTokenService::create_refresh_token(&db, target.id, 30)
            .await
            .expect("seed the target's session");

        service()
            .admin_force_change(&db, admin.id, target.id, &fresh_email(), None)
            .await
            .expect("override");

        assert_eq!(
            live_sessions(&db, target.id).await,
            0,
            "the rewritten account must re-authenticate"
        );
        assert_eq!(
            live_sessions(&db, admin.id).await,
            1,
            "the operator must not be logged out by their own action"
        );
    }

    // ── Retention ───────────────────────────────────────────────────────────

    /// RFC 0041 test 28 — `new_email` is personal data in clear text: it must
    /// not outlive the retention window.
    #[tokio::test]
    async fn cleanup_expired_deletes_rows_past_the_retention_window() {
        let _serialised = RETENTION_GUARD.lock().await;
        let db = test_db().await;
        let user = verified_user(&db).await;

        service()
            .request_change(&db, user.id, &fresh_email(), PASSWORD)
            .await
            .expect("request");
        let mut active: pending_email_change::ActiveModel =
            pending_rows(&db, user.id).await.remove(0).into();
        active.expires_at =
            Set(Utc::now() - Duration::days(PENDING_EMAIL_CHANGE_RETENTION_DAYS + 1));
        active.update(&db).await.expect("age the row");

        EmailChangeService::cleanup_expired(&db)
            .await
            .expect("cleanup");

        assert!(
            pending_rows(&db, user.id).await.is_empty(),
            "a row past retention must be hard-deleted"
        );
    }

    /// RFC 0041 test 29 — a merely expired row stays: the user who clicks late
    /// deserves "this link expired", not "unknown token".
    #[tokio::test]
    async fn cleanup_expired_keeps_rows_still_within_retention() {
        let _serialised = RETENTION_GUARD.lock().await;
        let db = test_db().await;
        let user = verified_user(&db).await;
        let candidate = fresh_email();

        service()
            .request_change(&db, user.id, &candidate, PASSWORD)
            .await
            .expect("request");
        let token = confirmation_token(&db, &candidate).await;
        expire(&db, pending_rows(&db, user.id).await.remove(0)).await;

        EmailChangeService::cleanup_expired(&db)
            .await
            .expect("cleanup");

        assert_eq!(pending_rows(&db, user.id).await.len(), 1);
        assert_eq!(
            EmailChangeService::inspect_confirmation_token(&db, &token)
                .await
                .expect("inspect"),
            EmailChangeLinkState::Expired,
            "within retention the link explains itself instead of looking unknown"
        );
    }

    /// RFC 0041 test 30 — retention is honoured even by a consumer that never
    /// starts the cleanup processor: the next request purges the user's own
    /// stale rows, and only those.
    #[tokio::test]
    async fn a_new_request_purges_the_users_own_stale_rows_only() {
        let _serialised = RETENTION_GUARD.lock().await;
        let db = test_db().await;
        let user = verified_user(&db).await;
        let neighbour = verified_user(&db).await;
        let svc = service();

        for owner in [&user, &neighbour] {
            svc.request_change(&db, owner.id, &fresh_email(), PASSWORD)
                .await
                .expect("seed a request");
            let mut active: pending_email_change::ActiveModel =
                pending_rows(&db, owner.id).await.remove(0).into();
            active.expires_at =
                Set(Utc::now() - Duration::days(PENDING_EMAIL_CHANGE_RETENTION_DAYS + 1));
            active.update(&db).await.expect("age the row");
        }

        svc.request_change(&db, user.id, &fresh_email(), PASSWORD)
            .await
            .expect("a fresh request triggers the opportunistic purge");

        let rows = pending_rows(&db, user.id).await;
        assert_eq!(rows.len(), 1, "only the fresh request must remain");
        assert!(rows[0].is_pending());
        assert_eq!(
            pending_rows(&db, neighbour.id).await.len(),
            1,
            "the purge must stay scoped to the acting user"
        );
    }

    // ── Locales ─────────────────────────────────────────────────────────────

    /// Every string this service and the interstitial pages resolve, in the six
    /// shipped locales.
    ///
    /// A missing key is silently replaced by the English fallback baked into
    /// the code, so a gap never shows up as a failure at runtime — only as a
    /// German user reading an English sentence. This test is the only place
    /// that gap becomes visible.
    #[test]
    fn the_rfc_0041_keys_exist_in_the_six_locales() {
        use yaml_rust2::YamlLoader;

        const LOCALES: [(&str, &str); 6] = [
            ("de", include_str!("../../locales/de.yml")),
            ("en", include_str!("../../locales/en.yml")),
            ("es", include_str!("../../locales/es.yml")),
            ("fr", include_str!("../../locales/fr.yml")),
            ("it", include_str!("../../locales/it.yml")),
            ("pt", include_str!("../../locales/pt.yml")),
        ];

        let keys_of = |source: &str, block: &str| -> Vec<String> {
            let docs = YamlLoader::load_from_str(source).expect("valid YAML");
            let mut keys: Vec<String> = docs[0][block]
                .as_hash()
                .unwrap_or_else(|| panic!("block `{block}` is missing"))
                .keys()
                .filter_map(|k| k.as_str().map(str::to_string))
                .collect();
            keys.sort();
            keys
        };

        for block in ["email_change", "email_change_page"] {
            let reference = keys_of(LOCALES[1].1, block);
            assert!(
                !reference.is_empty(),
                "en.yml must define the `{block}` block"
            );
            for (locale, source) in LOCALES {
                assert_eq!(
                    keys_of(source, block),
                    reference,
                    "{locale}.yml must define exactly the `{block}` keys of en.yml"
                );
            }
        }
    }
}