use std::time::Duration;

use auth::KenAuthError;
use bootstrap::AppState;
use models::repo::repos;
use models::room::rooms;
use models::space::{
    SpaceStatus, space_extras, space_group_permissions, space_groups, space_invites,
    space_join_requests, space_join_rules, space_member_extras, space_member_groups, space_members,
    space_warn, spaces,
};
use models::system::{account_deletion, space_deletion};
use models::users::{
    user, user_agreement_acceptances, user_blocks, user_external_identity, user_favorite_memes,
    user_followers, user_notifications, user_preferences, user_profiles, user_push_subscriptions,
    user_recent_mentions, user_saved_messages,
};
use proto::KenShellClient;
use sea_orm::sea_query::Expr;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, Condition, ConnectionTrait, DbBackend, EntityTrait,
    PaginatorTrait, QueryFilter, QuerySelect, Set, Statement,
};
use serde::Serialize;
use time::OffsetDateTime;
use tokio_util::sync::CancellationToken;
use tracing::Instrument;
use uuid::Uuid;

const RETRY_BASE_SECS: u64 = 30;
const RETRY_MAX_SECS: u64 = 3600;
const MAX_PURGE_ATTEMPTS: i32 = 20;
/// 一次只领取一行：purge 会跨 KenShell / KenAuth 多步执行，逐行领取能让关停
/// 在行与行之间生效，而不是把已领取的其余行晾到租约过期。
const CLAIM_BATCH: i64 = 1;
const CLAIM_LEASE_SECS: f64 = 600.0;

const CLAIM_SPACES: &str = "
WITH pending AS (
    SELECT candidate.id FROM space_deletions AS candidate
    WHERE candidate.status = 'pending'
      AND candidate.available_at <= CURRENT_TIMESTAMP
      AND (candidate.leased_until IS NULL OR candidate.leased_until <= CURRENT_TIMESTAMP)
    ORDER BY candidate.available_at
    LIMIT $1
    FOR UPDATE OF candidate SKIP LOCKED
)
UPDATE space_deletions AS d
SET lease_id = $2,
    leased_until = CURRENT_TIMESTAMP + make_interval(secs => $3),
    attempts = LEAST(d.attempts, 2147483646) + 1
FROM pending
WHERE d.id = pending.id
RETURNING d.id
";

const CLAIM_ACCOUNTS: &str = "
WITH pending AS (
    SELECT candidate.id FROM account_deletions AS candidate
    WHERE candidate.status = 'pending'
      AND candidate.available_at <= CURRENT_TIMESTAMP
      AND (candidate.leased_until IS NULL OR candidate.leased_until <= CURRENT_TIMESTAMP)
    ORDER BY candidate.available_at
    LIMIT $1
    FOR UPDATE OF candidate SKIP LOCKED
)
UPDATE account_deletions AS d
SET lease_id = $2,
    leased_until = CURRENT_TIMESTAMP + make_interval(secs => $3),
    attempts = LEAST(d.attempts, 2147483646) + 1
FROM pending
WHERE d.id = pending.id
RETURNING d.id
";

#[derive(Debug, thiserror::Error)]
pub enum DeletionError {
    #[error("database: {0}")]
    Database(#[from] sea_orm::DbErr),
    #[error("{0}")]
    Conflict(String),
    #[error("{0}")]
    NotFound(String),
    #[error("kenoauth: {0}")]
    KenAuth(String),
    #[error("kenshell: {0}")]
    KenShell(String),
    #[error("config: {0}")]
    Config(String),
}

impl DeletionError {
    pub fn is_retryable(&self) -> bool {
        !matches!(
            self,
            Self::Conflict(_) | Self::NotFound(_) | Self::Config(_)
        )
    }
}

fn ken_auth(error: KenAuthError) -> DeletionError {
    match error {
        KenAuthError::NotFound => {
            DeletionError::NotFound("account or organization not found".into())
        }
        KenAuthError::Rejected(message) => DeletionError::Conflict(message),
        other => DeletionError::KenAuth(other.to_string()),
    }
}

fn ken_shell(error: proto::KenShellError) -> DeletionError {
    DeletionError::KenShell(error.to_string())
}

fn now() -> OffsetDateTime {
    OffsetDateTime::now_utc()
}

fn grace_until(days: u32) -> OffsetDateTime {
    now() + time::Duration::days(i64::from(days.max(1)))
}

fn retry_at(attempts: i32) -> OffsetDateTime {
    let exponent = attempts.clamp(1, 8) as u32;
    let delay = (RETRY_BASE_SECS.saturating_mul(1u64 << exponent)).min(RETRY_MAX_SECS);
    now() + time::Duration::seconds(delay as i64)
}

#[derive(Debug, Clone, Serialize)]
pub struct AccountDeletionStatus {
    pub status: String,
    pub requested_at: OffsetDateTime,
    pub purge_after: OffsetDateTime,
    pub last_error: Option<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct SpaceDeletionStatus {
    pub status: String,
    pub requested_at: OffsetDateTime,
    pub purge_after: OffsetDateTime,
    pub last_error: Option<String>,
}

pub async fn request_account_deletion(
    state: &AppState,
    accounts: &auth::Accounts,
    user_id: i64,
    reason: &str,
) -> Result<AccountDeletionStatus, DeletionError> {
    let db = state.database.write();
    if let Some(existing) = account_deletion::Entity::find()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .filter(account_deletion::Column::Status.eq("pending"))
        .one(db)
        .await?
    {
        return Ok(status_of_account(&existing));
    }

    let owned: Vec<Uuid> = spaces::Entity::find()
        .filter(spaces::Column::OwnerUserId.eq(user_id))
        .filter(spaces::Column::DeletedAt.is_null())
        .select_only()
        .column(spaces::Column::Id)
        .into_tuple()
        .all(db)
        .await?;
    for space_id in owned {
        let pending = space_deletion::Entity::find()
            .filter(space_deletion::Column::SpaceId.eq(space_id))
            .filter(space_deletion::Column::Status.eq("pending"))
            .one(db)
            .await?;
        if pending.is_none() {
            return Err(DeletionError::Conflict(
                "delete or transfer owned spaces before deleting the account".into(),
            ));
        }
    }
    let client = accounts;
    let status = client.get_user_status(user_id).await.map_err(ken_auth)?;
    if status.deletion_requested_at.is_none() && status.deleted_at.is_none() {
        client
            .delete_account(user_id, reason)
            .await
            .map_err(ken_auth)?;
    }

    let now = now();
    let purge_after = grace_until(state.config.user.deletion_grace_days);
    let row = match account_deletion::Entity::find()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .one(db)
        .await?
    {
        Some(existing) => {
            let mut active: account_deletion::ActiveModel = existing.into();
            active.status = Set("pending".into());
            active.requested_at = Set(now);
            active.purge_after = Set(purge_after);
            active.available_at = Set(purge_after);
            active.last_error = Set(None);
            active.attempts = Set(0);
            active.lease_id = Set(None);
            active.leased_until = Set(None);
            active.updated_at = Set(now);
            active.update(db).await?
        }
        None => {
            account_deletion::ActiveModel {
                id: Set(Uuid::now_v7()),
                user_id: Set(user_id),
                requested_at: Set(now),
                purge_after: Set(purge_after),
                status: Set("pending".into()),
                last_error: Set(None),
                attempts: Set(0),
                available_at: Set(purge_after),
                lease_id: Set(None),
                leased_until: Set(None),
                created_at: Set(now),
                updated_at: Set(now),
            }
            .insert(db)
            .await?
        }
    };
    Ok(status_of_account(&row))
}

pub async fn cancel_account_deletion(
    state: &AppState,
    accounts: &auth::Accounts,
    user_id: i64,
) -> Result<(), DeletionError> {
    let db = state.database.write();
    let existing = account_deletion::Entity::find()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .one(db)
        .await?;
    if let Some(row) = &existing
        && row.lease_id.is_some()
        && row.leased_until.is_some_and(|until| until > now())
    {
        return Err(DeletionError::Conflict(
            "account deletion is currently being purged".into(),
        ));
    }
    let deleted = account_deletion::Entity::delete_many()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .filter(
            Condition::any()
                .add(account_deletion::Column::LeaseId.is_null())
                .add(account_deletion::Column::LeasedUntil.lte(now())),
        )
        .exec(db)
        .await?;
    if deleted.rows_affected == 0 && existing.is_some() {
        return Err(DeletionError::Conflict(
            "account deletion is currently being purged".into(),
        ));
    }
    // 先删本地行再撤销远端：这样 purge worker 不会再领到它；远端失败时必须把行
    // 放回去，否则本地记录消失、KenAuth 侧仍会继续删除账号，两侧状态分叉。
    if let Err(error) = accounts
        .cancel_account_deletion(user_id)
        .await
        .map_err(ken_auth)
    {
        if let Some(row) = existing {
            let restore = account_deletion::ActiveModel::from(row);
            if let Err(restore_error) = restore.insert(db).await {
                tracing::error!(
                    %restore_error,
                    user_id,
                    "failed to restore account deletion row after remote cancel failed"
                );
            }
        }
        return Err(error);
    }
    Ok(())
}

pub async fn account_deletion_status(
    state: &AppState,
    user_id: i64,
) -> Result<Option<AccountDeletionStatus>, DeletionError> {
    Ok(account_deletion::Entity::find()
        .filter(account_deletion::Column::UserId.eq(user_id))
        .one(state.database.read())
        .await?
        .as_ref()
        .map(status_of_account))
}

fn status_of_account(row: &account_deletion::Model) -> AccountDeletionStatus {
    AccountDeletionStatus {
        status: row.status.clone(),
        requested_at: row.requested_at,
        purge_after: row.purge_after,
        last_error: row.last_error.clone(),
    }
}

pub async fn run_worker(state: AppState, accounts: auth::Accounts, cancel: CancellationToken) {
    tracing::info!("deletion worker started");
    let mut interval = tokio::time::interval(Duration::from_secs(30));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        tokio::select! {
            _ = cancel.cancelled() => break,
            _ = interval.tick() => {}
        }
        let request_id = Uuid::now_v7().to_string();
        let span = tracing::info_span!("deletion.worker", request_id = %request_id);
        let result = api::middleware::with_request_id(
            Some(&request_id),
            purge_due(&state, &accounts, &cancel).instrument(span),
        )
        .await;
        if let Err(error) = result {
            tracing::warn!(%error, "deletion worker pass failed");
        }
    }
    tracing::info!("deletion worker stopped");
}

async fn purge_due(
    state: &AppState,
    accounts: &auth::Accounts,
    cancel: &CancellationToken,
) -> Result<(), DeletionError> {
    while !cancel.is_cancelled() {
        let lease_id = Uuid::now_v7();
        let batch = claim_spaces(state, lease_id).await?;
        if batch.is_empty() {
            break;
        }
        for row in batch {
            if let Err(error) = purge_space(state, &row).await {
                record_space_failure(state, &row, &error).await?;
            }
        }
    }

    while !cancel.is_cancelled() {
        let lease_id = Uuid::now_v7();
        let batch = claim_accounts(state, lease_id).await?;
        if batch.is_empty() {
            break;
        }
        for row in batch {
            if let Err(error) = purge_account(state, accounts, &row).await {
                record_account_failure(state, &row, &error).await?;
            }
        }
    }
    Ok(())
}

async fn claim_spaces(
    state: &AppState,
    lease_id: Uuid,
) -> Result<Vec<space_deletion::Model>, DeletionError> {
    let db = state.database.write();
    let claimed = db
        .query_all_raw(Statement::from_sql_and_values(
            DbBackend::Postgres,
            CLAIM_SPACES,
            [CLAIM_BATCH.into(), lease_id.into(), CLAIM_LEASE_SECS.into()],
        ))
        .await?;
    let mut ids = Vec::with_capacity(claimed.len());
    for row in claimed {
        let id: Uuid = row.try_get("", "id")?;
        ids.push(id);
    }
    if ids.is_empty() {
        return Ok(Vec::new());
    }
    Ok(space_deletion::Entity::find()
        .filter(space_deletion::Column::Id.is_in(ids))
        .all(db)
        .await?)
}

async fn claim_accounts(
    state: &AppState,
    lease_id: Uuid,
) -> Result<Vec<account_deletion::Model>, DeletionError> {
    let db = state.database.write();
    let claimed = db
        .query_all_raw(Statement::from_sql_and_values(
            DbBackend::Postgres,
            CLAIM_ACCOUNTS,
            [CLAIM_BATCH.into(), lease_id.into(), CLAIM_LEASE_SECS.into()],
        ))
        .await?;
    let mut ids = Vec::with_capacity(claimed.len());
    for row in claimed {
        let id: Uuid = row.try_get("", "id")?;
        ids.push(id);
    }
    if ids.is_empty() {
        return Ok(Vec::new());
    }
    Ok(account_deletion::Entity::find()
        .filter(account_deletion::Column::Id.is_in(ids))
        .all(db)
        .await?)
}

async fn purge_space(state: &AppState, row: &space_deletion::Model) -> Result<(), DeletionError> {
    // Never delete a space whose ownership changed after the request.
    let space = {
        let db = state.database.write();
        spaces::Entity::find_by_id(row.space_id).one(db).await?
    };
    let Some(space) = space else {
        finish_space_deletion(state, row).await?;
        return Ok(());
    };
    if space.owner_user_id != row.requested_by {
        let db = state.database.write();
        let mut active: spaces::ActiveModel = space.into();
        active.status = Set(SpaceStatus::Active);
        active.archived_at = Set(None);
        active.update(db).await?;
        finish_space_deletion(state, row).await?;
        return Ok(());
    }

    // 1. KenShell is the owner of repository/placement/LFS bytes.
    let shell = {
        let cfg = &state.config.git;
        KenShellClient::connect_with_key(
            &cfg.kenshell_endpoint,
            &cfg.kenshell_rpc_key,
            Duration::from_secs(cfg.kenshell_timeout_secs.max(1)),
        )
        .map(|client| client.with_request_id(api::middleware::request_id))
        .map_err(ken_shell)
    }?;
    let repos = shell
        .list_repositories_for_org(&row.space_id.to_string())
        .await
        .map_err(ken_shell)?;
    for repo in repos {
        match shell.delete_repository(&repo.id).await {
            Ok(()) => {}
            Err(error) if error.is_not_found() => {}
            Err(error) => return Err(ken_shell(error)),
        }
    }

    // 2. Local KenApi resources become unreachable and space-scoped metadata is removed.
    archive_local_space(state, row.space_id).await?;

    // 3. The space row and its members/roles/invites cascade away.
    spaces::Entity::delete_by_id(row.space_id)
        .exec(state.database.write())
        .await?;

    if finish_space_deletion(state, row).await? {
        tracing::info!(space_id = %row.space_id, "space purged");
    }
    Ok(())
}

async fn finish_space_deletion(
    state: &AppState,
    row: &space_deletion::Model,
) -> Result<bool, DeletionError> {
    let deleted = space_deletion::Entity::delete_many()
        .filter(space_deletion::Column::Id.eq(row.id))
        .filter(space_deletion::Column::LeaseId.eq(row.lease_id))
        .exec(state.database.write())
        .await?;
    if deleted.rows_affected == 0 {
        tracing::warn!(
            space_id = %row.space_id,
            "space deletion lease was reclaimed; skipping the completion write"
        );
        return Ok(false);
    }
    Ok(true)
}

async fn archive_local_space(state: &AppState, space_id: Uuid) -> Result<(), DeletionError> {
    let db = state.database.write();

    rooms::Entity::update_many()
        .col_expr(
            rooms::Column::IsActive,
            sea_orm::sea_query::Expr::value(false),
        )
        .filter(rooms::Column::SpaceId.eq(Some(space_id)))
        .exec(db)
        .await?;

    repos::Entity::update_many()
        .col_expr(
            repos::Column::IsArchived,
            sea_orm::sea_query::Expr::value(true),
        )
        .filter(repos::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;

    let group_ids: Vec<Uuid> = space_groups::Entity::find()
        .filter(space_groups::Column::SpaceId.eq(space_id))
        .select_only()
        .column(space_groups::Column::Id)
        .into_tuple()
        .all(db)
        .await?;
    if !group_ids.is_empty() {
        space_group_permissions::Entity::delete_many()
            .filter(space_group_permissions::Column::GroupId.is_in(group_ids))
            .exec(db)
            .await?;
    }
    space_member_groups::Entity::delete_many()
        .filter(space_member_groups::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    space_groups::Entity::delete_many()
        .filter(space_groups::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    space_invites::Entity::delete_many()
        .filter(space_invites::Column::OrganizationId.eq(space_id))
        .exec(db)
        .await?;
    space_join_requests::Entity::delete_many()
        .filter(space_join_requests::Column::OrganizationId.eq(space_id))
        .exec(db)
        .await?;
    space_join_rules::Entity::delete_many()
        .filter(space_join_rules::Column::OrganizationId.eq(space_id))
        .exec(db)
        .await?;
    space_warn::Entity::delete_many()
        .filter(space_warn::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    space_extras::Entity::delete_by_id(space_id)
        .exec(db)
        .await?;
    space_member_extras::Entity::delete_many()
        .filter(space_member_extras::Column::SpaceId.eq(space_id))
        .exec(db)
        .await?;
    Ok(())
}

async fn purge_account(
    state: &AppState,
    accounts: &auth::Accounts,
    row: &account_deletion::Model,
) -> Result<(), DeletionError> {
    let db = state.database.write();
    let owned = spaces::Entity::find()
        .filter(spaces::Column::OwnerUserId.eq(row.user_id))
        .filter(spaces::Column::DeletedAt.is_null())
        .count(db)
        .await?;
    if owned > 0 {
        return Err(DeletionError::Conflict(
            "owned spaces are still being purged".into(),
        ));
    }

    let now = now();

    space_members::Entity::delete_many()
        .filter(space_members::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_preferences::Entity::delete_many()
        .filter(user_preferences::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_profiles::Entity::delete_many()
        .filter(user_profiles::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_notifications::Entity::delete_many()
        .filter(user_notifications::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_push_subscriptions::Entity::delete_many()
        .filter(user_push_subscriptions::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_favorite_memes::Entity::delete_many()
        .filter(user_favorite_memes::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_recent_mentions::Entity::delete_many()
        .filter(user_recent_mentions::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_saved_messages::Entity::delete_many()
        .filter(user_saved_messages::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_external_identity::Entity::delete_many()
        .filter(user_external_identity::Column::UserId.eq(Some(row.user_id)))
        .exec(db)
        .await?;
    user_agreement_acceptances::Entity::delete_many()
        .filter(user_agreement_acceptances::Column::UserId.eq(row.user_id))
        .exec(db)
        .await?;
    user_blocks::Entity::delete_many()
        .filter(
            sea_orm::Condition::any()
                .add(user_blocks::Column::BlockerId.eq(row.user_id))
                .add(user_blocks::Column::BlockedId.eq(row.user_id)),
        )
        .exec(db)
        .await?;
    user_followers::Entity::delete_many()
        .filter(
            sea_orm::Condition::any()
                .add(user_followers::Column::FollowerId.eq(row.user_id))
                .add(user_followers::Column::FollowedId.eq(row.user_id)),
        )
        .exec(db)
        .await?;

    if let Some(user) = user::Entity::find_by_id(row.user_id).one(db).await? {
        let mut active: user::ActiveModel = user.into();
        active.is_restricted = Set(true);
        active.local_avatar_url = Set(String::new());
        active.visibility = Set(models::users::user::ProfileVisibility::Private);
        active.last_login_at = Set(None);
        active.last_login_ip_hash = Set(None);
        active.updated_at = Set(now);
        active.update(db).await?;
    }

    let client = accounts;
    match client.purge_account(row.user_id).await {
        Ok(()) | Err(KenAuthError::NotFound) => {}
        Err(error) => return Err(ken_auth(error)),
    }
    let deleted = account_deletion::Entity::delete_many()
        .filter(account_deletion::Column::Id.eq(row.id))
        .filter(account_deletion::Column::LeaseId.eq(row.lease_id))
        .exec(db)
        .await?;
    if deleted.rows_affected == 0 {
        tracing::warn!(
            user_id = %row.user_id,
            "account deletion lease was reclaimed; skipping the completion write"
        );
        return Ok(());
    }
    tracing::info!(user_id = %row.user_id, "account purged");
    Ok(())
}

async fn record_space_failure(
    state: &AppState,
    row: &space_deletion::Model,
    error: &DeletionError,
) -> Result<(), DeletionError> {
    let attempts = row.attempts.max(1);
    let retryable = error.is_retryable() && attempts < MAX_PURGE_ATTEMPTS;
    let updated = space_deletion::Entity::update_many()
        .col_expr(space_deletion::Column::Attempts, Expr::value(attempts))
        .col_expr(
            space_deletion::Column::AvailableAt,
            Expr::value(retry_at(attempts)),
        )
        .col_expr(
            space_deletion::Column::LastError,
            Expr::value(Some(error.to_string())),
        )
        .col_expr(
            space_deletion::Column::Status,
            Expr::value(if retryable { "pending" } else { "failed" }),
        )
        .col_expr(
            space_deletion::Column::LeaseId,
            Expr::value(Option::<Uuid>::None),
        )
        .col_expr(
            space_deletion::Column::LeasedUntil,
            Expr::value(Option::<OffsetDateTime>::None),
        )
        .col_expr(space_deletion::Column::UpdatedAt, Expr::value(now()))
        .filter(space_deletion::Column::Id.eq(row.id))
        .filter(space_deletion::Column::LeaseId.eq(row.lease_id))
        .exec(state.database.write())
        .await?;
    if updated.rows_affected == 0 {
        tracing::warn!(
            space_id = %row.space_id,
            %error,
            "space purge failed after its lease was reclaimed"
        );
        return Ok(());
    }
    if retryable {
        tracing::warn!(space_id = %row.space_id, %error, attempts, "space purge failed; will retry");
    } else {
        tracing::warn!(space_id = %row.space_id, %error, attempts, "space purge failed permanently");
    }
    Ok(())
}

async fn record_account_failure(
    state: &AppState,
    row: &account_deletion::Model,
    error: &DeletionError,
) -> Result<(), DeletionError> {
    let attempts = row.attempts.max(1);
    let retryable = error.is_retryable() && attempts < MAX_PURGE_ATTEMPTS;
    let updated = account_deletion::Entity::update_many()
        .col_expr(account_deletion::Column::Attempts, Expr::value(attempts))
        .col_expr(
            account_deletion::Column::AvailableAt,
            Expr::value(retry_at(attempts)),
        )
        .col_expr(
            account_deletion::Column::LastError,
            Expr::value(Some(error.to_string())),
        )
        .col_expr(
            account_deletion::Column::Status,
            Expr::value(if retryable { "pending" } else { "failed" }),
        )
        .col_expr(
            account_deletion::Column::LeaseId,
            Expr::value(Option::<Uuid>::None),
        )
        .col_expr(
            account_deletion::Column::LeasedUntil,
            Expr::value(Option::<OffsetDateTime>::None),
        )
        .col_expr(account_deletion::Column::UpdatedAt, Expr::value(now()))
        .filter(account_deletion::Column::Id.eq(row.id))
        .filter(account_deletion::Column::LeaseId.eq(row.lease_id))
        .exec(state.database.write())
        .await?;
    if updated.rows_affected == 0 {
        tracing::warn!(
            user_id = %row.user_id,
            %error,
            "account purge failed after its lease was reclaimed"
        );
        return Ok(());
    }
    if retryable {
        tracing::warn!(user_id = %row.user_id, %error, attempts, "account purge failed; will retry");
    } else {
        tracing::warn!(user_id = %row.user_id, %error, attempts, "account purge failed permanently");
    }
    Ok(())
}

impl From<DeletionError> for auth::AuthBusinessError {
    fn from(error: DeletionError) -> Self {
        match error {
            DeletionError::Database(error) => Self::from(error),
            DeletionError::Conflict(message) => Self::Conflict(message),
            DeletionError::NotFound(_) => Self::NotFound,
            other => Self::Internal(other.to_string()),
        }
    }
}