use axum::{
    Json,
    extract::State,
    http::{HeaderMap, StatusCode},
    response::{IntoResponse, Response},
};
use bcrypt::verify;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tracing::{error, info, warn};
use tranquil_db_traits::{ChannelVerificationStatus, SessionId, TokenFamilyId};
use tranquil_pds::api::error::{ApiError, DbResultExt};
use tranquil_pds::api::{EmptyResponse, PreferredLocaleOutput, SuccessResponse};
use tranquil_pds::auth::{
    Active, Auth, NormalizedLoginIdentifier, Permissive, require_legacy_session_mfa,
    require_reauth_window,
};
use tranquil_pds::rate_limit::{
    LoginLimit, RateLimited, RefreshSessionLimit, TotpVerifyLimit,
    check_user_rate_limit_with_message,
};
use tranquil_pds::state::AppState;
use tranquil_pds::types::{AccountState, AtIdentifier, Did, Handle, PlainPassword};
use tranquil_types::TokenId;

pub fn verification_blocks_login(channel_verification: &ChannelVerificationStatus) -> bool {
    !tranquil_config::get()
        .server
        .disable_account_verification_gate
        && !channel_verification.has_any_verified()
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateSessionInput {
    pub identifier: String,
    pub password: PlainPassword,
    #[serde(default)]
    pub allow_takendown: bool,
    pub auth_factor_token: Option<String>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CreateSessionOutput {
    pub access_jwt: String,
    pub refresh_jwt: String,
    pub handle: Handle,
    pub did: Did,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub did_doc: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub email: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub email_confirmed: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub email_auth_factor: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub active: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub status: Option<String>,
}

pub async fn create_session(
    State(state): State<AppState>,
    rate_limit: RateLimited<LoginLimit>,
    Json(input): Json<CreateSessionInput>,
) -> Result<Response, ApiError> {
    let client_ip = rate_limit.client_ip();
    info!(
        "create_session called with identifier: {}",
        input.identifier
    );
    let hostname_for_handles = tranquil_config::get().server.hostname_without_port();
    let normalized_identifier =
        NormalizedLoginIdentifier::normalize(&input.identifier, hostname_for_handles);
    info!(
        "Normalized identifier: {} -> {}",
        input.identifier, normalized_identifier
    );
    let Ok(login_identifier) = AtIdentifier::new(normalized_identifier.as_str()) else {
        let _ = verify(
            &input.password,
            "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.VTtYw1ZzQKZqmK",
        );
        warn!("Login identifier is not a valid handle or DID");
        return Err(ApiError::AuthenticationFailed(Some(
            "Invalid identifier or password".into(),
        )));
    };
    let row = match state
        .repos
        .user
        .get_login_full_by_identifier(&login_identifier)
        .await
    {
        Ok(Some(row)) => row,
        Ok(None) => {
            let _ = verify(
                &input.password,
                "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.VTtYw1ZzQKZqmK",
            );
            warn!("User not found for login attempt");
            return Err(ApiError::AuthenticationFailed(Some(
                "Invalid identifier or password".into(),
            )));
        }
        Err(e) => {
            error!("Database error fetching user: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    let key_bytes = match tranquil_pds::config::decrypt_key(&row.key_bytes, row.encryption_version)
    {
        Ok(k) => k,
        Err(e) => {
            error!("Failed to decrypt user key: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    let credential = crate::common::verify_credential(
        state.repos.session.as_ref(),
        row.id,
        &input.password,
        row.password_hash.as_ref(),
    )
    .await;
    let (app_password_name, app_password_scopes, app_password_controller) = match credential {
        Some(crate::common::CredentialMatch::MainPassword) => (None, None, None),
        Some(crate::common::CredentialMatch::AppPassword {
            name,
            scopes,
            controller_did,
        }) => (Some(name), scopes, controller_did),
        None => {
            warn!("Password verification failed for login attempt");
            return Err(ApiError::AuthenticationFailed(Some(
                "Invalid identifier or password".into(),
            )));
        }
    };
    let account_state = AccountState::from_db_fields(
        row.deactivated_at,
        row.takedown_ref.clone(),
        row.migrated_to_pds.clone(),
        None,
    );
    if account_state.is_takendown() && !input.allow_takendown {
        warn!("Login attempt for takendown account: {}", row.did);
        return Err(ApiError::AccountTakedown);
    }
    let is_delegated = state
        .repos
        .delegation
        .is_delegated_account(&row.did)
        .await
        .unwrap_or(false);
    if verification_blocks_login(&row.channel_verification) && !is_delegated {
        warn!("Login attempt for unverified account: {}", row.did);
        let resend_info = auto_resend_verification(&state, &row.did).await;
        let handle = resend_info
            .as_ref()
            .map(|r| r.handle.to_string())
            .unwrap_or_else(|| row.handle.to_string());
        let channel = resend_info
            .as_ref()
            .map(|r| r.channel.as_str())
            .unwrap_or(row.preferred_comms_channel.as_str());
        return Ok((
            StatusCode::FORBIDDEN,
            Json(json!({
                "error": "account_not_verified",
                "message": "Please verify your account before logging in",
                "did": row.did,
                "handle": handle,
                "channel": channel
            })),
        )
            .into_response());
    }
    let has_totp = row.totp_enabled;
    let email_2fa_enabled = row.email_2fa_enabled;
    let is_legacy_login = has_totp || email_2fa_enabled;
    let used_totp_factor = tranquil_pds::auth::legacy_2fa::used_totp_factor(
        has_totp,
        input.auth_factor_token.as_deref(),
    );
    let twofa_ctx = tranquil_pds::auth::legacy_2fa::Legacy2faContext {
        is_app_password: app_password_name.is_some(),
        email_2fa_enabled,
        has_totp,
        allow_legacy_login: row.allow_legacy_login,
    };
    if let Some(token) = input.auth_factor_token.as_deref()
        && has_totp
        && tranquil_pds::auth::legacy_2fa::looks_like_totp_token(token)
    {
        check_user_rate_limit_with_message::<TotpVerifyLimit>(
            &state,
            row.did.as_str(),
            "Too many verification attempts. Please try again in a few minutes.",
        )
        .await?;
    }
    match tranquil_pds::auth::legacy_2fa::process_legacy_2fa(
        state.cache.as_ref(),
        &row.did,
        &twofa_ctx,
        input.auth_factor_token.as_deref(),
        async |t: &str| {
            crate::server::totp::verify_totp_or_backup_for_user(&state, &row.did, t).await
        },
    )
    .await
    {
        Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::NotRequired) => {}
        Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::Blocked) => {
            warn!("Legacy login blocked for TOTP-enabled account: {}", row.did);
            return Err(ApiError::LegacyLoginBlocked);
        }
        Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::ChallengeSent(code)) => {
            let hostname = &tranquil_config::get().server.hostname;
            match tranquil_pds::comms::comms_repo::enqueue_notice(
                state.repos.user.as_ref(),
                state.repos.infra.as_ref(),
                row.id,
                tranquil_pds::comms::Notice::TwoFactorCode {
                    code: code.as_str(),
                },
                hostname,
            )
            .await
            {
                Ok(Some(_)) => {}
                Ok(None) => {
                    tranquil_pds::auth::legacy_2fa::clear_challenge(state.cache.as_ref(), &row.did)
                        .await;
                    return Err(ApiError::InvalidRequest(
                        "We couldn't deliver the verification code to your notification channels. Please contact the PDS owner."
                            .into(),
                    ));
                }
                Err(e) => {
                    error!("Failed to send 2FA code: {:?}", e);
                    tranquil_pds::auth::legacy_2fa::clear_challenge(state.cache.as_ref(), &row.did)
                        .await;
                    return Err(ApiError::InternalError(Some(
                        "Failed to send verification code. Please try again.".into(),
                    )));
                }
            }
            return Err(ApiError::AuthFactorTokenRequired);
        }
        Ok(tranquil_pds::auth::legacy_2fa::Legacy2faOutcome::Verified) => {}
        Err(tranquil_pds::auth::legacy_2fa::Legacy2faFlowError::Challenge(e)) => {
            use tranquil_pds::auth::legacy_2fa::ChallengeError;
            return match e {
                ChallengeError::CacheUnavailable => {
                    error!("Cache unavailable for 2FA, blocking legacy login");
                    Err(ApiError::ServiceUnavailable(Some(
                        "2FA service temporarily unavailable. Please try again later or use an OAuth client.".into(),
                    )))
                }
                ChallengeError::RateLimited => Err(ApiError::RateLimitExceeded(Some(
                    "Please wait before requesting a new verification code.".into(),
                ))),
                ChallengeError::CacheError => {
                    error!("Cache error during 2FA challenge creation");
                    Err(ApiError::InternalError(None))
                }
            };
        }
        Err(tranquil_pds::auth::legacy_2fa::Legacy2faFlowError::Validation(e)) => {
            use tranquil_pds::auth::legacy_2fa::ValidationError;
            warn!("Invalid 2FA code for {}: {:?}", row.did, e);
            let msg = match e {
                ValidationError::TooManyAttempts => "Too many attempts. Please request a new code.",
                ValidationError::ChallengeExpired => "Code has expired. Please request a new code.",
                ValidationError::CacheUnavailable => {
                    "2FA service temporarily unavailable. Please try again later."
                }
                ValidationError::ChallengeNotFound
                | ValidationError::InvalidCode
                | ValidationError::CacheError => "Invalid verification code",
            };
            return Err(ApiError::InvalidCode(Some(msg.into())));
        }
    }
    let access_meta = match tranquil_pds::auth::create_access_token_with_delegation(
        &row.did,
        &key_bytes,
        app_password_scopes.as_deref(),
        app_password_controller.as_ref(),
        None,
    ) {
        Ok(m) => m,
        Err(e) => {
            error!("Failed to create access token: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    let refresh_meta =
        match tranquil_pds::auth::create_refresh_token_with_metadata(&row.did, &key_bytes) {
            Ok(m) => m,
            Err(e) => {
                error!("Failed to create refresh token: {:?}", e);
                return Err(ApiError::InternalError(None));
            }
        };
    let did_for_doc = row.did.clone();
    let did_resolver = state.did_resolver.clone();
    let session_data = tranquil_db_traits::SessionTokenCreate {
        did: row.did.clone(),
        access_jti: access_meta.jti.clone(),
        refresh_jti: refresh_meta.jti.clone(),
        access_expires_at: access_meta.expires_at,
        refresh_expires_at: refresh_meta.expires_at,
        login_type: tranquil_db_traits::LoginType::from_legacy_flag(is_legacy_login),
        mfa_verified: false,
        scope: app_password_scopes.clone(),
        controller_did: app_password_controller.clone(),
        app_password_name: app_password_name.clone(),
    };
    let (insert_result, did_doc) = tokio::join!(
        state.repos.session.create_session(&session_data),
        did_resolver.fetch_did_document(&did_for_doc),
    );
    if let Err(e) = insert_result {
        error!("Failed to insert session: {:?}", e);
        return Err(ApiError::InternalError(None));
    }
    if is_legacy_login && !used_totp_factor {
        let alerts_enabled = match state.repos.infra.get_account_preferences(row.id).await {
            Ok(prefs) => prefs
                .iter()
                .find(|(name, _)| name == "legacy_login_alerts")
                .and_then(|(_, value)| value.as_bool())
                .unwrap_or(true),
            Err(e) => {
                warn!("Failed to fetch legacy login alert preference: {:?}", e);
                true
            }
        };

        if alerts_enabled {
            warn!(
                did = %row.did,
                ip = %client_ip,
                "Legacy login on TOTP-enabled account - sending notification"
            );
            let hostname = &tranquil_config::get().server.hostname;
            if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_notice(
                state.repos.user.as_ref(),
                state.repos.infra.as_ref(),
                row.id,
                tranquil_pds::comms::Notice::LegacyLoginAlert {
                    channel: row.preferred_comms_channel,
                    ip: client_ip,
                },
                hostname,
            )
            .await
            {
                error!("Failed to queue legacy login notification: {:?}", e);
            }
        }
    }
    let handle = row.handle.clone();
    let is_active = account_state.is_active();
    let status = account_state.status_for_session().map(String::from);
    let email_auth_factor_out = if email_2fa_enabled || has_totp {
        Some(true)
    } else {
        None
    };
    Ok((
        StatusCode::OK,
        Json(CreateSessionOutput {
            access_jwt: access_meta.token,
            refresh_jwt: refresh_meta.token,
            handle,
            did: row.did,
            did_doc: did_doc.ok(),
            email: row.email,
            email_confirmed: Some(row.channel_verification.email),
            email_auth_factor: email_auth_factor_out,
            active: Some(is_active),
            status,
        }),
    )
        .into_response())
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct GetSessionOutput {
    pub handle: Handle,
    pub did: Did,
    pub active: bool,
    pub preferred_channel: String,
    pub preferred_channel_verified: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub preferred_locale: Option<String>,
    pub is_admin: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub email: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub email_confirmed: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub email_auth_factor: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub status: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub migrated_to_pds: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub migrated_at: Option<chrono::DateTime<chrono::Utc>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub did_doc: Option<serde_json::Value>,
}

pub async fn get_session(
    State(state): State<AppState>,
    auth: Auth<Permissive>,
) -> Result<Json<GetSessionOutput>, ApiError> {
    let permissions = auth.permissions();
    let can_read_email = permissions.allows_email_read();

    let did_for_doc = auth.did.clone();
    let did_resolver = state.did_resolver.clone();
    let (db_result, did_doc) = tokio::join!(
        state.repos.user.get_session_info_by_did(&auth.did),
        did_resolver.fetch_did_document(&did_for_doc)
    );
    match db_result {
        Ok(Some(row)) => {
            let preferred_channel_verified = row
                .channel_verification
                .is_verified(row.preferred_comms_channel);
            let handle = row.handle.clone();
            let account_state = AccountState::from_db_fields(
                row.deactivated_at,
                row.takedown_ref.clone(),
                row.migrated_to_pds.clone(),
                row.migrated_at,
            );
            let email = match can_read_email {
                true => row.email.clone(),
                false => None,
            };
            let email_confirmed = match can_read_email {
                true => Some(row.channel_verification.email),
                false => None,
            };
            let email_auth_factor = match row.email_2fa_enabled || row.totp_enabled {
                true => Some(true),
                false => None,
            };
            let (migrated_to_pds, migrated_at) = match &account_state {
                AccountState::Migrated { to_pds, at } => (Some(to_pds.clone()), Some(*at)),
                _ => (None, None),
            };
            Ok(Json(GetSessionOutput {
                handle,
                did: auth.did.clone(),
                active: account_state.is_active(),
                preferred_channel: row.preferred_comms_channel.as_str().to_string(),
                preferred_channel_verified,
                preferred_locale: row.preferred_locale,
                is_admin: row.is_admin,
                email,
                email_confirmed,
                email_auth_factor,
                status: account_state.status_for_session().map(String::from),
                migrated_to_pds,
                migrated_at,
                did_doc: did_doc.ok(),
            }))
        }
        Ok(None) => Err(ApiError::AuthenticationFailed(None)),
        Err(e) => {
            error!("Database error in get_session: {:?}", e);
            Err(ApiError::InternalError(None))
        }
    }
}

pub async fn delete_session(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    auth: Auth<Active>,
) -> Result<Json<EmptyResponse>, ApiError> {
    let jti = tranquil_pds::auth::extract_jti_from_headers(&headers)
        .ok_or(ApiError::AuthenticationRequired)?;
    match state
        .repos
        .session
        .delete_session_by_access_jti(&jti, &auth.did)
        .await
    {
        Ok(rows) if rows > 0 => {
            let session_cache_key = tranquil_pds::cache_keys::session_key(&auth.did, &jti);
            let _ = state.cache.delete(&session_cache_key).await;
            Ok(Json(EmptyResponse {}))
        }
        Ok(_) => Err(ApiError::AuthenticationFailed(None)),
        Err(_) => Err(ApiError::AuthenticationFailed(None)),
    }
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RefreshSessionOutput {
    pub access_jwt: String,
    pub refresh_jwt: String,
    pub handle: Handle,
    pub did: Did,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub email: Option<String>,
    pub email_confirmed: bool,
    pub preferred_channel: String,
    pub preferred_channel_verified: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub preferred_locale: Option<String>,
    pub is_admin: bool,
    pub active: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub did_doc: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub status: Option<String>,
}

pub async fn refresh_session(
    State(state): State<AppState>,
    _rate_limit: RateLimited<RefreshSessionLimit>,
    headers: axum::http::HeaderMap,
) -> Result<Json<RefreshSessionOutput>, ApiError> {
    let extracted = match tranquil_pds::auth::extract_auth_token_from_header(
        tranquil_pds::util::get_header_str(&headers, http::header::AUTHORIZATION),
    ) {
        Some(t) => t,
        None => return Err(ApiError::AuthenticationRequired),
    };
    let refresh_token = extracted.token;
    let refresh_jti = match tranquil_pds::auth::get_jti_from_token(&refresh_token) {
        Ok(jti) => jti,
        Err(_) => {
            return Err(ApiError::AuthenticationFailed(Some(
                "Invalid token format".into(),
            )));
        }
    };
    if let Some(result) = dispatch_refresh_grace(
        &state,
        &refresh_token,
        &refresh_jti,
        state.repos.session.lookup_refresh_grace(&refresh_jti).await,
    )
    .await
    {
        return result;
    }
    let session_row = match state
        .repos
        .session
        .get_session_for_refresh(&refresh_jti)
        .await
    {
        Ok(Some(row)) => row,
        Ok(None) => {
            return dispatch_refresh_grace(
                &state,
                &refresh_token,
                &refresh_jti,
                state.repos.session.lookup_refresh_grace(&refresh_jti).await,
            )
            .await
            .unwrap_or_else(|| {
                Err(ApiError::AuthenticationFailed(Some(
                    "Invalid refresh token".into(),
                )))
            });
        }
        Err(e) => {
            error!("Database error fetching session: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    let key_bytes = match tranquil_pds::config::decrypt_key(
        &session_row.key_bytes,
        Some(session_row.encryption_version),
    ) {
        Ok(k) => k,
        Err(e) => {
            error!("Failed to decrypt user key: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    if tranquil_pds::auth::verify_refresh_token(&refresh_token, &key_bytes).is_err() {
        return Err(ApiError::AuthenticationFailed(Some(
            "Invalid refresh token".into(),
        )));
    }
    let new_access_meta = match tranquil_pds::auth::create_access_token_with_delegation(
        &session_row.did,
        &key_bytes,
        session_row.scope.as_deref(),
        session_row.controller_did.as_ref(),
        None,
    ) {
        Ok(m) => m,
        Err(e) => {
            error!("Failed to create access token: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    let new_refresh_meta = match tranquil_pds::auth::create_refresh_token_with_metadata(
        &session_row.did,
        &key_bytes,
    ) {
        Ok(m) => m,
        Err(e) => {
            error!("Failed to create refresh token: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    let refresh_data = tranquil_db_traits::SessionRefreshData {
        did: session_row.did.clone(),
        old_refresh_jti: refresh_jti.clone(),
        session_id: session_row.id,
        new_access_jti: new_access_meta.jti.clone(),
        new_refresh_jti: new_refresh_meta.jti.clone(),
        new_access_expires_at: new_access_meta.expires_at,
        new_refresh_expires_at: new_refresh_meta.expires_at,
    };
    let (access_jwt, refresh_jwt) = match state
        .repos
        .session
        .refresh_session_atomic(&refresh_data)
        .await
    {
        Ok(tranquil_db_traits::RefreshSessionResult::Success) => {
            (new_access_meta.token, new_refresh_meta.token)
        }
        Ok(tranquil_db_traits::RefreshSessionResult::GraceReplay(replay)) => {
            // Lost a benign concurrent rotation; re-mint the winner's tokens
            // using this session's signing key (same user, unchanged by rotation).
            info!(
                "Concurrent refresh within grace window for session_id: {}; replaying tokens",
                session_row.id
            );
            remint_grace_tokens(&replay, &key_bytes)?
        }
        Ok(tranquil_db_traits::RefreshSessionResult::Compromise) => {
            warn!(
                "Refresh token reuse outside grace window or unreplayable rotation conflict for session_id: {}",
                session_row.id
            );
            return Err(ApiError::AuthenticationFailed(Some(
                "Refresh token has been revoked due to suspected compromise".into(),
            )));
        }
        Err(e) => {
            error!("Database error during session refresh: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    build_refresh_session_output(&state, session_row.did, access_jwt, refresh_jwt).await
}

async fn dispatch_refresh_grace(
    state: &AppState,
    refresh_token: &str,
    presented_jti: &str,
    lookup: Result<tranquil_db_traits::RefreshGraceLookup, tranquil_db_traits::DbError>,
) -> Option<Result<Json<RefreshSessionOutput>, ApiError>> {
    match lookup {
        Ok(tranquil_db_traits::RefreshGraceLookup::NotUsed) => None,
        Ok(tranquil_db_traits::RefreshGraceLookup::Replay(replay)) => {
            Some(serve_refresh_grace_replay(state, refresh_token, presented_jti, replay).await)
        }
        Ok(tranquil_db_traits::RefreshGraceLookup::Compromised {
            did,
            session_id,
            key_bytes,
            encryption_version,
        }) => Some(Err(revoke_compromised_session(
            state,
            refresh_token,
            presented_jti,
            did,
            session_id,
            key_bytes,
            encryption_version,
        )
        .await)),
        Err(e) => {
            error!("Database error checking refresh token grace: {:?}", e);
            Some(Err(ApiError::InternalError(None)))
        }
    }
}

async fn serve_refresh_grace_replay(
    state: &AppState,
    refresh_token: &str,
    presented_jti: &str,
    replay: tranquil_db_traits::RefreshGraceReplay,
) -> Result<Json<RefreshSessionOutput>, ApiError> {
    let key =
        match tranquil_pds::config::decrypt_key(&replay.key_bytes, Some(replay.encryption_version))
        {
            Ok(k) => k,
            Err(e) => {
                error!("Failed to decrypt user key for grace replay: {:?}", e);
                return Err(ApiError::InternalError(None));
            }
        };
    if tranquil_pds::auth::verify_refresh_token(refresh_token, &key).is_err() {
        return Err(ApiError::AuthenticationFailed(Some(
            "Invalid refresh token".into(),
        )));
    }
    info!("Refresh token reuse within grace window for jti: {presented_jti}; replaying tokens");
    let (access_jwt, refresh_jwt) = remint_grace_tokens(&replay, &key)?;
    build_refresh_session_output(state, replay.did, access_jwt, refresh_jwt).await
}

async fn revoke_compromised_session(
    state: &AppState,
    refresh_token: &str,
    presented_jti: &str,
    did: Did,
    session_id: SessionId,
    key_bytes: Vec<u8>,
    encryption_version: i32,
) -> ApiError {
    let key = match tranquil_pds::config::decrypt_key(&key_bytes, Some(encryption_version)) {
        Ok(k) => k,
        Err(e) => {
            error!("Failed to decrypt user key for grace check: {:?}", e);
            return ApiError::InternalError(None);
        }
    };
    if tranquil_pds::auth::verify_refresh_token(refresh_token, &key).is_err() {
        return ApiError::AuthenticationFailed(Some("Invalid refresh token".into()));
    }
    warn!("Refresh token reuse outside grace window for jti: {presented_jti}; revoking session");
    if let Err(e) = state
        .repos
        .session
        .delete_session_by_id(session_id, &did)
        .await
    {
        error!(
            "Failed to revoke session {} for refresh token reuse: {:?}",
            session_id.as_i32(),
            e
        );
        return ApiError::InternalError(None);
    }
    ApiError::AuthenticationFailed(Some(
        "Refresh token has been revoked due to suspected compromise".into(),
    ))
}

/// Re-mint the access/refresh JWTs for a grace-window replay from the session's
/// current jtis and signing key. We never persist the signed JWTs; they are
/// reconstructed on demand so a benignly-racing client converges on the same
/// credentials the winning rotation produced. `key_bytes` is the owning user's
/// already-decrypted signing key.
fn remint_grace_tokens(
    replay: &tranquil_db_traits::RefreshGraceReplay,
    key_bytes: &[u8],
) -> Result<(String, String), ApiError> {
    let access_jwt = tranquil_pds::auth::create_access_token_with_jti(
        &replay.did,
        key_bytes,
        replay.scope.as_deref(),
        replay.controller_did.as_ref(),
        None,
        &replay.access_jti,
        replay.access_expires_at,
    )
    .map_err(|e| {
        error!("Failed to re-mint access token for grace replay: {:?}", e);
        ApiError::InternalError(None)
    })?;
    let refresh_jwt = tranquil_pds::auth::create_refresh_token_with_jti(
        &replay.did,
        key_bytes,
        &replay.refresh_jti,
        replay.refresh_expires_at,
    )
    .map_err(|e| {
        error!("Failed to re-mint refresh token for grace replay: {:?}", e);
        ApiError::InternalError(None)
    })?;
    Ok((access_jwt, refresh_jwt))
}

async fn build_refresh_session_output(
    state: &AppState,
    did: Did,
    access_jwt: String,
    refresh_jwt: String,
) -> Result<Json<RefreshSessionOutput>, ApiError> {
    let did_resolver = state.did_resolver.clone();
    let (db_result, did_doc) = tokio::join!(
        state.repos.user.get_session_info_by_did(&did),
        did_resolver.fetch_did_document(&did)
    );
    match db_result {
        Ok(Some(u)) => {
            let preferred_channel_verified = u
                .channel_verification
                .is_verified(u.preferred_comms_channel);
            let handle = u.handle.clone();
            let account_state =
                AccountState::from_db_fields(u.deactivated_at, u.takedown_ref.clone(), None, None);
            Ok(Json(RefreshSessionOutput {
                access_jwt,
                refresh_jwt,
                handle,
                did,
                email: u.email,
                email_confirmed: u.channel_verification.email,
                preferred_channel: u.preferred_comms_channel.as_str().to_string(),
                preferred_channel_verified,
                preferred_locale: u.preferred_locale,
                is_admin: u.is_admin,
                active: account_state.is_active(),
                did_doc: did_doc.ok(),
                status: account_state.status_for_session().map(String::from),
            }))
        }
        Ok(None) => {
            error!("User not found for existing session: {}", did);
            Err(ApiError::InternalError(None))
        }
        Err(e) => {
            error!("Database error fetching user: {:?}", e);
            Err(ApiError::InternalError(None))
        }
    }
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ConfirmSignupInput {
    pub did: Did,
    pub verification_code: String,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ConfirmSignupOutput {
    pub access_jwt: String,
    pub refresh_jwt: String,
    pub handle: Handle,
    pub did: Did,
    pub email: Option<String>,
    pub email_verified: bool,
    pub preferred_channel: tranquil_db_traits::CommsChannel,
    pub preferred_channel_verified: bool,
}

pub async fn confirm_signup(
    State(state): State<AppState>,
    Json(input): Json<ConfirmSignupInput>,
) -> Result<Json<ConfirmSignupOutput>, ApiError> {
    info!("confirm_signup called for DID: {}", input.did);
    let row = match state.repos.user.get_confirm_signup_by_did(&input.did).await {
        Ok(Some(row)) => row,
        Ok(None) => {
            warn!("User not found for confirm_signup: {}", input.did);
            return Err(ApiError::InvalidRequest(
                "Invalid DID or verification code".into(),
            ));
        }
        Err(e) => {
            error!("Database error in confirm_signup: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };

    let Some(id) = row.channel_identifier() else {
        warn!(
            did = %input.did,
            channel = ?row.channel,
            "We can't confirm signup because the account doesn't have an identifier on file"
        );
        return Err(ApiError::InvalidRequest("Invalid verification code".into()));
    };

    let normalized_token =
        tranquil_pds::auth::verification_token::normalize_token_input(&input.verification_code);
    match tranquil_pds::auth::verification_token::verify_signup_token(
        &normalized_token,
        row.channel,
        id,
    ) {
        Ok(token_data) => {
            if token_data.did != input.did {
                warn!(
                    "Token DID mismatch for confirm_signup: expected {}, got {}",
                    input.did, token_data.did
                );
                return Err(ApiError::InvalidRequest("Invalid verification code".into()));
            }
        }
        Err(tranquil_pds::auth::verification_token::VerifyError::Expired) => {
            warn!("Verification code expired for user: {}", input.did);
            return Err(ApiError::ExpiredToken(Some(
                "Verification code has expired".into(),
            )));
        }
        Err(e) => {
            warn!("Invalid verification code for user {}: {:?}", input.did, e);
            return Err(ApiError::InvalidRequest("Invalid verification code".into()));
        }
    }

    let key_bytes = match tranquil_pds::config::decrypt_key(&row.key_bytes, row.encryption_version)
    {
        Ok(k) => k,
        Err(e) => {
            error!("Failed to decrypt user key: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };

    if let Err(e) = state
        .repos
        .user
        .set_channel_verified(&input.did, row.channel)
        .await
    {
        error!("Failed to update verification status: {:?}", e);
        return Err(ApiError::InternalError(None));
    }

    let session = match crate::identity::provision::create_and_store_session(
        &state,
        &row.did,
        &key_bytes,
        "transition:generic transition:chat.bsky",
        None,
    )
    .await
    {
        Ok(s) => s,
        Err(_) => return Err(ApiError::InternalError(None)),
    };

    let hostname = &tranquil_config::get().server.hostname;
    if let Err(e) = tranquil_pds::comms::comms_repo::enqueue_notice(
        state.repos.user.as_ref(),
        state.repos.infra.as_ref(),
        row.id,
        tranquil_pds::comms::Notice::Welcome,
        hostname,
    )
    .await
    {
        warn!("Failed to enqueue welcome notification: {:?}", e);
    }
    Ok(Json(ConfirmSignupOutput {
        access_jwt: session.access_jwt,
        refresh_jwt: session.refresh_jwt,
        handle: row.handle,
        did: row.did,
        email: row.email,
        email_verified: matches!(row.channel, tranquil_db_traits::CommsChannel::Email),
        preferred_channel: row.channel,
        preferred_channel_verified: true,
    }))
}

async fn resend_signup_verification(
    state: &AppState,
    row: &tranquil_db_traits::UserResendVerification,
    did: &Did,
    context: &'static str,
) -> bool {
    let Some(id) = row.channel_identifier() else {
        warn!(did = %did, channel = ?row.channel, "We skipped {context} because the account doesn't have a recipient on file");
        return false;
    };
    match tranquil_pds::comms::VerificationTarget::resolve(row.channel, id, row.email.as_deref()) {
        Ok(target) => {
            crate::identity::provision::enqueue_signup_verification(
                state,
                row.id,
                did,
                row.channel,
                &target,
            )
            .await;
            true
        }
        Err(_) => {
            warn!(did = %did, channel = ?row.channel, "We skipped {context} because the account doesn't have a valid recipient");
            false
        }
    }
}

const AUTO_VERIFY_DEBOUNCE: std::time::Duration = std::time::Duration::from_secs(120);

pub struct AutoResendResult {
    pub handle: tranquil_types::Handle,
    pub channel: tranquil_db_traits::CommsChannel,
}

pub async fn auto_resend_verification(state: &AppState, did: &Did) -> Option<AutoResendResult> {
    let debounce_key = tranquil_pds::cache_keys::auto_verify_sent_key(did);
    let debounced = state.cache.get(&debounce_key).await.is_some();
    let row = match state.repos.user.get_resend_verification_by_did(did).await {
        Ok(Some(row)) => row,
        Ok(None) => return None,
        Err(e) => {
            warn!(
                "Failed to fetch resend verification info for {}: {:?}",
                did, e
            );
            return None;
        }
    };
    if row.channel_verification.has_any_verified() {
        return None;
    }
    let result = AutoResendResult {
        handle: row.handle.clone(),
        channel: row.channel,
    };
    if row.channel.verifies_via_bot() || debounced {
        return Some(result);
    }
    if resend_signup_verification(state, &row, did, "auto-resend verification").await {
        let _ = state
            .cache
            .set(&debounce_key, "1", AUTO_VERIFY_DEBOUNCE)
            .await;
    }
    Some(result)
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ResendVerificationInput {
    pub did: Did,
}

pub async fn resend_verification(
    State(state): State<AppState>,
    Json(input): Json<ResendVerificationInput>,
) -> Result<Json<SuccessResponse>, ApiError> {
    info!("resend_verification called for DID: {}", input.did);
    let row = match state
        .repos
        .user
        .get_resend_verification_by_did(&input.did)
        .await
    {
        Ok(Some(row)) => row,
        Ok(None) => {
            return Err(ApiError::InvalidRequest("User not found".into()));
        }
        Err(e) => {
            error!("Database error in resend_verification: {:?}", e);
            return Err(ApiError::InternalError(None));
        }
    };
    if row.channel_verification.has_any_verified() {
        return Err(ApiError::InvalidRequest(
            "Account is already verified".into(),
        ));
    }
    resend_signup_verification(&state, &row, &input.did, "resend verification").await;
    Ok(Json(SuccessResponse { success: true }))
}

#[derive(Serialize)]
#[serde(rename_all = "lowercase")]
pub enum SessionType {
    Legacy,
    OAuth,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SessionInfo {
    pub id: String,
    pub session_type: SessionType,
    pub client_name: Option<String>,
    pub created_at: String,
    pub expires_at: String,
    pub is_current: bool,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ListSessionsOutput {
    pub sessions: Vec<SessionInfo>,
}

pub async fn list_sessions(
    State(state): State<AppState>,
    headers: HeaderMap,
    auth: Auth<Active>,
) -> Result<Json<ListSessionsOutput>, ApiError> {
    let current_jti = tranquil_pds::auth::extract_jti_from_headers(&headers);

    let jwt_rows = state
        .repos
        .session
        .list_sessions_by_did(&auth.did)
        .await
        .log_db_err("fetching JWT sessions")?;

    let oauth_rows = state
        .repos
        .oauth
        .list_sessions_by_did(&auth.did)
        .await
        .log_db_err("fetching OAuth sessions")?;

    let jwt_sessions = jwt_rows.into_iter().map(|row| SessionInfo {
        id: format!("jwt:{}", row.id),
        session_type: SessionType::Legacy,
        client_name: None,
        created_at: row.created_at.to_rfc3339(),
        expires_at: row.refresh_expires_at.to_rfc3339(),
        is_current: current_jti.as_ref() == Some(&row.access_jti),
    });

    let is_oauth = auth.is_oauth();
    let oauth_sessions = oauth_rows.into_iter().map(|row| {
        let client_name = extract_client_name(&row.client_id);
        let is_current_oauth = is_oauth && current_jti.as_deref() == Some(row.token_id.as_str());
        SessionInfo {
            id: format!("oauth:{}", row.id),
            session_type: SessionType::OAuth,
            client_name: Some(client_name),
            created_at: row.created_at.to_rfc3339(),
            expires_at: row.expires_at.to_rfc3339(),
            is_current: is_current_oauth,
        }
    });

    let mut sessions: Vec<SessionInfo> = jwt_sessions.chain(oauth_sessions).collect();
    sessions.sort_by(|a, b| b.created_at.cmp(&a.created_at));

    Ok(Json(ListSessionsOutput { sessions }))
}

fn extract_client_name(client_id: &tranquil_types::ClientId) -> String {
    if client_id.starts_with("http://localhost") || client_id.starts_with("http://127.0.0.1") {
        "Localhost App".to_string()
    } else if let Ok(parsed) = reqwest::Url::parse(client_id.as_str()) {
        parsed.host_str().unwrap_or("Unknown App").to_string()
    } else {
        client_id.as_str().to_string()
    }
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RevokeSessionInput {
    pub session_id: String,
}

pub async fn revoke_session(
    State(state): State<AppState>,
    auth: Auth<Active>,
    Json(input): Json<RevokeSessionInput>,
) -> Result<Json<EmptyResponse>, ApiError> {
    if let Some(jwt_id) = input.session_id.strip_prefix("jwt:") {
        let session_id = jwt_id
            .parse::<i32>()
            .map(SessionId::new)
            .map_err(|_| ApiError::InvalidRequest("Invalid session ID".into()))?;
        let access_jti = state
            .repos
            .session
            .get_session_access_jti_by_id(session_id, &auth.did)
            .await
            .log_db_err("in revoke_session")?
            .ok_or(ApiError::SessionNotFound)?;
        state
            .repos
            .session
            .delete_session_by_id(session_id, &auth.did)
            .await
            .log_db_err("deleting session")?;
        let cache_key = tranquil_pds::cache_keys::session_key(&auth.did, &access_jti);
        if let Err(e) = state.cache.delete(&cache_key).await {
            warn!("Failed to invalidate session cache: {:?}", e);
        }
        info!(did = %&auth.did, session_id = %session_id, "JWT session revoked");
    } else if let Some(oauth_id) = input.session_id.strip_prefix("oauth:") {
        let session_id = oauth_id
            .parse::<i32>()
            .map(TokenFamilyId::new)
            .map_err(|_| ApiError::InvalidRequest("Invalid session ID".into()))?;
        let deleted = state
            .repos
            .oauth
            .delete_session_by_id(session_id, &auth.did)
            .await
            .log_db_err("deleting OAuth session")?;
        if deleted == 0 {
            return Err(ApiError::SessionNotFound);
        }
        info!(did = %&auth.did, session_id = %session_id, "OAuth session revoked");
    } else {
        return Err(ApiError::InvalidRequest("Invalid session ID format".into()));
    }
    Ok(Json(EmptyResponse {}))
}

pub async fn revoke_all_sessions(
    State(state): State<AppState>,
    headers: HeaderMap,
    auth: Auth<Active>,
) -> Result<Json<SuccessResponse>, ApiError> {
    let jti = tranquil_pds::auth::extract_jti_from_headers(&headers)
        .ok_or(ApiError::InvalidToken(None))?;

    if auth.is_oauth() {
        state
            .repos
            .session
            .delete_sessions_by_did(&auth.did)
            .await
            .log_db_err("revoking JWT sessions")?;
        let token_id = TokenId::from(jti.clone().into_inner());
        state
            .repos
            .oauth
            .delete_sessions_by_did_except(&auth.did, &token_id)
            .await
            .log_db_err("revoking OAuth sessions")?;
    } else {
        state
            .repos
            .session
            .delete_sessions_by_did_except_jti(&auth.did, &jti)
            .await
            .log_db_err("revoking JWT sessions")?;
        state
            .repos
            .oauth
            .delete_sessions_by_did(&auth.did)
            .await
            .log_db_err("revoking OAuth sessions")?;
    }

    info!(did = %&auth.did, "All other sessions revoked");
    Ok(Json(SuccessResponse { success: true }))
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct LegacyLoginPreferenceOutput {
    pub allow_legacy_login: bool,
    pub has_mfa: bool,
}

pub async fn get_legacy_login_preference(
    State(state): State<AppState>,
    auth: Auth<Active>,
) -> Result<Json<LegacyLoginPreferenceOutput>, ApiError> {
    let pref = state
        .repos
        .user
        .get_legacy_login_pref(&auth.did)
        .await
        .log_db_err("getting legacy login pref")?
        .ok_or(ApiError::AccountNotFound)?;
    Ok(Json(LegacyLoginPreferenceOutput {
        allow_legacy_login: pref.allow_legacy_login,
        has_mfa: pref.has_mfa,
    }))
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateLegacyLoginInput {
    pub allow_legacy_login: bool,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateLegacyLoginOutput {
    pub allow_legacy_login: bool,
}

pub async fn update_legacy_login_preference(
    State(state): State<AppState>,
    auth: Auth<Active>,
    Json(input): Json<UpdateLegacyLoginInput>,
) -> Result<Json<UpdateLegacyLoginOutput>, ApiError> {
    let session_mfa = require_legacy_session_mfa(&state, &auth).await?;

    let reauth_mfa = require_reauth_window(&state, &auth).await?;

    let updated = state
        .repos
        .user
        .update_legacy_login(reauth_mfa.did(), input.allow_legacy_login)
        .await
        .log_db_err("updating legacy login")?;
    if !updated {
        return Err(ApiError::AccountNotFound);
    }
    info!(
        did = %session_mfa.did(),
        allow_legacy_login = input.allow_legacy_login,
        "Legacy login preference updated"
    );
    Ok(Json(UpdateLegacyLoginOutput {
        allow_legacy_login: input.allow_legacy_login,
    }))
}

use tranquil_pds::comms::VALID_LOCALES;

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateLocaleInput {
    pub preferred_locale: String,
}

pub async fn update_locale(
    State(state): State<AppState>,
    auth: Auth<Active>,
    Json(input): Json<UpdateLocaleInput>,
) -> Result<Json<PreferredLocaleOutput>, ApiError> {
    if !VALID_LOCALES.contains(&input.preferred_locale.as_str()) {
        return Err(ApiError::InvalidRequest(format!(
            "Invalid locale. Valid options: {}",
            VALID_LOCALES.join(", ")
        )));
    }

    let updated = state
        .repos
        .user
        .update_locale(&auth.did, &input.preferred_locale)
        .await
        .log_db_err("updating locale")?;
    if !updated {
        return Err(ApiError::AccountNotFound);
    }
    info!(
        did = %&auth.did,
        locale = %input.preferred_locale,
        "User locale preference updated"
    );
    Ok(Json(PreferredLocaleOutput {
        preferred_locale: Some(input.preferred_locale),
    }))
}