//! HTTP server binary: PostgreSQL-backed Athena Auth with the full plugin set.

#[path = "server/auth_route_labels.rs"]
mod auth_route_labels;
#[path = "server/request_tracing.rs"]
mod request_tracing;
#[path = "server/schema_debug_html.rs"]
mod schema_debug_html;
#[path = "server/wide_events.rs"]
mod wide_events;

use auth_route_labels::AuthRouteLabelResolver;
use request_tracing::{
    ensure_request_trace_id, ensure_response_trace_headers, json_error_response, response_trace_id,
};
use std::env;
use std::fs;
use std::io::Write;
use std::net::{IpAddr, SocketAddr};
use std::path::Path;
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, Instant};
use wide_events::{
    LoggingConfig, LoggingFormat, LoggingRuntime, WideRequestEventInput, emit_request_wide_event,
};

use std::collections::{BTreeMap, BTreeSet};

use athena_auth::adapters::{PoolConfig, SqlxAdapter};
use athena_auth::core::router_composer::{AxumRouterComposer, RouterComposer, V07PathCheckPolicy};
use athena_auth::plugins::{
    AccountManagementPlugin, AdminPlugin, ApiKeyPlugin, EmailPasswordPlugin,
    EmailVerificationPlugin, OAuthPlugin, OrganizationPlugin, PasskeyPlugin,
    PasswordManagementPlugin, SessionManagementPlugin, TwoFactorPlugin, UserManagementPlugin,
};
use athena_auth::{AthenaAuth, dispatch_auth_axum_request};
use athena_auth::{
    AuthBuilder, AuthConfig, AvatarStorageConfig, AvatarStorageUrlStyle, CorsConfig,
    DatabaseLoggingEmailProvider, EmailProvider, GeoIpProvider, SameSite, SmtpEmailProvider,
    normalize_ip_address_value, resolve_client_ip_from_headers, snapshot_origin_trust_debug,
};
use athena_auth_multitenancy::{
    DEFAULT_AUTH_SCHEMA, DEFAULT_TENANT_HEADER, FALLBACK_TENANT_HEADERS, HeaderLookup,
    HeaderTenantResolver, MultiTenancyConfig, ParsedMultiTenancyConfig, SearchPath,
    TenantDatabaseUrlSource, TenantId, TenantRuntimeRegistry,
};
use athena_auth_observability::{
    HttpMetrics, HttpMetricsConfig as ObservabilityHttpMetricsConfig, HttpRequestObservation,
};
use axum::Json;
use axum::Router;
use axum::extract::{MatchedPath, Query, Request, State};
use axum::http::{HeaderMap, HeaderValue, StatusCode, Uri};
use axum::middleware::Next;
use axum::response::IntoResponse;
use axum::response::Response;
use axum::routing::any;
use clap::Parser;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use sqlx::Row;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
use tokio::net::TcpListener;
use tokio::sync::RwLock;
use tower::ServiceExt;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use url::Url;

type DynErr = Box<dyn std::error::Error + Send + Sync + 'static>;
const DEFAULT_CONFIG_FILE_PATH: &str = "config.yaml";
const DEFAULT_CONFIG_ENV_PREFIX: &str = "ATHENA_AUTH_CONFIG";
const DEFAULT_METRICS_PATH: &str = "/metrics";
const DEFAULT_METRICS_DASHBOARD_PATH: &str = "/metrics/view";
const DEFAULT_METRICS_PREFIX: &str = "athena_auth";
const DEBUG_SCHEMA_ENDPOINT_PATH: &str = "/debug/schema";
const DEBUG_SCHEMA_HTML_ENDPOINT_PATH: &str = "/debug/schema/view";
const DEBUG_SCHEMA_UNLOCK_HINT: &str = "Pass ?jwt_secret=<JWT_SECRET> where value matches env JWT_SECRET to unlock expanded diagnostics.";
const DEFAULT_PUBLIC_AUTH_BASE_PATH_PREFIX: &str = "/api/auth";
const WELL_KNOWN_WEBAUTHN_PATH: &str = "/.well-known/webauthn";
const INTERNAL_AXUM_REMOTE_IP_HEADER: &str = "x-athena-client-ip";
const AUTH_PROVISION_SQL: &str =
    include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/sql/provision.sql"));
const SELF_HEAL_AUDIT_SCHEMA: &str = "athena";
const SELF_HEAL_AUDIT_TABLE_DDL: &str = "audit_log_ddl";
const DEBUG_MAPPING_APPLICATION_NOTE: &str = "Runtime SQL resolves table names from Auth*Meta traits and always schema-qualifies auth tables with schema_detection.auth_schema (default: athena). Tenant search_path data is reported for diagnostics only; it is not a fallback that redirects auth-table reads into public.*. Table-name remapping entries are currently reported for diagnostics and compatibility planning. Schema debug also reports automatic singular/plural secondary table variants (for example users -> user) with explicit boolean readiness flags.";
const LOGICAL_TABLE_NAMES: [&str; 19] = [
    "users",
    "sessions",
    "accounts",
    "organization",
    "member",
    "invitation",
    "verifications",
    "two_factor",
    "api_keys",
    "passkeys",
    "emails",
    "email_send_failures",
    "email_templates",
    "email_event_types",
    "admin_audit_logs",
    "athena_clients",
    "oauth_clients",
    "ip_profiles",
    "session_ip_profiles",
];
const EXPECTED_COLUMNS_USERS: &[&str] = &[
    "id",
    "name",
    "email",
    "email_verified",
    "image",
    "username",
    "display_username",
    "two_factor_enabled",
    "role",
    "banned",
    "ban_reason",
    "ban_expires",
    "last_sign_in_at",
    "metadata",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_SESSIONS: &[&str] = &[
    "id",
    "expires_at",
    "token",
    "ip_address",
    "user_agent",
    "user_id",
    "impersonated_by",
    "active_organization_id",
    "active",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_ACCOUNTS: &[&str] = &[
    "id",
    "account_id",
    "provider_id",
    "user_id",
    "access_token",
    "refresh_token",
    "id_token",
    "access_token_expires_at",
    "refresh_token_expires_at",
    "scope",
    "password",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_ORGANIZATION: &[&str] = &["id", "name", "slug", "created_at", "updated_at"];
const EXPECTED_COLUMNS_MEMBER: &[&str] = &[
    "id",
    "organization_id",
    "user_id",
    "role",
    "created_at",
    "customer_id",
];
const EXPECTED_COLUMNS_INVITATION: &[&str] = &[
    "id",
    "organization_id",
    "email",
    "role",
    "status",
    "inviter_id",
    "expires_at",
    "created_at",
    "customer_id",
];
const EXPECTED_COLUMNS_VERIFICATIONS: &[&str] = &[
    "id",
    "identifier",
    "value",
    "expires_at",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_TWO_FACTOR: &[&str] = &[
    "id",
    "secret",
    "backup_codes",
    "user_id",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_API_KEYS: &[&str] = &[
    "id",
    "name",
    "start",
    "prefix",
    "key",
    "user_id",
    "refill_interval",
    "refill_amount",
    "last_refill_at",
    "enabled",
    "rate_limit_enabled",
    "rate_limit_time_window",
    "rate_limit_max",
    "request_count",
    "remaining",
    "last_request",
    "expires_at",
    "created_at",
    "updated_at",
    "permissions",
    "metadata",
];
const EXPECTED_COLUMNS_PASSKEYS: &[&str] = &[
    "id",
    "name",
    "public_key",
    "user_id",
    "credential_id",
    "counter",
    "device_type",
    "backed_up",
    "transports",
    "created_at",
];
const EXPECTED_COLUMNS_EMAILS: &[&str] = &[
    "id",
    "recipient_email",
    "subject",
    "from_address",
    "from_name",
    "text_body",
    "html_body",
    "provider",
    "metadata",
    "created_at",
    "flow",
    "updated_at",
];
const EXPECTED_COLUMNS_EMAIL_SEND_FAILURES: &[&str] = &[
    "id",
    "user_id",
    "recipient_email",
    "flow",
    "provider",
    "error_message",
    "metadata",
    "created_at",
    "resolved",
    "resolution_note",
    "updated_at",
];
const EXPECTED_COLUMNS_EMAIL_TEMPLATES: &[&str] = &[
    "id",
    "template_key",
    "locale",
    "subject_template",
    "text_template",
    "html_template",
    "variables",
    "is_active",
    "metadata",
    "created_at",
    "updated_at",
    "event_type",
];
const EXPECTED_COLUMNS_EMAIL_EVENT_TYPES: &[&str] = &[
    "event_type",
    "category",
    "description",
    "default_template_key",
    "required_variables",
    "optional_variables",
    "is_active",
    "is_system",
    "metadata",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_ADMIN_AUDIT_LOGS: &[&str] = &[
    "id",
    "actor_user_id",
    "actor_session_token",
    "action",
    "target_type",
    "target_id",
    "success",
    "error_message",
    "metadata",
    "created_at",
];
const EXPECTED_COLUMNS_ATHENA_CLIENTS: &[&str] = &[
    "id",
    "client_name",
    "description",
    "is_active",
    "pg_uri_env_var",
    "metadata",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_OAUTH_CLIENTS: &[&str] = &[
    "id",
    "client_name",
    "provider",
    "description",
    "response_type",
    "grant_type",
    "token_authentication_method",
    "redirect_uris",
    "client_url",
    "scopes",
    "is_active",
    "metadata",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_IP_PROFILES: &[&str] = &[
    "id",
    "ip_address",
    "provider",
    "country_code",
    "country_name",
    "region_name",
    "city",
    "latitude",
    "longitude",
    "timezone",
    "isp",
    "organization",
    "asn",
    "is_vpn",
    "is_proxy",
    "is_tor",
    "is_hosting",
    "raw_payload",
    "lookup_expires_at",
    "first_seen_at",
    "last_seen_at",
    "created_at",
    "updated_at",
];
const EXPECTED_COLUMNS_SESSION_IP_PROFILES: &[&str] = &[
    "id",
    "session_id",
    "session_token",
    "user_id",
    "ip_address",
    "ip_profile_id",
    "country_code",
    "country_name",
    "is_vpn",
    "is_proxy",
    "is_tor",
    "is_hosting",
    "country_changed",
    "previous_country_code",
    "provider",
    "created_at",
];

#[derive(Debug, Clone, Default, Deserialize)]
struct ServerConfigFile {
    #[serde(default)]
    auth: AuthRuntimeConfig,
    #[serde(default)]
    observability: ObservabilityConfig,
    #[serde(default)]
    multitenancy: MultiTenancyConfig,
    #[serde(default)]
    schema_detection: SchemaDetectionConfig,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct ObservabilityConfig {
    #[serde(default)]
    metrics: MetricsConfig,
    #[serde(default)]
    loki: LokiConfig,
    #[serde(default)]
    logging: LoggingConfig,
}

#[derive(Debug, Clone, Deserialize)]
struct MetricsConfig {
    #[serde(default)]
    enabled: bool,
    #[serde(default = "default_metrics_path")]
    path: String,
    #[serde(default = "default_metrics_dashboard_enabled")]
    dashboard_enabled: bool,
    #[serde(default = "default_metrics_dashboard_path")]
    dashboard_path: String,
    #[serde(default)]
    prefix: Option<String>,
}

#[derive(Debug, Clone, Deserialize)]
struct SchemaDetectionConfig {
    #[serde(default)]
    enabled: bool,
    #[serde(default)]
    auth_schema: Option<String>,
    #[serde(default = "default_schema_detection_output_path")]
    output_path: String,
    #[serde(default)]
    write_mapping_file: bool,
    #[serde(default = "default_schema_detection_min_column_match_ratio")]
    min_column_match_ratio: f32,
    #[serde(default)]
    table_substitutions: BTreeMap<String, Vec<String>>,
}

impl Default for SchemaDetectionConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            auth_schema: None,
            output_path: default_schema_detection_output_path(),
            write_mapping_file: false,
            min_column_match_ratio: default_schema_detection_min_column_match_ratio(),
            table_substitutions: BTreeMap::new(),
        }
    }
}

impl Default for MetricsConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            path: default_metrics_path(),
            dashboard_enabled: default_metrics_dashboard_enabled(),
            dashboard_path: default_metrics_dashboard_path(),
            prefix: Some(DEFAULT_METRICS_PREFIX.to_string()),
        }
    }
}

#[derive(Debug, Clone, Default, Deserialize)]
struct LokiConfig {
    #[serde(default)]
    enabled: bool,
    url: Option<String>,
    #[serde(default)]
    tenant_id: Option<String>,
    #[serde(default)]
    labels: BTreeMap<String, String>,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct AuthRuntimeConfig {
    #[serde(default)]
    base_url: Option<String>,
    #[serde(default)]
    base_path: Option<String>,
    #[serde(default)]
    public_base_path: Option<String>,
    #[serde(default)]
    trusted_origins: Vec<String>,
    #[serde(default)]
    avatar_storage: AvatarStorageRuntimeConfig,
    #[serde(default)]
    password: PasswordRuntimeConfig,
    #[serde(default)]
    passkey: PasskeyRuntimeConfig,
    #[serde(default)]
    session: SessionRuntimeConfig,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct PasswordRuntimeConfig {
    #[serde(default)]
    argon2: Argon2RuntimeConfig,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct AvatarStorageRuntimeConfig {
    #[serde(default)]
    enabled: bool,
    #[serde(default)]
    bucket: Option<String>,
    #[serde(default)]
    endpoint: Option<String>,
    #[serde(default)]
    access_key: Option<String>,
    #[serde(default)]
    secret_key: Option<String>,
    #[serde(default)]
    url_style: Option<String>,
    #[serde(default)]
    prefix_path: Option<String>,
    #[serde(default)]
    file_key_template: Option<String>,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct Argon2RuntimeConfig {
    #[serde(default)]
    memory_cost: Option<u32>,
    #[serde(default)]
    time_cost: Option<u32>,
    #[serde(default)]
    parallelism: Option<u32>,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct PasskeyRuntimeConfig {
    #[serde(default)]
    related_origins: Vec<String>,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct SessionRuntimeConfig {
    #[serde(default)]
    expires_in_seconds: Option<i64>,
    #[serde(default)]
    update_age_seconds: Option<i64>,
    #[serde(default)]
    disable_session_refresh: Option<bool>,
    #[serde(default)]
    fresh_age_seconds: Option<i64>,
    #[serde(default)]
    cookie_name: Option<String>,
    #[serde(default)]
    cookie_domain: Option<String>,
    #[serde(default)]
    cookie_secure: Option<bool>,
    #[serde(default)]
    cookie_http_only: Option<bool>,
    #[serde(default)]
    cookie_same_site: Option<String>,
    #[serde(default)]
    trusted_proxy_headers: Vec<String>,
    #[serde(default)]
    ip: SessionIpRuntimeConfig,
    #[serde(default)]
    intelligence: SessionIntelligenceRuntimeConfig,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct SessionIpRuntimeConfig {
    #[serde(default)]
    headers: Vec<String>,
    #[serde(default)]
    disable_tracking: Option<bool>,
    #[serde(default)]
    denylist: Vec<String>,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct SessionIntelligenceRuntimeConfig {
    #[serde(default)]
    enabled: Option<bool>,
    #[serde(default)]
    detect_country_change: Option<bool>,
    #[serde(default)]
    geoip: GeoIpRuntimeConfig,
}

#[derive(Debug, Clone, Default, Deserialize)]
struct GeoIpRuntimeConfig {
    #[serde(default)]
    enabled: Option<bool>,
    #[serde(default)]
    provider: Option<String>,
    #[serde(default)]
    base_url: Option<String>,
    #[serde(default)]
    api_key: Option<String>,
    #[serde(default)]
    api_key_env: Option<String>,
    #[serde(default)]
    api_key_header: Option<String>,
    #[serde(default)]
    timeout_ms: Option<u64>,
    #[serde(default)]
    cache_ttl_seconds: Option<i64>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct ConfigField {
    path: String,
    env_key: String,
}

impl ConfigField {
    fn from_path(path: String) -> Self {
        Self {
            env_key: config_path_to_env_key(&path),
            path,
        }
    }

    fn candidate_env_keys(&self) -> Vec<String> {
        let mut keys: Vec<String> = vec![self.env_key.clone()];
        for alias in legacy_config_env_aliases(self.path.as_str()) {
            if !keys.iter().any(|existing| existing == alias) {
                keys.push((*alias).to_string());
            }
        }
        keys
    }
}

fn config_path_to_env_key(path: &str) -> String {
    let mut key: String = format!("{DEFAULT_CONFIG_ENV_PREFIX}__");
    let normalized_path: String = path
        .split('.')
        .filter(|segment| !segment.is_empty())
        .map(|segment| {
            segment
                .chars()
                .map(|ch| {
                    if ch.is_ascii_alphanumeric() {
                        ch.to_ascii_uppercase()
                    } else {
                        '_'
                    }
                })
                .collect::<String>()
                .trim_matches('_')
                .to_string()
        })
        .filter(|segment| !segment.is_empty())
        .collect::<Vec<String>>()
        .join("__");
    key.push_str(&normalized_path);
    key
}

fn legacy_config_env_aliases(path: &str) -> &'static [&'static str] {
    match path {
        "auth.base_url" => &["BASE_URL"],
        "auth.base_path" => &["AUTH_BASE_PATH"],
        "auth.public_base_path" => &["PUBLIC_AUTH_BASE_PATH"],
        "auth.trusted_origins" => &["TRUSTED_ORIGINS"],
        "auth.session.cookie_domain" => &["ATHENA_AUTH_COOKIE_DOMAIN"],
        "auth.session.ip.denylist" => &["ATHENA_AUTH_BLOCKED_IPS"],
        "auth.password.argon2.memory_cost" => &["PASSWORD_ARGON2_MEMORY_COST"],
        "auth.password.argon2.time_cost" => &["PASSWORD_ARGON2_TIME_COST"],
        "auth.password.argon2.parallelism" => &["PASSWORD_ARGON2_PARALLELISM"],
        "auth.passkey.related_origins" => &["PASSKEY_RELATED_ORIGINS"],
        "schema_detection.auth_schema" => &["ATHENA_AUTH_SCHEMA"],
        _ => &[],
    }
}

struct ConfigResolver<E>
where
    E: Fn(&str) -> Option<String> + Copy,
{
    env_lookup: E,
}

impl<E> ConfigResolver<E>
where
    E: Fn(&str) -> Option<String> + Copy,
{
    fn new(env_lookup: E) -> Self {
        Self { env_lookup }
    }

    fn resolve_server_config(&self, raw: &str) -> Result<ServerConfigFile, DynErr> {
        let mut root: serde_yaml::Value = serde_yaml::from_str(raw).map_err(|error| {
            format!("failed to parse YAML config before env resolution: {error}")
        })?;
        let mut path_stack: Vec<String> = Vec::new();
        self.resolve_value_in_place(&mut root, &mut path_stack)?;
        serde_yaml::from_value(root)
            .map_err(|error| format!("failed to parse resolved YAML config: {error}").into())
    }

    fn collect_fields_from_value(&self, root: &serde_yaml::Value) -> Vec<ConfigField> {
        let mut fields: Vec<ConfigField> = Vec::new();
        let mut path_stack: Vec<String> = Vec::new();
        Self::collect_fields_recursive(root, &mut path_stack, &mut fields);
        fields
    }

    fn resolve_value_in_place(
        &self,
        value: &mut serde_yaml::Value,
        path_stack: &mut Vec<String>,
    ) -> Result<(), DynErr> {
        match value {
            serde_yaml::Value::Mapping(mapping) => {
                for (raw_key, child) in mapping.iter_mut() {
                    let Some(key) = raw_key
                        .as_str()
                        .map(str::trim)
                        .filter(|key| !key.is_empty())
                    else {
                        continue;
                    };
                    path_stack.push(key.to_string());
                    self.resolve_value_in_place(child, path_stack)?;
                    path_stack.pop();
                }
                Ok(())
            }
            serde_yaml::Value::Sequence(sequence) => {
                for child in sequence.iter_mut() {
                    self.resolve_value_in_place(child, path_stack)?;
                }
                self.apply_env_override(value, path_stack)
            }
            serde_yaml::Value::Null
            | serde_yaml::Value::Bool(_)
            | serde_yaml::Value::Number(_)
            | serde_yaml::Value::String(_)
            | serde_yaml::Value::Tagged(_) => self.apply_env_override(value, path_stack),
        }
    }

    fn apply_env_override(
        &self,
        value: &mut serde_yaml::Value,
        path_stack: &[String],
    ) -> Result<(), DynErr> {
        if path_stack.is_empty() {
            return Ok(());
        }

        let field: ConfigField = ConfigField::from_path(path_stack.join("."));
        let mut resolved_value: serde_yaml::Value =
            self.resolve_embedded_env(value, &field.path)?;

        for env_key in field.candidate_env_keys() {
            let Some(env_value) = (self.env_lookup)(env_key.as_str()) else {
                continue;
            };
            let normalized: String = strip_wrapping_quotes(&env_value);
            if normalized.trim().is_empty() {
                continue;
            }
            resolved_value =
                self.parse_env_override_as_yaml_value(&resolved_value, &normalized, &env_key)?;
            tracing::debug!(
                config_field = %field.path,
                env_key = %env_key,
                "applied environment override for config field"
            );
            break;
        }

        *value = resolved_value;
        Ok(())
    }

    fn resolve_embedded_env(
        &self,
        value: &serde_yaml::Value,
        field_path: &str,
    ) -> Result<serde_yaml::Value, DynErr> {
        match value {
            serde_yaml::Value::String(raw) => {
                let resolved: String = self.resolve_env_template(raw, field_path)?;
                Ok(serde_yaml::Value::String(resolved))
            }
            serde_yaml::Value::Sequence(sequence) => {
                let mut resolved: Vec<serde_yaml::Value> = Vec::with_capacity(sequence.len());
                for child in sequence {
                    resolved.push(self.resolve_embedded_env(child, field_path)?);
                }
                Ok(serde_yaml::Value::Sequence(resolved))
            }
            serde_yaml::Value::Mapping(mapping) => {
                let mut resolved: serde_yaml::Mapping = serde_yaml::Mapping::new();
                for (key, child) in mapping {
                    resolved.insert(key.clone(), self.resolve_embedded_env(child, field_path)?);
                }
                Ok(serde_yaml::Value::Mapping(resolved))
            }
            serde_yaml::Value::Tagged(tagged) => {
                let mut tagged: Box<serde_yaml::value::TaggedValue> = tagged.clone();
                tagged.value = self.resolve_embedded_env(&tagged.value, field_path)?;
                Ok(serde_yaml::Value::Tagged(tagged))
            }
            _ => Ok(value.clone()),
        }
    }

    fn resolve_env_template(&self, raw: &str, field_path: &str) -> Result<String, DynErr> {
        let mut output: String = String::with_capacity(raw.len());
        let mut cursor: usize = 0;
        while let Some(start_rel) = raw[cursor..].find("${") {
            let start: usize = cursor + start_rel;
            output.push_str(&raw[cursor..start]);
            let token_start: usize = start + 2;
            let Some(end_rel) = raw[token_start..].find('}') else {
                return Err(format!(
                    "unterminated env template in config field '{field_path}': '{raw}'"
                )
                .into());
            };
            let end: usize = token_start + end_rel;
            let token: &str = raw[token_start..end].trim();
            if token.is_empty() {
                return Err(
                    format!("empty env template in config field '{field_path}': '{raw}'").into(),
                );
            }

            let (env_key, fallback) = if let Some((key, fallback)) = token.split_once(":-") {
                (key.trim(), Some(fallback))
            } else {
                (token, None)
            };

            if env_key.is_empty() {
                return Err(format!(
                    "invalid env template in config field '{field_path}': '{raw}'"
                )
                .into());
            }

            let resolved_piece: String = (self.env_lookup)(env_key)
                .map(|value| strip_wrapping_quotes(&value))
                .or_else(|| fallback.map(strip_wrapping_quotes))
                .ok_or_else(|| {
                    format!(
                        "config field '{field_path}' references unset env '{env_key}' without default"
                    )
                })?;

            output.push_str(&resolved_piece);
            cursor = end + 1;
        }
        output.push_str(&raw[cursor..]);
        Ok(output)
    }

    fn parse_env_override_as_yaml_value(
        &self,
        current: &serde_yaml::Value,
        raw: &str,
        env_key: &str,
    ) -> Result<serde_yaml::Value, DynErr> {
        match current {
            serde_yaml::Value::Bool(_) => parse_bool_like(raw)
                .map(serde_yaml::Value::Bool)
                .ok_or_else(|| {
                    format!(
                        "env '{env_key}' value '{raw}' is not a valid boolean for config override"
                    )
                    .into()
                }),
            serde_yaml::Value::Number(number) => {
                if number.as_i64().is_some() {
                    let parsed: i64 = raw.parse::<i64>().map_err(|error| {
                        format!(
                            "env '{env_key}' value '{raw}' is not a valid i64 config override: {error}"
                        )
                    })?;
                    Ok(serde_yaml::Value::Number(parsed.into()))
                } else if number.as_u64().is_some() {
                    let parsed: u64 = raw.parse::<u64>().map_err(|error| {
                        format!(
                            "env '{env_key}' value '{raw}' is not a valid u64 config override: {error}"
                        )
                    })?;
                    Ok(serde_yaml::Value::Number(parsed.into()))
                } else {
                    let parsed: f64 = raw.parse::<f64>().map_err(|error| {
                        format!(
                            "env '{env_key}' value '{raw}' is not a valid f64 config override: {error}"
                        )
                    })?;
                    serde_yaml::to_value(parsed).map_err(|error| {
                        format!(
                            "failed to convert env '{env_key}' float override to YAML value: {error}"
                        )
                        .into()
                    })
                }
            }
            serde_yaml::Value::Sequence(_) => {
                if raw.trim_start().starts_with('[') {
                    let parsed: serde_yaml::Value = serde_yaml::from_str(raw).map_err(|error| {
                        format!(
                            "env '{env_key}' must contain a YAML sequence for list override: {error}"
                        )
                    })?;
                    if matches!(parsed, serde_yaml::Value::Sequence(_)) {
                        return Ok(parsed);
                    }
                }
                let list = raw
                    .split(',')
                    .map(str::trim)
                    .filter(|item| !item.is_empty())
                    .map(|item| serde_yaml::Value::String(item.to_string()))
                    .collect::<Vec<serde_yaml::Value>>();
                Ok(serde_yaml::Value::Sequence(list))
            }
            serde_yaml::Value::Mapping(_) => {
                let parsed: serde_yaml::Value = serde_yaml::from_str(raw).map_err(|error| {
                    format!(
                        "env '{env_key}' must contain YAML mapping text for object override: {error}"
                    )
                })?;
                if !matches!(parsed, serde_yaml::Value::Mapping(_)) {
                    return Err(format!(
                        "env '{env_key}' override must resolve to a YAML mapping/object"
                    )
                    .into());
                }
                Ok(parsed)
            }
            serde_yaml::Value::Null => Ok(serde_yaml::Value::String(raw.to_string())),
            serde_yaml::Value::String(_) => Ok(serde_yaml::Value::String(raw.to_string())),
            serde_yaml::Value::Tagged(tagged) => {
                self.parse_env_override_as_yaml_value(&tagged.value, raw, env_key)
            }
        }
    }

    fn collect_fields_recursive(
        value: &serde_yaml::Value,
        path_stack: &mut Vec<String>,
        fields: &mut Vec<ConfigField>,
    ) {
        match value {
            serde_yaml::Value::Mapping(mapping) => {
                for (key, child) in mapping {
                    let Some(key) = key
                        .as_str()
                        .map(str::trim)
                        .filter(|value| !value.is_empty())
                    else {
                        continue;
                    };
                    path_stack.push(key.to_string());
                    Self::collect_fields_recursive(child, path_stack, fields);
                    path_stack.pop();
                }
            }
            serde_yaml::Value::Sequence(sequence) => {
                if !path_stack.is_empty() {
                    fields.push(ConfigField::from_path(path_stack.join(".")));
                }
                for child in sequence {
                    Self::collect_fields_recursive(child, path_stack, fields);
                }
            }
            serde_yaml::Value::Null
            | serde_yaml::Value::Bool(_)
            | serde_yaml::Value::Number(_)
            | serde_yaml::Value::String(_)
            | serde_yaml::Value::Tagged(_) => {
                if !path_stack.is_empty() {
                    fields.push(ConfigField::from_path(path_stack.join(".")));
                }
            }
        }
    }
}

fn parse_bool_like(value: &str) -> Option<bool> {
    match value.trim().to_ascii_lowercase().as_str() {
        "1" | "true" | "yes" | "on" => Some(true),
        "0" | "false" | "no" | "off" => Some(false),
        _ => None,
    }
}

type TenantAuthRegistry = TenantRuntimeRegistry<Arc<AthenaAuth<SqlxAdapter>>>;
type SchemaDetectionResolvedCandidate = (String, String, usize, Vec<String>, Vec<String>, f32);

#[derive(Debug, Clone)]
struct RuntimeObservability {
    metrics: Option<Arc<HttpMetrics>>,
    metrics_path: Option<String>,
    metrics_dashboard_path: Option<String>,
    loki_enabled: bool,
    logging: LoggingRuntime,
}

#[derive(Debug, Clone)]
struct ServerMiddlewareState {
    metrics: Option<Arc<HttpMetrics>>,
    logging: LoggingRuntime,
    loki_enabled: bool,
    auth_route_labels: AuthRouteLabelResolver,
    session_cookie_name: String,
    trusted_proxy_headers: Vec<String>,
    ip_headers: Vec<String>,
    blocked_ips: Arc<BTreeSet<String>>,
}

#[derive(Debug, Clone)]
struct AuthRequestTraceContext {
    method: String,
    path: String,
    query: Option<String>,
    origin: Option<String>,
    referer: Option<String>,
    authorization_header_present: bool,
    cookie_header_present: bool,
    session_cookie_name: String,
    session_cookie_present: bool,
    disable_cookie_cache: Option<String>,
}

fn build_auth_route_labels(registry: &TenantAuthRegistry) -> AuthRouteLabelResolver {
    let mut resolver: AuthRouteLabelResolver = AuthRouteLabelResolver::default();
    for tenant_id in registry.known_tenants() {
        if let Some(auth) = registry.runtime(tenant_id) {
            resolver.register_auth(auth.as_ref());
        }
    }
    resolver
}

#[derive(Clone)]
struct SchemaDebugRuntime {
    multitenancy_enabled: bool,
    header_name: String,
    default_tenant: TenantId,
    resolver: HeaderTenantResolver,
    tenants: std::collections::BTreeMap<TenantId, SchemaDebugTenant>,
    access_gate: SchemaDebugAccessGate,
    auth_config: Arc<AuthConfig>,
    enabled_features: Value,
}

#[derive(Debug, Clone)]
struct SchemaDebugAccessGate {
    jwt_secret: Option<String>,
    config_snapshot: SchemaDebugConfigSnapshot,
}

#[derive(Debug, Clone)]
struct SchemaDebugConfigSnapshot {
    path: String,
    exists: bool,
    read_status: &'static str,
    read_error: Option<String>,
    raw_contents: Option<String>,
    resolved_runtime_config: Option<String>,
}

#[derive(Debug, Default, Deserialize)]
struct SchemaDebugQuery {
    jwt_secret: Option<String>,
}

#[derive(Debug)]
struct SchemaDebugAccessEvaluation {
    privileged: bool,
    reason: &'static str,
    query_param_present: bool,
}

#[derive(Debug, Clone)]
struct SchemaDebugTenant {
    is_default: bool,
    database_source_kind: &'static str,
    database_source_key: Option<String>,
    database_url: String,
    database_url_redacted: String,
    configured_auth_schema: String,
    configured_search_path_sql: Option<String>,
    configured_search_path_schemas: Vec<String>,
    configured_mapping: std::collections::BTreeMap<String, SchemaDebugMapping>,
}

#[derive(Debug, Clone, Serialize)]
struct SchemaDebugMapping {
    mapped_table: String,
    source: &'static str,
}

#[derive(Debug)]
struct SchemaDebugRequestContext {
    configured_header_name: String,
    configured_header_value: Option<String>,
    fallback_header_values: BTreeMap<String, Option<String>>,
    resolved_tenant: Option<TenantId>,
    resolution_status: &'static str,
    resolution_error: Option<String>,
}

#[derive(Debug)]
struct SchemaDebugDatabaseProbe {
    current_schema: Option<String>,
    effective_search_path: Vec<String>,
    search_path_setting: Option<String>,
    table_schemas: BTreeMap<String, Vec<String>>,
}

#[derive(Debug)]
struct SchemaCatalogSnapshot {
    current_schema: Option<String>,
    effective_search_path: Vec<String>,
    search_path_setting: Option<String>,
    columns_by_table: BTreeMap<String, BTreeMap<String, Vec<String>>>,
}

struct AxumHeaderLookup<'a> {
    headers: &'a HeaderMap,
}

impl HeaderLookup for AxumHeaderLookup<'_> {
    fn header(&self, name: &str) -> Option<&str> {
        self.headers.get(name).and_then(|value| value.to_str().ok())
    }
}

#[derive(Parser, Debug)]
#[command(name = "server", about = "Athena Auth HTTP server (Axum + PostgreSQL)")]
struct Args {
    #[arg(long)]
    host: Option<String>,
    #[arg(short = 'p', long)]
    port: Option<u16>,
}

/// Returns the default HTTP path for Prometheus metrics exposure.
fn default_metrics_path() -> String {
    DEFAULT_METRICS_PATH.to_string()
}

/// Returns the default HTTP path for the human-readable metrics dashboard.
fn default_metrics_dashboard_path() -> String {
    DEFAULT_METRICS_DASHBOARD_PATH.to_string()
}

/// Keeps the operator-facing dashboard enabled whenever metrics are enabled.
fn default_metrics_dashboard_enabled() -> bool {
    true
}

/// Default location for generated schema-detection mapping output.
fn default_schema_detection_output_path() -> String {
    "schema-mapping.generated.yaml".to_string()
}

/// Default minimum expected-column match ratio for accepting a substitution candidate.
fn default_schema_detection_min_column_match_ratio() -> f32 {
    0.6
}

fn is_valid_schema_identifier(value: &str) -> bool {
    let mut chars: std::str::Chars<'_> = value.chars();
    match chars.next() {
        Some(first) if first.is_ascii_alphabetic() || first == '_' => {
            chars.all(|ch| ch.is_ascii_alphanumeric() || ch == '_')
        }
        _ => false,
    }
}

fn normalize_auth_schema_for_runtime(auth_schema: &str) -> Result<String, DynErr> {
    let normalized: &str = auth_schema.trim();
    let schema: &str = if normalized.is_empty() {
        DEFAULT_AUTH_SCHEMA
    } else {
        normalized
    };

    if !is_valid_schema_identifier(schema) {
        return Err(format!(
            "invalid auth schema value '{schema}'; expected [a-zA-Z_][a-zA-Z0-9_]*"
        )
        .into());
    }

    Ok(schema.to_string())
}

fn schema_detection_auth_schema(
    config_file: Option<&ServerConfigFile>,
) -> Result<Option<String>, DynErr> {
    let configured: Option<&str> = config_file
        .and_then(|cfg| cfg.schema_detection.auth_schema.as_deref())
        .map(str::trim)
        .filter(|value| !value.is_empty());

    let Some(schema) = configured else {
        return Ok(None);
    };

    if !is_valid_schema_identifier(schema) {
        return Err(format!(
            "invalid schema_detection.auth_schema value '{schema}'; expected [a-zA-Z_][a-zA-Z0-9_]*"
        )
        .into());
    }

    Ok(Some(schema.to_string()))
}

fn quote_pg_identifier(value: &str) -> String {
    format!("\"{}\"", value.replace('"', "\"\""))
}

fn auth_provision_sql_for_schema(auth_schema: &str) -> String {
    let normalized_auth_schema = auth_schema.trim();
    let effective_auth_schema = if normalized_auth_schema.is_empty() {
        DEFAULT_AUTH_SCHEMA
    } else {
        normalized_auth_schema
    };
    let quoted_schema = quote_pg_identifier(effective_auth_schema);
    AUTH_PROVISION_SQL
        .replace(
            "CREATE SCHEMA IF NOT EXISTS athena;",
            format!("CREATE SCHEMA IF NOT EXISTS {quoted_schema};").as_str(),
        )
        .replace("athena.", format!("{quoted_schema}.").as_str())
}

/// Returns the schema list that must exist before provision SQL is applied.
///
/// `athena` is always created first because shared operational tables and
/// startup audit surfaces are anchored there, while tenant/default auth tables
/// are provisioned in `auth_schema`.
fn schema_bootstrap_targets(auth_schema: &str) -> Vec<&str> {
    if auth_schema == DEFAULT_AUTH_SCHEMA {
        return vec![DEFAULT_AUTH_SCHEMA];
    }

    vec![DEFAULT_AUTH_SCHEMA, auth_schema]
}

fn build_effective_search_path(auth_schema: &str, search_path: Option<&SearchPath>) -> String {
    let normalized_auth_schema: &str = auth_schema.trim();
    let effective_auth_schema: &str = if normalized_auth_schema.is_empty() {
        DEFAULT_AUTH_SCHEMA
    } else {
        normalized_auth_schema
    };
    let auth_schema_fragment: String = quote_pg_identifier(effective_auth_schema);

    if let Some(configured_search_path) = search_path
        .map(SearchPath::as_sql_fragment)
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        return format!("{auth_schema_fragment}, {configured_search_path}");
    }

    auth_schema_fragment
}

fn build_effective_search_path_from_sql(
    auth_schema: &str,
    search_path_sql: Option<&str>,
) -> String {
    let normalized_auth_schema: &str = auth_schema.trim();
    let effective_auth_schema: &str = if normalized_auth_schema.is_empty() {
        DEFAULT_AUTH_SCHEMA
    } else {
        normalized_auth_schema
    };
    let auth_schema_fragment: String = quote_pg_identifier(effective_auth_schema);

    if let Some(configured_search_path) = search_path_sql
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        return format!("{auth_schema_fragment}, {configured_search_path}");
    }

    auth_schema_fragment
}

/// Normalizes a route-like path and falls back when the value is empty.
fn normalize_route_path(path: &str, fallback: &str) -> String {
    let trimmed: &str = path.trim();
    if trimmed.is_empty() {
        return fallback.to_string();
    }
    if trimmed == "/" {
        tracing::info!("trimmed is '/'");
        return "/".to_string();
    }
    if trimmed.starts_with('/') {
        trimmed.trim_end_matches('/').to_string()
    } else {
        format!("/{}", trimmed.trim_end_matches('/'))
    }
}

/// Normalizes schema-mapping key/value pairs for stable debug output.
fn normalize_mapping_for_debug(raw: BTreeMap<String, String>) -> BTreeMap<String, String> {
    raw.into_iter()
        .filter_map(|(key, value)| {
            let key: &str = key.trim();
            let value: &str = value.trim();
            if key.is_empty() || value.is_empty() {
                return None;
            }
            Some((key.to_string(), value.to_string()))
        })
        .collect()
}

/// Parses a quoted SQL search-path fragment into plain schema names.
fn parse_search_path_schemas(search_path_sql: Option<&str>) -> Vec<String> {
    let Some(raw) = search_path_sql
        .map(str::trim)
        .filter(|value| !value.is_empty())
    else {
        return Vec::new();
    };

    raw.split(',')
        .map(str::trim)
        .filter(|segment| !segment.is_empty())
        .map(|segment| {
            segment
                .trim_matches('"')
                .replace("\"\"", "\"")
                .trim()
                .to_string()
        })
        .filter(|segment| !segment.is_empty())
        .collect()
}

/// Parses comma-separated origin values into a deduplicated list.
fn parse_origins_csv(raw: &str) -> Vec<String> {
    raw.split(',')
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
        .fold(Vec::new(), |mut acc, value| {
            if !acc.iter().any(|existing| existing == &value) {
                acc.push(value);
            }
            acc
        })
}

fn parse_blocked_ip_value(raw: &str, source: &str) -> Result<String, DynErr> {
    let normalized = normalize_ip_address_value(raw)
        .ok_or_else(|| format!("invalid blocked IP value '{raw}' configured via {source}"))?;
    let ip = IpAddr::from_str(normalized.as_str()).map_err(|error| {
        format!(
            "blocked IP value '{raw}' from {source} must resolve to an IP address literal: {error}"
        )
    })?;
    Ok(ip.to_string())
}

fn parse_blocked_ips(
    values: impl IntoIterator<Item = String>,
    source: &str,
) -> Result<BTreeSet<String>, DynErr> {
    let mut blocked_ips: BTreeSet<String> = BTreeSet::new();
    for raw in values {
        let trimmed = raw.trim();
        if trimmed.is_empty() {
            continue;
        }
        blocked_ips.insert(parse_blocked_ip_value(trimmed, source)?);
    }
    Ok(blocked_ips)
}

fn load_blocked_ips(config_file: Option<&ServerConfigFile>) -> Result<BTreeSet<String>, DynErr> {
    if let Some(raw) = env_var_clean("ATHENA_AUTH_BLOCKED_IPS") {
        let values = raw.split(',').map(str::to_string);
        return parse_blocked_ips(values, "ATHENA_AUTH_BLOCKED_IPS");
    }

    let values = config_file
        .map(|cfg| cfg.auth.session.ip.denylist.clone())
        .unwrap_or_default();
    parse_blocked_ips(values, "auth.session.ip.denylist")
}

/// Normalizes a list of configured origins and removes empty or duplicate values.
fn normalize_origins(origins: &[String]) -> Vec<String> {
    origins
        .iter()
        .map(String::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
        .fold(Vec::new(), |mut acc, value| {
            if !acc.iter().any(|existing| existing == &value) {
                acc.push(value);
            }
            acc
        })
}

/// Extracts a normalized string header value.
fn header_value_trimmed(headers: &HeaderMap, name: &str) -> Option<String> {
    headers
        .get(name)
        .and_then(|value| value.to_str().ok())
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
}

/// Redacts credentials from a PostgreSQL connection URL.
fn redact_database_url(database_url: &str) -> String {
    let Ok(mut parsed) = Url::parse(database_url) else {
        return "<invalid_database_url>".to_string();
    };

    if !parsed.username().is_empty() {
        let _ = parsed.set_username("***");
    }
    if parsed.password().is_some() {
        let _ = parsed.set_password(Some("***"));
    }
    parsed.to_string()
}

const DATABASE_URL_ERROR_PREVIEW_MAX_LEN: usize = 96;

/// Truncates diagnostic text for error messages without exposing oversized env values.
fn truncate_for_error_preview(value: &str, max_len: usize) -> String {
    let mut preview = String::new();
    for ch in value.chars().take(max_len) {
        if ch.is_ascii_graphic() || ch == ' ' {
            preview.push(ch);
        } else {
            preview.push('?');
        }
    }
    if value.len() > max_len {
        preview.push('…');
    }
    preview
}

/// Summarizes a DATABASE_URL value for parse errors without leaking credentials.
fn describe_database_url_for_error(database_url: &str) -> String {
    let raw_len = database_url.len();
    let trimmed = database_url.trim();
    let trimmed_len = trimmed.len();

    let mut details = vec![format!("len={raw_len}")];
    if trimmed_len != raw_len {
        details.push("has_leading_or_trailing_whitespace=true".to_string());
    }

    if trimmed.is_empty() {
        details.push("value=<empty>".to_string());
        return details.join(", ");
    }

    let redacted = redact_database_url(trimmed);
    let preview = truncate_for_error_preview(&redacted, DATABASE_URL_ERROR_PREVIEW_MAX_LEN);
    details.push(format!("preview='{preview}'"));

    let scheme_ok = trimmed.starts_with("postgresql://") || trimmed.starts_with("postgres://");
    if !scheme_ok {
        let prefix = truncate_for_error_preview(trimmed, 24);
        details.push("expected_scheme=postgresql:// or postgres://".to_string());
        details.push(format!("received_prefix='{prefix}'"));
    }

    details.join(", ")
}

/// Parses a PostgreSQL connection URL into sqlx connect options with actionable errors.
fn parse_pg_connect_options(database_url: &str) -> Result<PgConnectOptions, String> {
    let normalized = strip_leading_env_noise(database_url);
    PgConnectOptions::from_str(normalized).map_err(|error| {
        format!(
            "invalid PostgreSQL connection string: {error} ({})",
            describe_database_url_for_error(database_url)
        )
    })
}

/// Resolves the effective database URL and source metadata for a tenant.
fn resolve_database_url_for_debug(
    source: &TenantDatabaseUrlSource,
    tenant_id: &TenantId,
    fallback_database_url: &str,
) -> Result<(String, &'static str, Option<String>), DynErr> {
    let resolved_url: String = source
        .resolve(tenant_id, fallback_database_url, env_var_clean)
        .map_err(|error| {
            format!(
                "failed to resolve database URL for tenant '{}': {error}",
                tenant_id
            )
        })?;

    let (source_kind, source_key) = match source {
        TenantDatabaseUrlSource::Explicit(_) => ("explicit", None),
        TenantDatabaseUrlSource::EnvVar(env_key) => ("env_var", Some(env_key.clone())),
        TenantDatabaseUrlSource::FallbackDatabaseUrl => ("fallback_database_url", None),
    };

    Ok((resolved_url, source_kind, source_key))
}

/// Builds a non-sensitive runtime feature matrix for `/debug/schema`.
fn build_schema_debug_enabled_features(
    file_config: Option<&ServerConfigFile>,
    parsed_multi: &ParsedMultiTenancyConfig,
    schema_detection_auth_schema: Option<&str>,
) -> Value {
    let auth_cfg: AuthRuntimeConfig = file_config
        .map(|value| value.auth.clone())
        .unwrap_or_default();
    let observability_cfg: ObservabilityConfig = file_config
        .map(|value| value.observability.clone())
        .unwrap_or_default();
    let schema_detection_cfg: SchemaDetectionConfig = file_config
        .map(|value| value.schema_detection.clone())
        .unwrap_or_default();
    let base_url_env: Option<String> = env_var_clean("BASE_URL");
    let auth_base_path_env: Option<String> = env_var_clean("AUTH_BASE_PATH");
    let public_auth_base_path_env: Option<String> = env_var_clean("PUBLIC_AUTH_BASE_PATH");

    let trusted_origins_env: Option<String> = env_var_clean("TRUSTED_ORIGINS");
    let trusted_origins_env_entries: Vec<String> = trusted_origins_env
        .as_deref()
        .map(|value| {
            value
                .split(',')
                .map(str::trim)
                .filter(|entry| !entry.is_empty())
                .map(str::to_string)
                .collect()
        })
        .unwrap_or_default();

    let session_cfg: SessionRuntimeConfig = auth_cfg.session.clone();
    let cookie_domain_env: Option<String> =
        env_var_clean("ATHENA_AUTH_CONFIG__AUTH__SESSION__COOKIE_DOMAIN")
            .or_else(|| env_var_clean("ATHENA_AUTH_COOKIE_DOMAIN"));
    let cookie_domain_configured_in_yaml = session_cfg
        .cookie_domain
        .as_deref()
        .is_some_and(|value| !value.trim().is_empty());
    let cookie_domain_configured_in_env = cookie_domain_env
        .as_deref()
        .is_some_and(|value| !value.trim().is_empty());
    let geoip_cfg: GeoIpRuntimeConfig = session_cfg.intelligence.geoip.clone();
    let geoip_api_key_from_env: bool = geoip_cfg
        .api_key_env
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .and_then(env_var_clean)
        .is_some_and(|value| !value.is_empty());
    let geoip_api_key_inline: bool = geoip_cfg
        .api_key
        .as_deref()
        .is_some_and(|value| !value.trim().is_empty());

    let passkey_rp_id_env: Option<String> = env_var_clean("PASSKEY_RP_ID");
    let passkey_rp_name_env: Option<String> = env_var_clean("PASSKEY_RP_NAME");
    let passkey_origin_env: Option<String> = env_var_clean("PASSKEY_ORIGIN");
    let passkey_related_origins_env: Option<String> = env_var_clean("PASSKEY_RELATED_ORIGINS");
    let passkey_related_origins_env_entries: Vec<String> = passkey_related_origins_env
        .as_deref()
        .map(parse_origins_csv)
        .unwrap_or_default();
    let passkey_related_origins_config_entries: Vec<String> =
        normalize_origins(&auth_cfg.passkey.related_origins);
    let smtp_host_env: Option<String> = env_var_clean("SMTP_HOST");
    let smtp_from_address_env: Option<String> = env_var_clean("SMTP_FROM_ADDRESS");

    json!({
        "config_file_loaded": file_config.is_some(),
        "plugins": {
            "email_password": { "enabled": true, "signup_enabled": true },
            "session_management": { "enabled": true },
            "password_management": { "enabled": true, "require_current_password": true },
            "account_management": { "enabled": true },
            "email_verification": { "enabled": true },
            "api_keys": { "enabled": true, "enable_session_for_api_keys": true },
            "oauth": { "enabled": true, "providers": ["google", "apple", "microsoft", "github", "cloudflare"] },
            "two_factor": { "enabled": true },
            "organization": { "enabled": true },
            "passkey": { "enabled": true },
            "admin": { "enabled": true },
            "user_management": { "enabled": true, "change_email_enabled": true, "delete_user_enabled": true }
        },
        "auth_runtime": {
            "base_url": {
                "configured_by_env": base_url_env.is_some(),
                "configured_by_config": auth_cfg.base_url.as_deref().is_some_and(|value| !value.trim().is_empty()),
                "effective_source": if base_url_env.is_some() { "env" } else if auth_cfg.base_url.as_deref().is_some_and(|value| !value.trim().is_empty()) { "config" } else { "default" }
            },
            "auth_base_path": {
                "configured_by_env": auth_base_path_env.is_some(),
                "configured_by_config": auth_cfg.base_path.as_deref().is_some_and(|value| !value.trim().is_empty()),
                "effective_source": if auth_base_path_env.is_some() { "env" } else if auth_cfg.base_path.as_deref().is_some_and(|value| !value.trim().is_empty()) { "config" } else { "default" }
            },
            "public_auth_base_path": {
                "configured_by_env": public_auth_base_path_env.is_some(),
                "configured_by_config": auth_cfg.public_base_path.as_deref().is_some_and(|value| !value.trim().is_empty()),
                "effective_source": if public_auth_base_path_env.is_some() { "env" } else if auth_cfg.public_base_path.as_deref().is_some_and(|value| !value.trim().is_empty()) { "config" } else { "auth_base_path" }
            },
            "trusted_origins": {
                "configured_by_env": trusted_origins_env.is_some(),
                "configured_by_config": !auth_cfg.trusted_origins.is_empty(),
                "env_entry_count": trusted_origins_env_entries.len(),
                "config_entry_count": auth_cfg.trusted_origins.len()
            }
        },
        "session_runtime": {
            "expires_in_seconds_configured": session_cfg.expires_in_seconds.is_some(),
            "update_age_seconds_configured": session_cfg.update_age_seconds.is_some(),
            "fresh_age_seconds_configured": session_cfg.fresh_age_seconds.is_some(),
            "disable_session_refresh_enabled": session_cfg.disable_session_refresh.unwrap_or(false),
            "cookie": {
                "name_configured": session_cfg.cookie_name.as_deref().is_some_and(|value| !value.trim().is_empty()),
                "domain_configured": cookie_domain_configured_in_yaml || cookie_domain_configured_in_env,
                "domain_configured_by_env": cookie_domain_configured_in_env,
                "domain_configured_by_config": cookie_domain_configured_in_yaml,
                "secure_configured": session_cfg.cookie_secure.is_some(),
                "http_only_configured": session_cfg.cookie_http_only.is_some(),
                "same_site_configured": session_cfg.cookie_same_site.as_deref().is_some_and(|value| !value.trim().is_empty())
            },
            "trusted_proxy_headers_count": session_cfg.trusted_proxy_headers.len(),
            "ip_tracking": {
                "enabled": !session_cfg.ip.disable_tracking.unwrap_or(false),
                "disable_tracking_override": session_cfg.ip.disable_tracking,
                "header_count": session_cfg.ip.headers.len()
            },
            "intelligence": {
                "enabled": session_cfg.intelligence.enabled.unwrap_or(false),
                "detect_country_change_enabled": session_cfg.intelligence.detect_country_change.unwrap_or(false),
                "geoip": {
                    "enabled": geoip_cfg.enabled.unwrap_or(false),
                    "provider_configured": geoip_cfg.provider.as_deref().is_some_and(|value| !value.trim().is_empty()),
                    "base_url_configured": geoip_cfg.base_url.as_deref().is_some_and(|value| !value.trim().is_empty()),
                    "api_key_configured": geoip_api_key_inline || geoip_api_key_from_env,
                    "api_key_source": if geoip_api_key_inline { "config_inline" } else if geoip_api_key_from_env { "env_reference" } else { "none" },
                    "api_key_header_configured": geoip_cfg.api_key_header.as_deref().is_some_and(|value| !value.trim().is_empty()),
                    "timeout_ms_configured": geoip_cfg.timeout_ms.is_some(),
                    "cache_ttl_seconds_configured": geoip_cfg.cache_ttl_seconds.is_some()
                }
            }
        },
        "observability": {
            "metrics": {
                "enabled": observability_cfg.metrics.enabled,
                "path": normalize_route_path(&observability_cfg.metrics.path, DEFAULT_METRICS_PATH),
                "dashboard_enabled": observability_cfg.metrics.dashboard_enabled,
                "dashboard_path": normalize_route_path(&observability_cfg.metrics.dashboard_path, DEFAULT_METRICS_DASHBOARD_PATH),
                "prefix_configured": observability_cfg.metrics.prefix.as_deref().is_some_and(|value| !value.trim().is_empty())
            },
            "loki": {
                "enabled": observability_cfg.loki.enabled,
                "url_configured": observability_cfg.loki.url.as_deref().is_some_and(|value| !value.trim().is_empty()),
                "tenant_id_configured": observability_cfg.loki.tenant_id.as_deref().is_some_and(|value| !value.trim().is_empty()),
                "label_count": observability_cfg.loki.labels.len()
            },
            "logging": {
                "format": if observability_cfg.logging.format.trim().is_empty() {
                    "pretty".to_string()
                } else {
                    observability_cfg.logging.format.trim().to_string()
                },
                "wide_events": observability_cfg.logging.wide_events,
                "environment": if observability_cfg.logging.environment.trim().is_empty() {
                    "local".to_string()
                } else {
                    observability_cfg.logging.environment.trim().to_string()
                },
                "region": observability_cfg.logging.region.clone(),
                "deployment_id_configured": observability_cfg.logging.deployment_id.as_deref().is_some_and(|value| !value.trim().is_empty()),
                "instance_id_configured": observability_cfg.logging.instance_id.as_deref().is_some_and(|value| !value.trim().is_empty()),
                "commit_hash_configured": observability_cfg.logging.commit_hash.as_deref().is_some_and(|value| !value.trim().is_empty())
            }
        },
        "schema_detection": {
            "enabled": schema_detection_cfg.enabled,
            "effective_auth_schema": schema_detection_auth_schema,
            "write_mapping_file": schema_detection_cfg.write_mapping_file,
            "output_path": schema_detection_cfg.output_path,
            "min_column_match_ratio": schema_detection_cfg.min_column_match_ratio,
            "table_substitution_entry_count": schema_detection_cfg.table_substitutions.len()
        },
        "multitenancy": {
            "enabled": parsed_multi.is_multi_tenant_enabled(),
            "tenant_header_name": parsed_multi.header_name.as_str(),
            "default_tenant": parsed_multi.default_tenant.to_string(),
            "configured_tenant_count": parsed_multi.tenants.len(),
            "fallback_header_candidates": FALLBACK_TENANT_HEADERS
        },
        "passkey_runtime": {
            "enabled": true,
            "rp_id_configured_by_env": passkey_rp_id_env.is_some(),
            "rp_name_configured_by_env": passkey_rp_name_env.is_some(),
            "origin_configured_by_env": passkey_origin_env.is_some(),
            "related_origins_configured_by_env": passkey_related_origins_env.is_some(),
            "related_origins_env_entry_count": passkey_related_origins_env_entries.len(),
            "related_origins_config_entry_count": passkey_related_origins_config_entries.len(),
            "allow_insecure_unverified_assertion": parse_bool_env("PASSKEY_ALLOW_INSECURE")
        },
        "email_delivery": {
            "smtp_host_configured": smtp_host_env.is_some(),
            "smtp_from_address_configured": smtp_from_address_env.is_some(),
            "database_logging_wrapper_enabled": smtp_host_env.is_some()
        }
    })
}

/// Builds schema-debug runtime state aligned with active multitenancy config.
fn build_schema_debug_runtime(
    fallback_database_url: &str,
    file_config: Option<&ServerConfigFile>,
    access_gate: SchemaDebugAccessGate,
    auth_config: Arc<AuthConfig>,
) -> Result<Arc<SchemaDebugRuntime>, DynErr> {
    let schema_detection_auth_schema: Option<String> = schema_detection_auth_schema(file_config)?;
    let raw_multi: MultiTenancyConfig = file_config
        .map(|value| value.multitenancy.clone())
        .unwrap_or_default();
    let parsed_multi: ParsedMultiTenancyConfig =
        ParsedMultiTenancyConfig::from_raw(raw_multi.clone())?;
    let enabled_features: Value = build_schema_debug_enabled_features(
        file_config,
        &parsed_multi,
        schema_detection_auth_schema.as_deref(),
    );
    let resolver: HeaderTenantResolver = HeaderTenantResolver::new(
        parsed_multi.header_name.clone(),
        parsed_multi.default_tenant.clone(),
    );

    let mut tenants: BTreeMap<TenantId, SchemaDebugTenant> = BTreeMap::new();

    if !parsed_multi.is_multi_tenant_enabled() {
        let mut mapping: BTreeMap<String, SchemaDebugMapping> = BTreeMap::new();
        for (logical, mapped) in &parsed_multi.default_schema_mapping {
            mapping.insert(
                logical.clone(),
                SchemaDebugMapping {
                    mapped_table: mapped.clone(),
                    source: "default",
                },
            );
        }

        let default_tenant: TenantId = parsed_multi.default_tenant.clone();
        let configured_auth_schema: String = schema_detection_auth_schema
            .as_deref()
            .unwrap_or(parsed_multi.default_auth_schema.as_str())
            .to_string();
        tenants.insert(
            default_tenant,
            SchemaDebugTenant {
                is_default: true,
                database_source_kind: "fallback_database_url",
                database_source_key: None,
                database_url: fallback_database_url.to_string(),
                database_url_redacted: redact_database_url(fallback_database_url),
                configured_auth_schema,
                configured_search_path_sql: None,
                configured_search_path_schemas: Vec::new(),
                configured_mapping: mapping,
            },
        );
    } else {
        for (tenant_id, tenant_config) in &parsed_multi.tenants {
            let (database_url, source_kind, source_key) = resolve_database_url_for_debug(
                &tenant_config.database_url_source,
                tenant_id,
                fallback_database_url,
            )?;

            let tenant_override_mapping: BTreeMap<String, String> = raw_multi
                .tenants
                .get(tenant_id.as_str())
                .map(|cfg| normalize_mapping_for_debug(cfg.schema_mapping.clone()))
                .unwrap_or_default();

            let mut mapping: BTreeMap<String, SchemaDebugMapping> = BTreeMap::new();
            for (logical, mapped) in &tenant_config.schema_mapping {
                let source: &str = if tenant_override_mapping.contains_key(logical) {
                    "tenant_override"
                } else if parsed_multi.default_schema_mapping.contains_key(logical) {
                    "default"
                } else {
                    "unknown"
                };
                mapping.insert(
                    logical.clone(),
                    SchemaDebugMapping {
                        mapped_table: mapped.clone(),
                        source,
                    },
                );
            }

            let configured_search_path_sql: Option<String> = tenant_config
                .search_path
                .as_ref()
                .map(SearchPath::as_sql_fragment)
                .map(str::to_string);
            let configured_auth_schema: String = schema_detection_auth_schema
                .as_deref()
                .unwrap_or(tenant_config.auth_schema.as_str())
                .to_string();

            tenants.insert(
                tenant_id.clone(),
                SchemaDebugTenant {
                    is_default: tenant_id == &parsed_multi.default_tenant,
                    database_source_kind: source_kind,
                    database_source_key: source_key,
                    database_url_redacted: redact_database_url(&database_url),
                    database_url,
                    configured_auth_schema,
                    configured_search_path_schemas: parse_search_path_schemas(
                        configured_search_path_sql.as_deref(),
                    ),
                    configured_search_path_sql,
                    configured_mapping: mapping,
                },
            );
        }

        if !tenants.contains_key(&parsed_multi.default_tenant) {
            let mut mapping: BTreeMap<String, SchemaDebugMapping> = BTreeMap::new();
            for (logical, mapped) in &parsed_multi.default_schema_mapping {
                mapping.insert(
                    logical.clone(),
                    SchemaDebugMapping {
                        mapped_table: mapped.clone(),
                        source: "default",
                    },
                );
            }

            let configured_auth_schema: String = schema_detection_auth_schema
                .as_deref()
                .unwrap_or(parsed_multi.default_auth_schema.as_str())
                .to_string();
            tenants.insert(
                parsed_multi.default_tenant.clone(),
                SchemaDebugTenant {
                    is_default: true,
                    database_source_kind: "fallback_database_url",
                    database_source_key: None,
                    database_url: fallback_database_url.to_string(),
                    database_url_redacted: redact_database_url(fallback_database_url),
                    configured_auth_schema,
                    configured_search_path_sql: None,
                    configured_search_path_schemas: Vec::new(),
                    configured_mapping: mapping,
                },
            );
        }
    }

    Ok(Arc::new(SchemaDebugRuntime {
        multitenancy_enabled: parsed_multi.is_multi_tenant_enabled(),
        header_name: parsed_multi.header_name.as_str().to_string(),
        default_tenant: parsed_multi.default_tenant.clone(),
        resolver,
        tenants,
        access_gate,
        auth_config,
        enabled_features,
    }))
}

/// Resolves tenant debug metadata from request headers.
fn resolve_schema_debug_request_context(
    runtime: &SchemaDebugRuntime,
    headers: &HeaderMap,
) -> SchemaDebugRequestContext {
    let configured_header_value: Option<String> =
        header_value_trimmed(headers, runtime.header_name.as_str());
    let mut fallback_header_values: BTreeMap<String, Option<String>> = BTreeMap::new();
    for name in FALLBACK_TENANT_HEADERS {
        fallback_header_values.insert(name.to_string(), header_value_trimmed(headers, name));
    }

    match runtime
        .resolver
        .resolve_tenant_id(&AxumHeaderLookup { headers })
    {
        Ok(resolved_tenant) => {
            let resolution_status: &str = if runtime.tenants.contains_key(&resolved_tenant) {
                "ok"
            } else {
                "unknown_tenant"
            };

            SchemaDebugRequestContext {
                configured_header_name: runtime.header_name.clone(),
                configured_header_value,
                fallback_header_values,
                resolved_tenant: Some(resolved_tenant),
                resolution_status,
                resolution_error: None,
            }
        }
        Err(error) => SchemaDebugRequestContext {
            configured_header_name: runtime.header_name.clone(),
            configured_header_value,
            fallback_header_values,
            resolved_tenant: None,
            resolution_status: "invalid_header",
            resolution_error: Some(error.to_string()),
        },
    }
}

/// Queries live schema/table visibility for the selected tenant.
async fn query_schema_debug_probe(
    database_url: &str,
    auth_schema: &str,
    search_path_sql: Option<&str>,
    table_names: &[String],
) -> Result<SchemaDebugDatabaseProbe, String> {
    let mut pool_options: sqlx::pool::PoolOptions<sqlx::Postgres> = PgPoolOptions::new()
        .max_connections(1)
        .min_connections(0)
        .acquire_timeout(Duration::from_secs(4));

    let effective_search_path: String =
        build_effective_search_path_from_sql(auth_schema, search_path_sql);
    let set_sql: String = format!("SET search_path TO {effective_search_path}");
    pool_options = pool_options.after_connect(move |conn, _meta| {
        let set_sql = set_sql.clone();
        Box::pin(async move {
            sqlx::query(&set_sql).execute(conn).await?;
            Ok(())
        })
    });

    let connect_options: PgConnectOptions = parse_pg_connect_options(database_url)?;
    let pool: sqlx::Pool<sqlx::Postgres> = pool_options
        .connect_with(connect_options)
        .await
        .map_err(|error| format!("failed to connect to PostgreSQL: {error}"))?;

    let row: sqlx::postgres::PgRow = sqlx::query(
        "SELECT \
            current_schema()::text AS current_schema, \
            current_schemas(false)::text[] AS effective_search_path, \
            current_setting('search_path')::text AS search_path_setting",
    )
    .fetch_one(&pool)
    .await
    .map_err(|error| format!("failed to query schema state: {error}"))?;

    let current_schema: Option<String> = row
        .try_get::<Option<String>, _>("current_schema")
        .unwrap_or(None);
    let effective_search_path: Vec<String> = row
        .try_get::<Option<Vec<String>>, _>("effective_search_path")
        .unwrap_or(None)
        .unwrap_or_default();
    let search_path_setting: Option<String> = row
        .try_get::<Option<String>, _>("search_path_setting")
        .unwrap_or(None);

    let mut table_schemas: BTreeMap<String, Vec<String>> = BTreeMap::new();
    if !table_names.is_empty() {
        let rows = sqlx::query(
            "SELECT table_schema::text AS table_schema, table_name::text AS table_name \
             FROM information_schema.tables \
             WHERE table_type = 'BASE TABLE' \
               AND table_name = ANY($1::text[]) \
             ORDER BY table_name ASC, table_schema ASC",
        )
        .bind(table_names)
        .fetch_all(&pool)
        .await
        .map_err(|error| format!("failed to query table metadata: {error}"))?;

        for row in rows {
            let table_schema: String = row
                .try_get("table_schema")
                .map_err(|error| format!("failed to decode table_schema: {error}"))?;
            let table_name: String = row
                .try_get("table_name")
                .map_err(|error| format!("failed to decode table_name: {error}"))?;
            let entry = table_schemas.entry(table_name).or_default();
            if !entry.iter().any(|value| value == &table_schema) {
                entry.push(table_schema);
            }
        }
    }

    pool.close().await;

    Ok(SchemaDebugDatabaseProbe {
        current_schema,
        effective_search_path,
        search_path_setting,
        table_schemas,
    })
}

/// Returns schemas that are both present and part of the effective search path.
fn schemas_in_search_path(schemas: &[String], effective_search_path: &[String]) -> Vec<String> {
    schemas
        .iter()
        .filter(|schema| {
            effective_search_path
                .iter()
                .any(|effective| effective == *schema)
        })
        .cloned()
        .collect()
}

/// Returns the expected core column set for a logical auth table.
fn expected_columns_for_table(logical_table: &str) -> &'static [&'static str] {
    match logical_table {
        "users" => EXPECTED_COLUMNS_USERS,
        "sessions" => EXPECTED_COLUMNS_SESSIONS,
        "accounts" => EXPECTED_COLUMNS_ACCOUNTS,
        "organization" => EXPECTED_COLUMNS_ORGANIZATION,
        "member" => EXPECTED_COLUMNS_MEMBER,
        "invitation" => EXPECTED_COLUMNS_INVITATION,
        "verifications" => EXPECTED_COLUMNS_VERIFICATIONS,
        "two_factor" => EXPECTED_COLUMNS_TWO_FACTOR,
        "api_keys" => EXPECTED_COLUMNS_API_KEYS,
        "passkeys" => EXPECTED_COLUMNS_PASSKEYS,
        "emails" => EXPECTED_COLUMNS_EMAILS,
        "email_send_failures" => EXPECTED_COLUMNS_EMAIL_SEND_FAILURES,
        "email_templates" => EXPECTED_COLUMNS_EMAIL_TEMPLATES,
        "email_event_types" => EXPECTED_COLUMNS_EMAIL_EVENT_TYPES,
        "admin_audit_logs" => EXPECTED_COLUMNS_ADMIN_AUDIT_LOGS,
        "athena_clients" => EXPECTED_COLUMNS_ATHENA_CLIENTS,
        "oauth_clients" => EXPECTED_COLUMNS_OAUTH_CLIENTS,
        "ip_profiles" => EXPECTED_COLUMNS_IP_PROFILES,
        "session_ip_profiles" => EXPECTED_COLUMNS_SESSION_IP_PROFILES,
        _ => &[],
    }
}

/// Returns whether the table naming looks plural by suffix.
fn is_plural_table_name(table_name: &str) -> bool {
    table_name.ends_with('s')
}

/// Returns a best-effort singular/plural counterpart for debug lookup comparisons.
fn secondary_table_name_variant(table_name: &str) -> Option<String> {
    let normalized: &str = table_name.trim();
    if normalized.is_empty() {
        return None;
    }

    if is_plural_table_name(normalized) {
        if normalized.len() > 1 {
            Some(normalized[..normalized.len() - 1].to_string())
        } else {
            None
        }
    } else {
        Some(format!("{normalized}s"))
    }
}

/// Returns primary + secondary table-name variants (deduplicated, ordered).
fn table_name_variants_for_debug(table_name: &str) -> Vec<String> {
    let normalized: &str = table_name.trim();
    if normalized.is_empty() {
        return Vec::new();
    }

    let mut variants: Vec<String> = vec![normalized.to_string()];
    if let Some(secondary) = secondary_table_name_variant(normalized)
        && secondary != normalized
    {
        variants.push(secondary);
    }
    variants
}

/// Builds schema-qualified lookup candidates for each table variant.
fn build_lookup_candidates_for_tables(
    schemas: &[String],
    table_variants: &[String],
) -> Vec<String> {
    let mut candidates: Vec<String> = Vec::new();
    for table in table_variants {
        for schema in schemas {
            let candidate: String = format!("{schema}.{table}");
            if !candidates.iter().any(|existing| existing == &candidate) {
                candidates.push(candidate);
            }
        }
    }
    candidates
}

/// Classifies a table naming style for quick format diagnostics.
fn classify_table_name_format(table_name: &str) -> &'static str {
    let has_underscore: bool = table_name.contains('_');
    let has_uppercase: bool = table_name.chars().any(|ch| ch.is_ascii_uppercase());
    let plural: bool = is_plural_table_name(table_name);

    match (has_underscore, has_uppercase, plural) {
        (true, false, true) => "snake_case_plural",
        (true, false, false) => "snake_case_singular",
        (false, false, true) => "lowercase_plural",
        (false, false, false) => "lowercase_singular",
        (true, true, _) => "mixed_case_with_underscores",
        (false, true, _) => "camel_or_pascal_case",
    }
}

/// Builds a ranked candidate list (expected table first, then configured fallbacks).
fn ranked_candidate_tables(logical_table: &str, settings: &SchemaDetectionConfig) -> Vec<String> {
    let mut candidates: Vec<String> = Vec::new();
    candidates.push(logical_table.to_string());

    if let Some(configured) = settings.table_substitutions.get(logical_table) {
        for entry in configured {
            let normalized: &str = entry.trim();
            if normalized.is_empty() {
                continue;
            }
            if !candidates.iter().any(|value| value == normalized) {
                candidates.push(normalized.to_string());
            }
        }
    }

    candidates
}

/// Selects the best schema variant for a table using effective search-path order.
fn best_schema_variant_for_table(
    snapshot: &SchemaCatalogSnapshot,
    table_name: &str,
) -> Option<(String, Vec<String>)> {
    let by_schema: &BTreeMap<String, Vec<String>> = snapshot.columns_by_table.get(table_name)?;
    if by_schema.is_empty() {
        return None;
    }

    for schema in &snapshot.effective_search_path {
        if let Some(columns) = by_schema.get(schema) {
            return Some((schema.clone(), columns.clone()));
        }
    }

    by_schema
        .iter()
        .next()
        .map(|(schema, columns)| (schema.clone(), columns.clone()))
}

/// Computes expected-column match diagnostics for a candidate table.
fn evaluate_column_match(
    expected_columns: &[&str],
    candidate_columns: &[String],
) -> (Vec<String>, Vec<String>, Vec<String>, f32) {
    let expected_lower: Vec<String> = expected_columns
        .iter()
        .map(|value| value.to_string())
        .collect();
    let candidate_set: std::collections::BTreeSet<String> = candidate_columns
        .iter()
        .map(|value| value.to_ascii_lowercase())
        .collect();

    let mut matched: Vec<String> = Vec::new();
    let mut missing: Vec<String> = Vec::new();
    for expected in expected_lower {
        if candidate_set.contains(expected.as_str()) {
            matched.push(expected);
        } else {
            missing.push(expected);
        }
    }

    let expected_set: std::collections::BTreeSet<String> = expected_columns
        .iter()
        .map(|value| value.to_ascii_lowercase())
        .collect();
    let extras: Vec<String> = candidate_columns
        .iter()
        .map(|value| value.to_ascii_lowercase())
        .filter(|value| !expected_set.contains(value.as_str()))
        .collect();

    let ratio = if expected_columns.is_empty() {
        0.0
    } else {
        matched.len() as f32 / expected_columns.len() as f32
    };

    (matched, missing, extras, ratio)
}

/// Reads all table/column metadata from schemas visible to the current connection.
async fn query_schema_catalog_snapshot(
    database_url: &str,
    auth_schema: &str,
    search_path_sql: Option<&str>,
) -> Result<SchemaCatalogSnapshot, String> {
    let mut pool_options = PgPoolOptions::new()
        .max_connections(1)
        .min_connections(0)
        .acquire_timeout(Duration::from_secs(6));

    let effective_search_path: String =
        build_effective_search_path_from_sql(auth_schema, search_path_sql);
    let set_sql: String = format!("SET search_path TO {effective_search_path}");
    pool_options = pool_options.after_connect(move |conn, _meta| {
        let set_sql = set_sql.clone();
        Box::pin(async move {
            sqlx::query(&set_sql).execute(conn).await?;
            Ok(())
        })
    });

    let connect_options: PgConnectOptions = parse_pg_connect_options(database_url)?;
    let pool: sqlx::Pool<sqlx::Postgres> = pool_options
        .connect_with(connect_options)
        .await
        .map_err(|error| format!("failed to connect to PostgreSQL: {error}"))?;

    let state_row: sqlx::postgres::PgRow = sqlx::query(
        "SELECT \
            current_schema()::text AS current_schema, \
            current_schemas(false)::text[] AS effective_search_path, \
            current_setting('search_path')::text AS search_path_setting",
    )
    .fetch_one(&pool)
    .await
    .map_err(|error| format!("failed to query schema state: {error}"))?;

    let current_schema: Option<String> = state_row
        .try_get::<Option<String>, _>("current_schema")
        .unwrap_or(None);
    let effective_search_path: Vec<String> = state_row
        .try_get::<Option<Vec<String>>, _>("effective_search_path")
        .unwrap_or(None)
        .unwrap_or_default();
    let search_path_setting: Option<String> = state_row
        .try_get::<Option<String>, _>("search_path_setting")
        .unwrap_or(None);

    let rows: Vec<sqlx::postgres::PgRow> = sqlx::query(
        "SELECT table_schema::text AS table_schema, table_name::text AS table_name, column_name::text AS column_name \
         FROM information_schema.columns \
         WHERE table_schema = ANY(current_schemas(false)) \
         ORDER BY table_schema ASC, table_name ASC, ordinal_position ASC",
    )
    .fetch_all(&pool)
    .await
    .map_err(|error| format!("failed to query table column catalog: {error}"))?;

    let mut columns_by_table: BTreeMap<String, BTreeMap<String, Vec<String>>> = BTreeMap::new();
    for row in rows {
        let table_schema: String = row
            .try_get("table_schema")
            .map_err(|error| format!("failed to decode table_schema: {error}"))?;
        let table_name: String = row
            .try_get("table_name")
            .map_err(|error| format!("failed to decode table_name: {error}"))?;
        let column_name: String = row
            .try_get("column_name")
            .map_err(|error| format!("failed to decode column_name: {error}"))?;

        let schema_entry = columns_by_table.entry(table_name).or_default();
        let columns = schema_entry.entry(table_schema).or_default();
        columns.push(column_name.to_ascii_lowercase());
    }

    pool.close().await;

    Ok(SchemaCatalogSnapshot {
        current_schema,
        effective_search_path,
        search_path_setting,
        columns_by_table,
    })
}

/// Runs schema auto-detection and optionally writes a generated mapping file.
async fn run_schema_autodetection(
    config_file: Option<&ServerConfigFile>,
    runtime: &SchemaDebugRuntime,
) -> Result<(), DynErr> {
    let Some(config_file) = config_file else {
        return Ok(());
    };
    let settings: &SchemaDetectionConfig = &config_file.schema_detection;
    if !settings.enabled {
        return Ok(());
    }

    let min_ratio: f32 = settings.min_column_match_ratio.clamp(0.0, 1.0);
    let mut tenant_reports: Vec<Value> = Vec::new();
    let mut substitution_total: usize = 0;

    for (tenant_id, tenant) in &runtime.tenants {
        match query_schema_catalog_snapshot(
            tenant.database_url.as_str(),
            tenant.configured_auth_schema.as_str(),
            tenant.configured_search_path_sql.as_deref(),
        )
        .await
        {
            Ok(snapshot) => {
                let mut substitutions: BTreeMap<String, String> = BTreeMap::new();
                let mut table_evaluations: Vec<Value> = Vec::new();

                for logical_table in LOGICAL_TABLE_NAMES {
                    let expected_columns: &[&str] = expected_columns_for_table(logical_table);
                    let candidates: Vec<String> = ranked_candidate_tables(logical_table, settings);
                    let mut candidate_reports: Vec<Value> = Vec::new();
                    let mut resolved: Option<SchemaDetectionResolvedCandidate> = None;

                    for (rank, candidate) in candidates.iter().enumerate() {
                        if let Some((schema_name, candidate_columns)) =
                            best_schema_variant_for_table(&snapshot, candidate)
                        {
                            let (matched_columns, missing_columns, extra_columns, ratio) =
                                evaluate_column_match(expected_columns, &candidate_columns);
                            let accepted: bool = ratio >= min_ratio;

                            candidate_reports.push(json!({
                                "rank": rank + 1,
                                "table": candidate,
                                "schema": schema_name,
                                "exists": true,
                                "expected_columns": expected_columns,
                                "candidate_columns": candidate_columns,
                                "matched_columns": matched_columns,
                                "missing_columns": missing_columns,
                                "extra_columns": extra_columns,
                                "column_match_ratio": ratio,
                                "accepted": accepted,
                            }));

                            if accepted {
                                resolved = Some((
                                    candidate.clone(),
                                    schema_name,
                                    rank + 1,
                                    matched_columns,
                                    missing_columns,
                                    ratio,
                                ));
                                break;
                            }
                        } else {
                            candidate_reports.push(json!({
                                "rank": rank + 1,
                                "table": candidate,
                                "exists": false,
                                "accepted": false,
                            }));
                        }
                    }

                    let (
                        resolution,
                        resolved_table,
                        resolved_schema,
                        resolved_rank,
                        matched_columns,
                        missing_columns,
                        column_match_ratio,
                    ) = if let Some((table, schema, rank, matched, missing, ratio)) = resolved {
                        let resolution: &str = if table == logical_table {
                            "exact"
                        } else {
                            substitutions.insert(logical_table.to_string(), table.clone());
                            substitution_total += 1;
                            "substituted"
                        };
                        (
                            resolution,
                            Some(table),
                            Some(schema),
                            Some(rank),
                            matched,
                            missing,
                            Some(ratio),
                        )
                    } else {
                        (
                            "missing_or_column_mismatch",
                            None,
                            None,
                            None,
                            Vec::new(),
                            expected_columns
                                .iter()
                                .map(|value| value.to_string())
                                .collect(),
                            None,
                        )
                    };

                    table_evaluations.push(json!({
                        "logical_table": logical_table,
                        "expected_table": logical_table,
                        "expected_table_format": classify_table_name_format(logical_table),
                        "resolution": resolution,
                        "resolved_table": resolved_table,
                        "resolved_table_format": resolved_table.as_ref().map(|value| classify_table_name_format(value)),
                        "resolved_schema": resolved_schema,
                        "resolved_rank": resolved_rank,
                        "column_match_ratio": column_match_ratio,
                        "matched_columns": matched_columns,
                        "missing_columns": missing_columns,
                        "candidates_tried": candidate_reports,
                    }));
                }

                tenant_reports.push(json!({
                    "tenant_id": tenant_id.to_string(),
                    "database_source": {
                        "kind": tenant.database_source_kind,
                        "key": tenant.database_source_key,
                        "resolved_url_redacted": tenant.database_url_redacted,
                    },
                    "detected_schema": {
                        "current_schema": snapshot.current_schema,
                        "effective_search_path": snapshot.effective_search_path,
                        "search_path_setting": snapshot.search_path_setting,
                    },
                    "suggested_schema_mapping": substitutions,
                    "table_evaluations": table_evaluations,
                }));
            }
            Err(error) => {
                tenant_reports.push(json!({
                    "tenant_id": tenant_id.to_string(),
                    "status": "probe_error",
                    "error": error,
                }));
            }
        }
    }

    let report: Value = json!({
        "generated_at": chrono::Utc::now().to_rfc3339(),
        "schema_detection": {
            "min_column_match_ratio": min_ratio,
            "write_mapping_file": settings.write_mapping_file,
            "output_path": settings.output_path,
        },
        "runtime_context": {
            "tenant_header_name": runtime.header_name,
            "default_tenant": runtime.default_tenant.to_string(),
            "known_tenant_count": runtime.tenants.len(),
        },
        "summary": {
            "tenant_count": tenant_reports.len(),
            "substitution_count": substitution_total,
        },
        "tenants": tenant_reports,
    });

    if settings.write_mapping_file && substitution_total > 0 {
        let output_path: &Path = std::path::Path::new(settings.output_path.as_str());
        if let Some(parent) = output_path.parent()
            && !parent.as_os_str().is_empty()
        {
            fs::create_dir_all(parent).map_err(|error| {
                format!(
                    "failed to create schema-detection output directory '{}': {error}",
                    parent.to_string_lossy()
                )
            })?;
        }

        let yaml: String = serde_yaml::to_string(&report)
            .map_err(|error| format!("failed to serialize schema-detection report: {error}"))?;
        fs::write(output_path, yaml).map_err(|error| {
            format!(
                "failed to write schema-detection output file '{}': {error}",
                output_path.to_string_lossy()
            )
        })?;

        tracing::info!(
            output_path = %output_path.to_string_lossy(),
            substitution_count = substitution_total,
            "schema auto-detection generated mapping file"
        );
    } else {
        tracing::info!(
            substitution_count = substitution_total,
            write_mapping_file = settings.write_mapping_file,
            "schema auto-detection completed"
        );
    }

    Ok(())
}

impl RuntimeObservability {
    fn disabled() -> Self {
        Self {
            metrics: None,
            metrics_path: None,
            metrics_dashboard_path: None,
            loki_enabled: false,
            logging: LoggingRuntime::from_config(None)
                .expect("default logging runtime should always parse"),
        }
    }
}

/// Resolves the effective config.yaml path from env or default.
fn resolve_server_config_path() -> String {
    env::var("ATHENA_AUTH_CONFIG_FILE")
        .ok()
        .map(|path| path.trim().to_string())
        .filter(|path| !path.is_empty())
        .unwrap_or_else(|| DEFAULT_CONFIG_FILE_PATH.to_string())
}

/// Captures config-file diagnostics for guarded debug-route expansion.
fn capture_schema_debug_config_snapshot(
    config_file: Option<&ServerConfigFile>,
) -> SchemaDebugConfigSnapshot {
    let path: String = resolve_server_config_path();
    let path_ref: &Path = Path::new(path.as_str());
    let resolved_runtime_config: Option<String> = config_file.map(|cfg| format!("{cfg:#?}"));

    if !path_ref.exists() {
        return SchemaDebugConfigSnapshot {
            path,
            exists: false,
            read_status: "missing",
            read_error: None,
            raw_contents: None,
            resolved_runtime_config,
        };
    }

    match fs::read_to_string(path_ref) {
        Ok(raw_contents) => SchemaDebugConfigSnapshot {
            path,
            exists: true,
            read_status: "ok",
            read_error: None,
            raw_contents: Some(raw_contents),
            resolved_runtime_config,
        },
        Err(error) => SchemaDebugConfigSnapshot {
            path,
            exists: true,
            read_status: "read_error",
            read_error: Some(error.to_string()),
            raw_contents: None,
            resolved_runtime_config,
        },
    }
}

/// Builds gate metadata used to unlock privileged schema debug payloads.
fn build_schema_debug_access_gate(config_file: Option<&ServerConfigFile>) -> SchemaDebugAccessGate {
    SchemaDebugAccessGate {
        jwt_secret: env_var_clean("JWT_SECRET"),
        config_snapshot: capture_schema_debug_config_snapshot(config_file),
    }
}

/// Evaluates whether a debug request can access privileged diagnostics.
fn evaluate_schema_debug_access(
    query_secret: Option<&str>,
    expected_secret: Option<&str>,
) -> SchemaDebugAccessEvaluation {
    let normalized_query: Option<String> = normalize_debug_secret(query_secret);
    let query_param_present: bool = normalized_query.is_some();
    let expected: Option<String> = normalize_debug_secret(expected_secret);
    if expected.is_none() {
        return SchemaDebugAccessEvaluation {
            privileged: false,
            reason: "server_jwt_secret_unavailable",
            query_param_present,
        };
    }
    let Some(provided_secret) = normalized_query.as_deref() else {
        return SchemaDebugAccessEvaluation {
            privileged: false,
            reason: "jwt_secret_query_missing",
            query_param_present,
        };
    };
    if expected
        .as_deref()
        .is_some_and(|expected_secret| secret_matches_expected(provided_secret, expected_secret))
    {
        return SchemaDebugAccessEvaluation {
            privileged: true,
            reason: "jwt_secret_query_matches_env",
            query_param_present: true,
        };
    }
    SchemaDebugAccessEvaluation {
        privileged: false,
        reason: "jwt_secret_query_mismatch",
        query_param_present: true,
    }
}

/// Normalizes a JWT-secret debug input from env or query.
fn normalize_debug_secret(value: Option<&str>) -> Option<String> {
    value
        .map(strip_wrapping_quotes)
        .filter(|candidate| !candidate.is_empty())
}

/// Compares expected secret against query input with compatibility normalization.
fn secret_matches_expected(provided: &str, expected: &str) -> bool {
    if provided == expected {
        return true;
    }
    // Query parsing can decode '+' as space for x-www-form-urlencoded values.
    if provided.contains(' ') && expected.contains('+') {
        return provided.replace(' ', "+") == expected;
    }
    false
}

/// Computes a stable SHA-256 hex digest.
fn sha256_hex(value: &str) -> String {
    let mut hasher = Sha256::new();
    hasher.update(value.as_bytes());
    format!("{:x}", hasher.finalize())
}

/// Builds a normalized environment snapshot value.
fn env_debug_snapshot_value(key: &str, redact: bool) -> Value {
    match env_var_clean(key) {
        Some(value) => {
            let normalized = if redact {
                "<redacted>".to_string()
            } else {
                value
            };
            json!({
                "present": true,
                "value": normalized,
            })
        }
        None => json!({
            "present": false,
            "value": null,
        }),
    }
}

/// Builds a normalized environment snapshot from the first matching key.
fn env_debug_snapshot_first(keys: &[&str], redact: bool) -> Value {
    for key in keys {
        if let Some(value) = env_var_clean(key) {
            let normalized = if redact {
                "<redacted>".to_string()
            } else {
                value
            };
            return json!({
                "present": true,
                "resolved_from": key,
                "value": normalized,
            });
        }
    }
    json!({
        "present": false,
        "resolved_from": null,
        "value": null,
    })
}

fn parse_oauth_callback_urls_from_env(raw: &str) -> Vec<String> {
    raw.split(',')
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
        .collect()
}

fn oauth_callback_urls_match(requested: &str, allowed: &str) -> bool {
    if requested == allowed {
        return true;
    }
    requested.trim_end_matches('/') == allowed.trim_end_matches('/')
}

fn debug_oauth_default_redirect_uri(auth_config: &AuthConfig, provider_name: &str) -> String {
    let env_key = format!("{}_CALLBACK_URL", provider_name.to_uppercase());
    if let Ok(raw) = std::env::var(&env_key) {
        if let Some(first) = parse_oauth_callback_urls_from_env(raw.as_str())
            .into_iter()
            .next()
        {
            return first;
        }
    }
    auth_config.public_auth_url(&format!("callback/{provider_name}"))
}

fn debug_configured_oauth_callback_urls(
    auth_config: &AuthConfig,
    provider_name: &str,
) -> Vec<String> {
    let mut urls = Vec::new();
    let env_key = format!("{}_CALLBACK_URL", provider_name.to_uppercase());
    if let Ok(raw) = std::env::var(&env_key) {
        urls.extend(parse_oauth_callback_urls_from_env(raw.as_str()));
    }
    urls.push(auth_config.public_auth_url(&format!("callback/{provider_name}")));
    urls
}

fn simulate_oauth_callback_url_resolution(
    auth_config: &AuthConfig,
    provider_name: &str,
    requested: &str,
) -> Value {
    let default_redirect_uri = debug_oauth_default_redirect_uri(auth_config, provider_name);
    let configured_urls = debug_configured_oauth_callback_urls(auth_config, provider_name);

    if configured_urls
        .iter()
        .any(|allowed| oauth_callback_urls_match(requested, allowed))
    {
        return json!({
            "requested": requested,
            "outcome": "oauth_redirect_uri",
            "oauth_redirect_uri": requested,
            "app_callback_url": null,
            "note": "callbackURL matches a configured OAuth redirect URI.",
        });
    }

    let trusted_app_callback =
        requested.starts_with('/') || auth_config.is_origin_trusted(requested);
    if trusted_app_callback {
        return json!({
            "requested": requested,
            "outcome": "trusted_app_callback",
            "oauth_redirect_uri": default_redirect_uri,
            "app_callback_url": requested,
            "note": "Better Auth-style callbackURL treated as post-login redirect; provider redirect_uri stays on Athena callback route.",
        });
    }

    json!({
        "requested": requested,
        "outcome": "rejected",
        "oauth_redirect_uri": null,
        "app_callback_url": null,
        "error": format!("callbackURL is not registered for provider: {provider_name}"),
        "note": "Not in configured OAuth redirect URIs and origin is not trusted.",
    })
}

fn build_schema_debug_oauth_provider_snapshot(
    auth_config: &AuthConfig,
    provider_name: &str,
    client_id_env_keys: &[&str],
    client_secret_env_keys: &[&str],
) -> Value {
    let client_id_snapshot = env_debug_snapshot_first(client_id_env_keys, false);
    let client_secret_snapshot = env_debug_snapshot_first(client_secret_env_keys, true);
    let callback_env_key = format!("{}_CALLBACK_URL", provider_name.to_uppercase());
    let callback_env_raw = env_var_clean(callback_env_key.as_str());
    let configured_callback_urls = debug_configured_oauth_callback_urls(auth_config, provider_name);
    let default_redirect_uri = debug_oauth_default_redirect_uri(auth_config, provider_name);
    let client_id_present = client_id_snapshot
        .get("present")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let client_secret_present = client_secret_snapshot
        .get("present")
        .and_then(Value::as_bool)
        .unwrap_or(false);

    json!({
        "provider": provider_name,
        "credentials": {
            "client_id": client_id_snapshot,
            "client_secret": client_secret_snapshot,
            "ready": client_id_present && client_secret_present,
        },
        "callback": {
            "env_key": callback_env_key,
            "env_present": callback_env_raw.is_some(),
            "env_entries": callback_env_raw
                .as_deref()
                .map(parse_oauth_callback_urls_from_env)
                .unwrap_or_default(),
            "configured_redirect_uris": configured_callback_urls,
            "default_oauth_redirect_uri": default_redirect_uri,
            "provider_callback_route": auth_config.public_auth_url(&format!("callback/{provider_name}")),
        },
    })
}

/// Builds OAuth/social sign-in runtime diagnostics for privileged schema debug output.
fn build_schema_debug_oauth_runtime(runtime: &SchemaDebugRuntime) -> Value {
    let auth_config = runtime.auth_config.as_ref();
    let base_url = auth_config.base_url.clone();
    let public_base_path = auth_config.effective_public_base_path().to_string();
    let trusted_origins = auth_config.trusted_origins.clone();

    let sample_callback_urls = vec![
        format!("{base_url}/session"),
        format!("{base_url}/"),
        auth_config.public_auth_url("callback/cloudflare"),
        "https://evil.example/session".to_string(),
    ];
    let callback_resolution_samples: Vec<Value> = sample_callback_urls
        .iter()
        .map(|requested| {
            simulate_oauth_callback_url_resolution(auth_config, "cloudflare", requested)
        })
        .collect();

    json!({
        "semantics": {
            "callback_url_field": "Better Auth clients send callbackURL as the post-login app redirect (for example /session), not as the provider OAuth redirect_uri.",
            "provider_redirect_uri": "GitHub (and other providers) always receive Athena's /callback/{provider} route unless callbackURL exactly matches a configured OAuth redirect URI.",
            "post_auth_handoff": "After OAuth callback, users with app_callback_url are redirected to app_callback_url?token=...&provider=... with session cookie set.",
        },
        "effective_auth_urls": {
            "base_url": base_url,
            "base_path": auth_config.base_path.clone(),
            "public_base_path": public_base_path,
            "sign_in_social_route": auth_config.public_auth_url("sign-in/social"),
        },
        "trusted_origins": {
            "count": trusted_origins.len(),
            "entries": trusted_origins,
        },
        "providers": {
            "cloudflare": build_schema_debug_oauth_provider_snapshot(
                auth_config,
                "cloudflare",
                &[
                    "CLOUDFLARE_CLIENT_ID",
                    "CLOUDFLARE_OAUTH_CLIENT_ID",
                ],
                &[
                    "CLOUDFLARE_CLIENT_SECRET",
                    "CLOUDFLARE_OAUTH_CLIENT_SECRET",
                    "CLOUDFLARE_OAUTH_CLIENT_PUBLISHER",
                    "cloudflare_oauth_client_publisher",
                ],
            ),
            "github": build_schema_debug_oauth_provider_snapshot(
                auth_config,
                "github",
                &[
                    "GITHUB_CLIENT_ID",
                    "GITHUB_OAUTH_CLIENT_ID",
                    "GITHUB_APP_CLIENT_ID",
                ],
                &[
                    "GITHUB_CLIENT_SECRET",
                    "GITHUB_OAUTH_CLIENT_SECRET",
                    "GITHUB_APP_CLIENT_SECRET",
                ],
            ),
            "google": build_schema_debug_oauth_provider_snapshot(
                auth_config,
                "google",
                &["GOOGLE_CLIENT_ID"],
                &["GOOGLE_CLIENT_SECRET"],
            ),
            "apple": build_schema_debug_oauth_provider_snapshot(
                auth_config,
                "apple",
                &["APPLE_CLIENT_ID"],
                &["APPLE_CLIENT_SECRET"],
            ),
            "microsoft": build_schema_debug_oauth_provider_snapshot(
                auth_config,
                "microsoft",
                &["MICROSOFT_CLIENT_ID"],
                &["MICROSOFT_CLIENT_SECRET", "MICROSOFT_SECRET"],
            ),
        },
        "cloudflare_callback_resolution_samples": callback_resolution_samples,
    })
}

/// Builds an expanded debug payload unlocked by matching `?jwt_secret=`.
fn build_schema_debug_privileged_payload(runtime: &SchemaDebugRuntime) -> Value {
    let jwt_secret_meta: Value = runtime
        .access_gate
        .jwt_secret
        .as_deref()
        .map(|value| {
            json!({
                "configured": true,
                "length": value.len(),
                "sha256": sha256_hex(value),
            })
        })
        .unwrap_or_else(|| {
            json!({
                "configured": false,
            })
        });

    let config_snapshot: &SchemaDebugConfigSnapshot = &runtime.access_gate.config_snapshot;
    let tenant_runtime_details: Vec<Value> = runtime
        .tenants
        .iter()
        .map(|(tenant_id, tenant)| {
            let mapping_entries: Vec<Value> = tenant
                .configured_mapping
                .iter()
                .map(|(logical_table, mapping)| {
                    json!({
                        "logical_table": logical_table,
                        "mapped_table": mapping.mapped_table,
                        "source": mapping.source,
                    })
                })
                .collect();
            json!({
                "tenant_id": tenant_id.to_string(),
                "is_default": tenant.is_default,
                "database_source_kind": tenant.database_source_kind,
                "database_source_key": tenant.database_source_key,
                "database_url": tenant.database_url,
                "database_url_redacted": tenant.database_url_redacted,
                "configured_auth_schema": tenant.configured_auth_schema,
                "configured_search_path_sql": tenant.configured_search_path_sql,
                "configured_search_path_schemas": tenant.configured_search_path_schemas,
                "configured_mapping_entry_count": mapping_entries.len(),
                "configured_mapping_entries": mapping_entries,
            })
        })
        .collect();

    let database_url: Option<String> =
        database_url_from_env().map(|value| redact_database_url(value.as_str()));
    let direct_database_url: Option<String> =
        env_var_clean("DATABASE_URL").map(|value| redact_database_url(value.as_str()));
    let process_cwd: Option<String> = env::current_dir()
        .ok()
        .map(|path| path.to_string_lossy().to_string());
    let process_executable: Option<String> = env::current_exe()
        .ok()
        .map(|path| path.to_string_lossy().to_string());

    json!({
        "authentication": {
            "mode": "query_param_jwt_secret",
            "jwt_secret_env_metadata": jwt_secret_meta,
        },
        "config_file": {
            "path": config_snapshot.path,
            "exists": config_snapshot.exists,
            "read_status": config_snapshot.read_status,
            "read_error": config_snapshot.read_error,
            "raw_contents": config_snapshot.raw_contents,
            "resolved_runtime_config_debug": config_snapshot.resolved_runtime_config,
        },
        "environment": {
            "DATABASE_URL": {
                "present": direct_database_url.is_some(),
                "value": direct_database_url,
            },
            "database_url_resolved": {
                "present": database_url.is_some(),
                "value": database_url,
            },
            "BASE_URL": env_debug_snapshot_value("BASE_URL", false),
            "AUTH_BASE_PATH": env_debug_snapshot_value("AUTH_BASE_PATH", false),
            "PUBLIC_AUTH_BASE_PATH": env_debug_snapshot_value("PUBLIC_AUTH_BASE_PATH", false),
            "HOST": env_debug_snapshot_value("HOST", false),
            "PORT": env_debug_snapshot_value("PORT", false),
            "TRUSTED_ORIGINS": env_debug_snapshot_value("TRUSTED_ORIGINS", false),
            "PASSKEY_RELATED_ORIGINS": env_debug_snapshot_value("PASSKEY_RELATED_ORIGINS", false),
            "GITHUB_CLIENT_ID": env_debug_snapshot_first(
                &[
                    "GITHUB_CLIENT_ID",
                    "GITHUB_OAUTH_CLIENT_ID",
                    "GITHUB_APP_CLIENT_ID",
                ],
                false,
            ),
            "GITHUB_CLIENT_SECRET": env_debug_snapshot_first(
                &[
                    "GITHUB_CLIENT_SECRET",
                    "GITHUB_OAUTH_CLIENT_SECRET",
                    "GITHUB_APP_CLIENT_SECRET",
                ],
                true,
            ),
            "GITHUB_CALLBACK_URL": env_debug_snapshot_value("GITHUB_CALLBACK_URL", false),
            "CLOUDFLARE_CLIENT_ID": env_debug_snapshot_first(
                &[
                    "CLOUDFLARE_CLIENT_ID",
                    "CLOUDFLARE_OAUTH_CLIENT_ID",
                ],
                false,
            ),
            "CLOUDFLARE_CLIENT_SECRET": env_debug_snapshot_first(
                &[
                    "CLOUDFLARE_CLIENT_SECRET",
                    "CLOUDFLARE_OAUTH_CLIENT_SECRET",
                    "CLOUDFLARE_OAUTH_CLIENT_PUBLISHER",
                    "cloudflare_oauth_client_publisher",
                ],
                true,
            ),
            "CLOUDFLARE_CALLBACK_URL": env_debug_snapshot_value("CLOUDFLARE_CALLBACK_URL", false),
            "GOOGLE_CLIENT_ID": env_debug_snapshot_value("GOOGLE_CLIENT_ID", false),
            "GOOGLE_CLIENT_SECRET": env_debug_snapshot_value("GOOGLE_CLIENT_SECRET", true),
            "APPLE_CLIENT_ID": env_debug_snapshot_value("APPLE_CLIENT_ID", false),
            "APPLE_CLIENT_SECRET": env_debug_snapshot_value("APPLE_CLIENT_SECRET", true),
            "MICROSOFT_CLIENT_ID": env_debug_snapshot_value("MICROSOFT_CLIENT_ID", false),
            "MICROSOFT_CLIENT_SECRET": env_debug_snapshot_first(
                &["MICROSOFT_CLIENT_SECRET", "MICROSOFT_SECRET"],
                true,
            ),
            "DATABASE_KEY": env_debug_snapshot_value("DATABASE_KEY", true),
            "ATHENA_AUTH_CONFIG_FILE": env_debug_snapshot_value("ATHENA_AUTH_CONFIG_FILE", false),
            "RUST_LOG": env_debug_snapshot_value("RUST_LOG", false),
            "JWT_SECRET": env_debug_snapshot_value("JWT_SECRET", true),
        },
        "oauth_runtime": build_schema_debug_oauth_runtime(runtime),
        "process": {
            "cwd": process_cwd,
            "executable": process_executable,
            "version": env!("CARGO_PKG_VERSION"),
            "tenant_count": runtime.tenants.len(),
        },
        "tenant_runtime_details": {
            "tenant_count": tenant_runtime_details.len(),
            "entries": tenant_runtime_details,
        },
    })
}

/// Loads optional YAML runtime configuration if `ATHENA_AUTH_CONFIG_FILE` exists.
fn load_server_config_file() -> Result<Option<ServerConfigFile>, DynErr> {
    let config_path = resolve_server_config_path();
    let path = Path::new(&config_path);
    if !path.exists() {
        return Ok(None);
    }

    let raw = fs::read_to_string(path).map_err(|error| {
        format!(
            "failed to read config file '{}': {error}",
            path.to_string_lossy()
        )
    })?;

    if raw.trim().is_empty() {
        return Ok(Some(ServerConfigFile::default()));
    }

    let raw_yaml: serde_yaml::Value = serde_yaml::from_str(&raw).map_err(|error| {
        format!(
            "failed to parse YAML config '{}': {error}",
            path.to_string_lossy()
        )
    })?;

    let resolver: ConfigResolver<fn(&str) -> Option<String>> = ConfigResolver::new(env_var_clean);
    let fields: Vec<ConfigField> = resolver.collect_fields_from_value(&raw_yaml);
    let parsed: ServerConfigFile = resolver.resolve_server_config(&raw).map_err(|error| {
        format!(
            "failed to resolve YAML config '{}' with env overrides: {error}",
            path.to_string_lossy()
        )
    })?;
    tracing::debug!(
        config_field_count = fields.len(),
        config_path = %path.to_string_lossy(),
        "resolved config.yaml fields with env-aware precedence"
    );
    Ok(Some(parsed))
}

/// Builds optional runtime observability state from config and environment.
fn build_observability(
    config_file: Option<&ServerConfigFile>,
    logging: LoggingRuntime,
) -> Result<RuntimeObservability, DynErr> {
    let mut runtime: RuntimeObservability = RuntimeObservability::disabled();
    runtime.logging = logging;

    let Some(config_file) = config_file else {
        tracing::trace!("observability config file missing; runtime observability disabled");
        return Ok(runtime);
    };

    if config_file.observability.metrics.enabled {
        let metrics_config: &MetricsConfig = &config_file.observability.metrics;
        let metrics_path: String = normalize_route_path(&metrics_config.path, DEFAULT_METRICS_PATH);
        let metrics_dashboard_path = metrics_config.dashboard_enabled.then(|| {
            normalize_route_path(
                &metrics_config.dashboard_path,
                DEFAULT_METRICS_DASHBOARD_PATH,
            )
        });
        let metrics: HttpMetrics = HttpMetrics::from_config(ObservabilityHttpMetricsConfig {
            prefix: metrics_config.prefix.clone(),
            service_name: env!("CARGO_PKG_NAME").to_string(),
            service_version: env!("CARGO_PKG_VERSION").to_string(),
        })?;
        runtime.metrics_path = Some(metrics_path);
        runtime.metrics_dashboard_path = metrics_dashboard_path;
        runtime.metrics = Some(Arc::new(metrics));
        tracing::trace!(
            metrics_path = %runtime.metrics_path.as_deref().unwrap_or(DEFAULT_METRICS_PATH),
            metrics_dashboard_path = ?runtime.metrics_dashboard_path,
            "observability metrics enabled"
        );
    }

    runtime.loki_enabled = config_file.observability.loki.enabled;
    tracing::trace!(
        loki_enabled = runtime.loki_enabled,
        "observability config evaluated"
    );
    Ok(runtime)
}

/// Initializes tracing and optional Loki shipping for the server process.
fn init_tracing(
    env_filter: EnvFilter,
    loki_config: Option<&LokiConfig>,
    logging: &LoggingRuntime,
) -> Result<Option<tokio::task::JoinHandle<()>>, DynErr> {
    if let Some(loki) = loki_config.filter(|candidate| candidate.enabled) {
        let endpoint = loki
            .url
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .ok_or(
                "observability.loki.enabled=true requires observability.loki.url in config.yaml",
            )?;
        let endpoint = Url::parse(endpoint)
            .map_err(|error| format!("invalid observability.loki.url '{endpoint}': {error}"))?;

        let mut builder = tracing_loki::builder();
        for (key, value) in &loki.labels {
            builder = builder
                .label(key.clone(), value.as_str())
                .map_err(|error| format!("invalid Loki label '{key}': {error}"))?;
        }
        if let Some(tenant_id) = loki.tenant_id.as_ref().map(|value| value.trim())
            && !tenant_id.is_empty()
        {
            builder = builder
                .http_header("X-Scope-OrgID", tenant_id)
                .map_err(|error| format!("invalid Loki tenant header: {error}"))?;
        }

        let (loki_layer, task) = builder
            .build_url(endpoint)
            .map_err(|error| format!("failed to initialize Loki layer: {error}"))?;

        match logging.format {
            LoggingFormat::Json => {
                tracing_subscriber::registry()
                    .with(env_filter)
                    .with(
                        tracing_subscriber::fmt::layer()
                            .json()
                            .flatten_event(true)
                            .with_current_span(false)
                            .with_span_list(false)
                            .with_target(false)
                            .with_writer(std::io::stdout),
                    )
                    .with(loki_layer)
                    .init();
            }
            LoggingFormat::Pretty => {
                tracing_subscriber::registry()
                    .with(env_filter)
                    .with(
                        tracing_subscriber::fmt::layer()
                            .with_target(false)
                            .with_writer(std::io::stdout),
                    )
                    .with(loki_layer)
                    .init();
            }
        }

        tracing::trace!("tracing initialized with Loki layer");
        return Ok(Some(tokio::spawn(task)));
    }

    match logging.format {
        LoggingFormat::Json => {
            tracing_subscriber::registry()
                .with(env_filter)
                .with(
                    tracing_subscriber::fmt::layer()
                        .json()
                        .flatten_event(true)
                        .with_current_span(false)
                        .with_span_list(false)
                        .with_target(false)
                        .with_writer(std::io::stdout),
                )
                .init();
        }
        LoggingFormat::Pretty => {
            tracing_subscriber::registry()
                .with(env_filter)
                .with(
                    tracing_subscriber::fmt::layer()
                        .with_target(false)
                        .with_writer(std::io::stdout),
                )
                .init();
        }
    }
    tracing::trace!("tracing initialized without Loki layer");
    Ok(None)
}

/// Removes matching wrapping quotes from environment-derived strings.
fn strip_wrapping_quotes(value: &str) -> String {
    let trimmed = value.trim();
    if trimmed.len() >= 2 {
        let first = trimmed.as_bytes()[0] as char;
        let last = trimmed.as_bytes()[trimmed.len() - 1] as char;
        if (first == '"' && last == '"') || (first == '\'' && last == '\'') {
            return trimmed[1..trimmed.len() - 1].to_string();
        }
    }
    trimmed.to_string()
}

fn strip_leading_env_noise(value: &str) -> &str {
    let trimmed = value.trim();
    let without_control_prefix =
        trimmed.trim_start_matches(|ch: char| ch.is_control() || ch == '\u{feff}');

    for scheme in ["postgresql://", "postgres://"] {
        if let Some(index) = without_control_prefix.find(scheme) {
            if index > 0 {
                return &without_control_prefix[index..];
            }
        }
    }

    without_control_prefix
}

/// Reads and trims an environment variable, returning `None` when empty.
fn env_var_clean(key: &str) -> Option<String> {
    env::var(key).ok().map(|raw| {
        let unquoted = strip_wrapping_quotes(&raw);
        let cleaned = strip_leading_env_noise(&unquoted).to_string();
        if cleaned != raw {
            tracing::warn!(env = %key, "normalized environment variable before runtime use");
        }
        cleaned
    })
}

/// Reads an environment variable and returns `None` when it is missing or blank.
fn env_var_non_empty(key: &str) -> Option<String> {
    env_var_clean(key).and_then(|value| {
        let trimmed = value.trim();
        if trimmed.is_empty() {
            None
        } else {
            Some(trimmed.to_string())
        }
    })
}

/// Resolves OAuth provider credentials from environment with optional alias + fallback default.
fn oauth_env_or_default(key: &str, alias: Option<&str>, default: &str) -> String {
    env_var_non_empty(key)
        .or_else(|| alias.and_then(env_var_non_empty))
        .unwrap_or_else(|| default.to_string())
}

/// Resolves OAuth credentials from the first non-empty env var in `keys`, else `default`.
fn oauth_env_first(keys: &[&str], default: &str) -> String {
    for key in keys {
        if let Some(value) = env_var_non_empty(key) {
            return value;
        }
    }
    default.to_string()
}

fn database_url_from_env() -> Option<String> {
    env_var_clean("DATABASE_URL")
}

/// Parses a permissive boolean environment variable with common truthy values.
fn parse_bool_env(key: &str) -> bool {
    env_var_clean(key)
        .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
        .unwrap_or(false)
}

/// Parses positive seconds into a checked `chrono::Duration`.
fn parse_positive_duration_seconds(
    value: i64,
    field_name: &str,
) -> Result<chrono::Duration, DynErr> {
    if value <= 0 {
        return Err(format!("{field_name} must be greater than 0 seconds").into());
    }
    chrono::Duration::try_seconds(value)
        .ok_or_else(|| format!("{field_name} is out of range").into())
}

/// Parses SameSite cookie mode from normalized configuration text.
fn parse_same_site(value: &str) -> Result<SameSite, DynErr> {
    match value.trim().to_ascii_lowercase().as_str() {
        "strict" => Ok(SameSite::Strict),
        "lax" => Ok(SameSite::Lax),
        "none" => Ok(SameSite::None),
        _ => Err(format!(
            "invalid auth.session.cookie_same_site value '{value}' (expected strict|lax|none)"
        )
        .into()),
    }
}

/// Parses configured GeoIP provider identifier into the internal enum.
fn parse_geoip_provider(value: &str) -> Result<GeoIpProvider, DynErr> {
    match value.trim().to_ascii_lowercase().as_str() {
        "ipwhois" | "ipwho.is" => Ok(GeoIpProvider::IpWhoIs),
        "ipapi" | "ipapi.co" => Ok(GeoIpProvider::IpApiCo),
        _ => Err(format!(
            "invalid auth.session.intelligence.geoip.provider value '{value}' (expected ipwhois|ipapi)"
        )
        .into()),
    }
}

fn parse_avatar_storage_url_style(value: &str) -> Result<AvatarStorageUrlStyle, DynErr> {
    match value.trim().to_ascii_lowercase().as_str() {
        "path" => Ok(AvatarStorageUrlStyle::Path),
        "virtual_host" | "virtual-host" | "virtualhost" => Ok(AvatarStorageUrlStyle::VirtualHost),
        _ => Err(format!(
            "invalid auth.avatar_storage.url_style value '{value}' (expected path|virtual_host)"
        )
        .into()),
    }
}

fn parse_optional_u32_env(env_keys: &[&str], field_name: &str) -> Result<Option<u32>, DynErr> {
    for env_key in env_keys {
        let Some(raw_value) = env_var_clean(env_key) else {
            continue;
        };
        let parsed = raw_value
            .parse::<u32>()
            .map_err(|error| format!("invalid {field_name} value from {env_key}: {error}"))?;
        return Ok(Some(parsed));
    }

    Ok(None)
}

/// Normalizes auth base path for Axum nesting and merge decisions.
fn nest_path(base_path: &str) -> String {
    let s = base_path.trim();
    if s.is_empty() || s == "/" {
        return "/".to_string();
    }
    s.trim_end_matches('/').to_string()
}

/// Extracts the host component used for passkey relying-party id derivation.
fn host_for_rp_id(base_url: &str) -> String {
    Url::parse(base_url)
        .ok()
        .and_then(|u| u.host_str().map(std::string::ToString::to_string))
        .filter(|h| !h.is_empty())
        .unwrap_or_else(|| "localhost".to_string())
}

/// Loads auth runtime config from env and optional YAML overlays.
fn load_config(port: u16, config_file: Option<&ServerConfigFile>) -> Result<AuthConfig, DynErr> {
    let jwt_secret = env_var_clean("JWT_SECRET")
        .ok_or("JWT_SECRET is not set (must be at least 32 characters for AuthConfig::new)")?;
    if jwt_secret.len() < 32 {
        return Err("JWT_SECRET must be at least 32 characters".into());
    }

    let yaml_auth = config_file.map(|cfg| &cfg.auth);

    let base_url = env_var_clean("BASE_URL")
        .or_else(|| {
            yaml_auth
                .and_then(|auth| auth.base_url.as_deref())
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(str::to_string)
        })
        .unwrap_or_else(|| format!("http://localhost:{port}"));

    let base_path = env_var_clean("AUTH_BASE_PATH")
        .or_else(|| {
            yaml_auth
                .and_then(|auth| auth.base_path.as_deref())
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .map(str::to_string)
        })
        .unwrap_or_else(|| "/api/auth".to_string());
    let public_base_path = env_var_clean("PUBLIC_AUTH_BASE_PATH").or_else(|| {
        yaml_auth
            .and_then(|auth| auth.public_base_path.as_deref())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(str::to_string)
    });

    let mut config = AuthConfig::new(jwt_secret)
        .base_url(base_url)
        .base_path(base_path)
        .password_min_length(8);
    if let Some(public_base_path) = public_base_path {
        config = config.public_base_path(public_base_path);
    }

    if let Some(origins) = env_var_clean("TRUSTED_ORIGINS") {
        for origin in origins
            .split(',')
            .map(str::trim)
            .filter(|value| !value.is_empty())
        {
            config = config.trusted_origin(origin.to_string());
        }
    } else if let Some(auth) = yaml_auth {
        for origin in &auth.trusted_origins {
            let origin: &str = origin.trim();
            if !origin.is_empty() {
                config = config.trusted_origin(origin.to_string());
            }
        }
    }

    if let Some(avatar_storage) = yaml_auth.map(|auth| &auth.avatar_storage) {
        let mut resolved_avatar_storage = AvatarStorageConfig::default();
        resolved_avatar_storage.enabled = avatar_storage.enabled;
        if let Some(bucket) = avatar_storage.bucket.as_deref().map(str::trim)
            && !bucket.is_empty()
        {
            resolved_avatar_storage.bucket = bucket.to_string();
        }
        if let Some(endpoint) = avatar_storage.endpoint.as_deref().map(str::trim)
            && !endpoint.is_empty()
        {
            resolved_avatar_storage.endpoint = endpoint.to_string();
        }
        if let Some(access_key) = avatar_storage.access_key.as_deref().map(str::trim)
            && !access_key.is_empty()
        {
            resolved_avatar_storage.access_key = access_key.to_string();
        }
        if let Some(secret_key) = avatar_storage.secret_key.as_deref().map(str::trim)
            && !secret_key.is_empty()
        {
            resolved_avatar_storage.secret_key = secret_key.to_string();
        }
        if let Some(url_style) = avatar_storage.url_style.as_deref().map(str::trim)
            && !url_style.is_empty()
        {
            resolved_avatar_storage.url_style = parse_avatar_storage_url_style(url_style)?;
        }
        resolved_avatar_storage.prefix_path = avatar_storage
            .prefix_path
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(str::to_string);
        if let Some(file_key_template) = avatar_storage.file_key_template.as_deref().map(str::trim)
            && !file_key_template.is_empty()
        {
            resolved_avatar_storage.file_key_template = file_key_template.to_string();
        }
        config.avatar_storage = resolved_avatar_storage;
    }

    if let Some(session) = yaml_auth.map(|auth| &auth.session) {
        if let Some(expires_in_seconds) = session.expires_in_seconds {
            config.session.expires_in = parse_positive_duration_seconds(
                expires_in_seconds,
                "auth.session.expires_in_seconds",
            )?;
        }

        if let Some(update_age_seconds) = session.update_age_seconds {
            config.session.update_age = if update_age_seconds == 0 {
                None
            } else {
                Some(parse_positive_duration_seconds(
                    update_age_seconds,
                    "auth.session.update_age_seconds",
                )?)
            };
        }

        if let Some(disable_session_refresh) = session.disable_session_refresh {
            config.session.disable_session_refresh = disable_session_refresh;
        }

        if let Some(fresh_age_seconds) = session.fresh_age_seconds {
            config.session.fresh_age = if fresh_age_seconds == 0 {
                None
            } else {
                Some(parse_positive_duration_seconds(
                    fresh_age_seconds,
                    "auth.session.fresh_age_seconds",
                )?)
            };
        }

        if let Some(cookie_name) = session.cookie_name.as_deref().map(str::trim)
            && !cookie_name.is_empty()
        {
            config.session.cookie_name = cookie_name.to_string();
        }
        if let Some(cookie_domain) = session.cookie_domain.as_deref().map(str::trim) {
            config.session.cookie_domain = if cookie_domain.is_empty() {
                None
            } else {
                Some(cookie_domain.to_string())
            };
        }
        if let Some(cookie_secure) = session.cookie_secure {
            config.session.cookie_secure = cookie_secure;
        }
        if let Some(cookie_http_only) = session.cookie_http_only {
            config.session.cookie_http_only = cookie_http_only;
        }
        if let Some(cookie_same_site) = session.cookie_same_site.as_deref() {
            config.session.cookie_same_site = parse_same_site(cookie_same_site)?;
        }

        if !session.trusted_proxy_headers.is_empty() {
            let trusted_proxy_headers: Vec<String> = session
                .trusted_proxy_headers
                .iter()
                .map(|value| value.trim().to_ascii_lowercase())
                .filter(|value| !value.is_empty())
                .collect();
            if !trusted_proxy_headers.is_empty() {
                config.advanced.trusted_proxy_headers = trusted_proxy_headers;
            }
        }
        if !session.ip.headers.is_empty() {
            let ip_headers: Vec<String> = session
                .ip
                .headers
                .iter()
                .map(|value| value.trim().to_ascii_lowercase())
                .filter(|value| !value.is_empty())
                .collect();
            if !ip_headers.is_empty() {
                config.advanced.ip_address.headers = ip_headers;
            }
        }
        if let Some(disable_tracking) = session.ip.disable_tracking {
            config.advanced.ip_address.disable_ip_tracking = disable_tracking;
        }

        if let Some(enabled) = session.intelligence.enabled {
            config.advanced.session_intelligence.enabled = enabled;
        }
        if let Some(detect_country_change) = session.intelligence.detect_country_change {
            config.advanced.session_intelligence.detect_country_change = detect_country_change;
        }

        let geoip = &session.intelligence.geoip;
        if let Some(enabled) = geoip.enabled {
            config.advanced.session_intelligence.geoip.enabled = enabled;
        }
        if let Some(provider) = geoip.provider.as_deref() {
            config.advanced.session_intelligence.geoip.provider = parse_geoip_provider(provider)?;
        }
        if let Some(base_url) = geoip
            .base_url
            .as_deref()
            .map(str::trim)
            .filter(|v| !v.is_empty())
        {
            config.advanced.session_intelligence.geoip.base_url = base_url.to_string();
        }
        if let Some(api_key_header) = geoip
            .api_key_header
            .as_deref()
            .map(str::trim)
            .filter(|v| !v.is_empty())
        {
            config.advanced.session_intelligence.geoip.api_key_header =
                Some(api_key_header.to_string());
        }
        if let Some(timeout_ms) = geoip.timeout_ms {
            config.advanced.session_intelligence.geoip.timeout_ms = timeout_ms.max(100);
        }
        if let Some(cache_ttl_seconds) = geoip.cache_ttl_seconds {
            config.advanced.session_intelligence.geoip.cache_ttl = parse_positive_duration_seconds(
                cache_ttl_seconds,
                "auth.session.intelligence.geoip.cache_ttl_seconds",
            )?;
        }

        config.advanced.session_intelligence.geoip.api_key = geoip
            .api_key
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(str::to_string)
            .or_else(|| {
                geoip
                    .api_key_env
                    .as_deref()
                    .map(str::trim)
                    .filter(|value| !value.is_empty())
                    .and_then(env_var_clean)
            });
    }

    if let Some(password) = yaml_auth.map(|auth| &auth.password) {
        if let Some(memory_cost) = password.argon2.memory_cost {
            config.password.argon2_config.memory_cost = memory_cost;
        }
        if let Some(time_cost) = password.argon2.time_cost {
            config.password.argon2_config.time_cost = time_cost;
        }
        if let Some(parallelism) = password.argon2.parallelism {
            config.password.argon2_config.parallelism = parallelism;
        }
    }

    if let Some(memory_cost) = parse_optional_u32_env(
        &[
            "ATHENA_AUTH_CONFIG__AUTH__PASSWORD__ARGON2__MEMORY_COST",
            "PASSWORD_ARGON2_MEMORY_COST",
        ],
        "auth.password.argon2.memory_cost",
    )? {
        config.password.argon2_config.memory_cost = memory_cost;
    }
    if let Some(time_cost) = parse_optional_u32_env(
        &[
            "ATHENA_AUTH_CONFIG__AUTH__PASSWORD__ARGON2__TIME_COST",
            "PASSWORD_ARGON2_TIME_COST",
        ],
        "auth.password.argon2.time_cost",
    )? {
        config.password.argon2_config.time_cost = time_cost;
    }
    if let Some(parallelism) = parse_optional_u32_env(
        &[
            "ATHENA_AUTH_CONFIG__AUTH__PASSWORD__ARGON2__PARALLELISM",
            "PASSWORD_ARGON2_PARALLELISM",
        ],
        "auth.password.argon2.parallelism",
    )? {
        config.password.argon2_config.parallelism = parallelism;
    }

    if let Some(cookie_domain) = env_var_clean("ATHENA_AUTH_CONFIG__AUTH__SESSION__COOKIE_DOMAIN")
        .or_else(|| env_var_clean("ATHENA_AUTH_COOKIE_DOMAIN"))
    {
        config = config.session_cookie_domain(cookie_domain);
    }

    Ok(config)
}

/// Resolves related origins used by `/.well-known/webauthn` for WebAuthn ROR.
fn resolve_passkey_related_origins(config_file: Option<&ServerConfigFile>) -> Vec<String> {
    if let Some(origins) = env_var_clean("PASSKEY_RELATED_ORIGINS") {
        return parse_origins_csv(origins.as_str());
    }

    config_file
        .map(|cfg| normalize_origins(&cfg.auth.passkey.related_origins))
        .unwrap_or_default()
}

/// Creates a passkey plugin aligned with the configured server origin.
fn passkey_plugin_for(base_url: &str) -> PasskeyPlugin {
    let rp_id: String = env_var_clean("PASSKEY_RP_ID").unwrap_or_else(|| host_for_rp_id(base_url));
    let rp_name: String =
        env_var_clean("PASSKEY_RP_NAME").unwrap_or_else(|| "Athena Auth".to_string());
    let origin: String = env_var_clean("PASSKEY_ORIGIN").unwrap_or_else(|| base_url.to_string());
    // Legacy compatibility knob: passkey verification is always server-side.
    let allow_insecure: bool = parse_bool_env("PASSKEY_ALLOW_INSECURE");

    PasskeyPlugin::new()
        .rp_id(rp_id)
        .rp_name(rp_name)
        .origin(origin)
        .allow_insecure_unverified_assertion(allow_insecure)
}

#[derive(Debug, Clone)]
struct SelfHealingAuditEvent {
    auth_schema: String,
    schema_existed_before: bool,
    missing_tables_before: Vec<String>,
}

#[derive(Debug, Clone)]
struct AuditTableColumn {
    name: String,
    data_type: String,
    udt_name: String,
    is_nullable: bool,
    has_default: bool,
}

fn escape_sql_string_literal(value: &str) -> String {
    value.replace('\'', "''")
}

fn sql_string_literal(value: &str) -> String {
    format!("'{}'", escape_sql_string_literal(value))
}

fn build_self_heal_audit_json(event: &SelfHealingAuditEvent) -> String {
    json!({
        "source": "athena-auth",
        "event": "startup_self_heal",
        "auth_schema": event.auth_schema,
        "schema_existed_before": event.schema_existed_before,
        "missing_tables_before": event.missing_tables_before,
        "table_name": format!("{SELF_HEAL_AUDIT_SCHEMA}.{SELF_HEAL_AUDIT_TABLE_DDL}")
    })
    .to_string()
}

fn preferred_self_heal_audit_columns() -> &'static [&'static str] {
    &[
        "request_id",
        "client_name",
        "schema_name",
        "function_name",
        "function_signature",
        "action",
        "status",
        "ddl_sql",
        "previous_definition",
        "next_definition",
        "error_message",
        "api_key_public_id",
        "user_id",
        "company_id",
        "organization_id",
        "remote_addr",
        "ipv4_address",
        "user_agent",
        "duration_ms",
        "time",
        "operation",
        "object_type",
        "table_name",
        "table",
        "target",
        "route",
        "request",
        "request_payload",
        "resource",
        "resource_id",
        "status_message",
        "message",
    ]
}

fn self_heal_audit_literal_for_column(
    column: &AuditTableColumn,
    event: &SelfHealingAuditEvent,
    payload_json: &str,
) -> Option<String> {
    let message: String = format!(
        "startup self-heal ensured schema '{}' with missing tables before apply: {}",
        event.auth_schema,
        if event.missing_tables_before.is_empty() {
            "<none>".to_string()
        } else {
            event.missing_tables_before.join(",")
        }
    );
    let ddl_sql_message = format!(
        "athena-auth startup self-heal via sql/provision.sql in schema {}",
        event.auth_schema
    );
    let table_ref: String = format!("{SELF_HEAL_AUDIT_SCHEMA}.{SELF_HEAL_AUDIT_TABLE_DDL}");

    match column.name.as_str() {
        "request_id" => Some("gen_random_uuid()".to_string()),
        "client_name" => Some(sql_string_literal("athena-auth")),
        "schema_name" => Some(sql_string_literal(event.auth_schema.as_str())),
        "function_name" => Some(sql_string_literal("ensure_auth_schema_provisioned")),
        "function_signature" => Some(sql_string_literal("startup")),
        "action" | "operation" => Some(sql_string_literal("startup_self_heal")),
        "status" => Some(sql_string_literal("success")),
        "ddl_sql" => Some(sql_string_literal(ddl_sql_message.as_str())),
        "previous_definition" => Some("'[]'::jsonb".to_string()),
        "next_definition" => Some(format!(
            "'{}'::jsonb",
            escape_sql_string_literal(payload_json)
        )),
        "error_message" => Some("NULL".to_string()),
        "api_key_public_id" => Some(sql_string_literal("system:self-heal")),
        "user_id" | "company_id" | "organization_id" => Some("NULL".to_string()),
        "remote_addr" | "ipv4_address" => Some(sql_string_literal("127.0.0.1")),
        "user_agent" => Some(sql_string_literal("athena-auth/startup-self-heal")),
        "duration_ms" => Some("0".to_string()),
        "time" => Some("EXTRACT(EPOCH FROM now())::bigint".to_string()),
        "object_type" => Some(sql_string_literal("schema")),
        "table_name" | "table" | "target" => Some(sql_string_literal(event.auth_schema.as_str())),
        "route" => Some(sql_string_literal("server.startup")),
        "request" | "request_payload" | "resource" => Some(format!(
            "'{}'::jsonb",
            escape_sql_string_literal(payload_json)
        )),
        "resource_id" => Some(sql_string_literal("startup_self_heal")),
        "status_message" | "message" => Some(sql_string_literal(message.as_str())),
        _ => {
            let data_type = column.data_type.to_ascii_lowercase();
            let udt_name = column.udt_name.to_ascii_lowercase();
            match data_type.as_str() {
                "uuid" => Some("gen_random_uuid()".to_string()),
                "boolean" => Some("TRUE".to_string()),
                "json" => Some(format!(
                    "'{}'::json",
                    escape_sql_string_literal(payload_json)
                )),
                "jsonb" => Some(format!(
                    "'{}'::jsonb",
                    escape_sql_string_literal(payload_json)
                )),
                "timestamp with time zone" => Some("now()".to_string()),
                "timestamp without time zone" => Some("now()::timestamp".to_string()),
                "smallint" | "integer" | "bigint" | "numeric" | "real" | "double precision" => {
                    Some("0".to_string())
                }
                "text" | "character varying" | "character" => {
                    Some(sql_string_literal(message.as_str()))
                }
                "user-defined" if udt_name == "uuid" => Some("gen_random_uuid()".to_string()),
                _ => None,
            }
        }
    }
    .or_else(|| {
        if column.is_nullable || column.has_default {
            Some("DEFAULT".to_string())
        } else {
            None
        }
    })
    .or_else(|| {
        tracing::debug!(
            column = %column.name,
            data_type = %column.data_type,
            udt_name = %column.udt_name,
            table = %table_ref,
            "no literal mapper for self-heal audit column"
        );
        None
    })
}

fn build_self_heal_audit_insert_sql(
    columns: &[AuditTableColumn],
    event: &SelfHealingAuditEvent,
) -> Option<String> {
    let payload_json = build_self_heal_audit_json(event);
    let preferred: std::collections::BTreeSet<&str> = preferred_self_heal_audit_columns()
        .iter()
        .copied()
        .collect();

    let mut insert_columns: Vec<String> = Vec::new();
    let mut insert_values: Vec<String> = Vec::new();
    let mut required_columns_without_value: Vec<String> = Vec::new();

    for column in columns {
        let should_attempt: bool =
            !column.is_nullable || !column.has_default || preferred.contains(column.name.as_str());
        if !should_attempt {
            continue;
        }

        let Some(literal) =
            self_heal_audit_literal_for_column(column, event, payload_json.as_str())
        else {
            if !column.is_nullable && !column.has_default {
                required_columns_without_value.push(column.name.clone());
            }
            continue;
        };

        insert_columns.push(quote_pg_identifier(column.name.as_str()));
        insert_values.push(literal);
    }

    if !required_columns_without_value.is_empty() {
        tracing::warn!(
            missing_columns = %required_columns_without_value.join(","),
            table = %format!("{SELF_HEAL_AUDIT_SCHEMA}.{SELF_HEAL_AUDIT_TABLE_DDL}"),
            "self-heal audit skipped because required columns could not be mapped"
        );
        return None;
    }

    if insert_columns.is_empty() {
        return None;
    }

    Some(format!(
        "INSERT INTO {}.{} ({}) VALUES ({})",
        quote_pg_identifier(SELF_HEAL_AUDIT_SCHEMA),
        quote_pg_identifier(SELF_HEAL_AUDIT_TABLE_DDL),
        insert_columns.join(", "),
        insert_values.join(", ")
    ))
}

/// Writes self-healing startup activity to athena.audit_log_ddl when that table exists.
async fn record_self_heal_audit_log(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    event: &SelfHealingAuditEvent,
) -> Result<(), DynErr> {
    let rows: Vec<sqlx::postgres::PgRow> = sqlx::query(
        "SELECT \
            column_name::text AS column_name, \
            data_type::text AS data_type, \
            udt_name::text AS udt_name, \
            is_nullable::text AS is_nullable, \
            column_default::text AS column_default \
         FROM information_schema.columns \
         WHERE table_schema = $1 AND table_name = $2 \
         ORDER BY ordinal_position ASC",
    )
    .bind(SELF_HEAL_AUDIT_SCHEMA)
    .bind(SELF_HEAL_AUDIT_TABLE_DDL)
    .fetch_all(&mut **tx)
    .await
    .map_err(|error| format!("failed to inspect self-heal audit table columns: {error}"))?;

    if rows.is_empty() {
        return Ok(());
    }

    let mut columns: Vec<AuditTableColumn> = Vec::with_capacity(rows.len());
    for row in rows {
        let name: String = row
            .try_get("column_name")
            .map_err(|error| format!("failed to decode audit column_name: {error}"))?;
        let data_type: String = row
            .try_get("data_type")
            .map_err(|error| format!("failed to decode audit data_type: {error}"))?;
        let udt_name: String = row
            .try_get("udt_name")
            .map_err(|error| format!("failed to decode audit udt_name: {error}"))?;
        let is_nullable_raw: String = row
            .try_get("is_nullable")
            .map_err(|error| format!("failed to decode audit is_nullable: {error}"))?;
        let has_default = row
            .try_get::<Option<String>, _>("column_default")
            .unwrap_or(None)
            .is_some();

        columns.push(AuditTableColumn {
            name: name.to_ascii_lowercase(),
            data_type: data_type.to_ascii_lowercase(),
            udt_name: udt_name.to_ascii_lowercase(),
            is_nullable: !is_nullable_raw.eq_ignore_ascii_case("NO"),
            has_default,
        });
    }

    let Some(insert_sql) = build_self_heal_audit_insert_sql(&columns, event) else {
        return Ok(());
    };

    sqlx::query(insert_sql.as_str())
        .execute(&mut **tx)
        .await
        .map_err(|error| format!("failed to insert self-heal audit row: {error}"))?;
    Ok(())
}

async fn ensure_pg_schema_exists(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    schema_name: &str,
) -> Result<(), DynErr> {
    let create_schema_sql = format!(
        "CREATE SCHEMA IF NOT EXISTS {}",
        quote_pg_identifier(schema_name)
    );
    sqlx::query(create_schema_sql.as_str())
        .execute(&mut **tx)
        .await
        .map_err(|error| format!("failed to ensure schema '{schema_name}': {error}"))?;
    Ok(())
}

/// Ensures the auth schema exists and applies core auth DDL before runtime traffic starts.
async fn ensure_auth_schema_provisioned(
    database_url: &str,
    auth_schema: &str,
) -> Result<(), DynErr> {
    let auth_schema = normalize_auth_schema_for_runtime(auth_schema)?;
    let connect_options =
        parse_pg_connect_options(database_url).map_err(|error| DynErr::from(error))?;
    let pool = PgPoolOptions::new()
        .max_connections(1)
        .min_connections(0)
        .acquire_timeout(Duration::from_secs(20))
        .connect_with(connect_options)
        .await
        .map_err(|error| format!("failed to connect for schema provisioning: {error}"))?;

    let mut tx = pool
        .begin()
        .await
        .map_err(|error| format!("failed to start schema provisioning transaction: {error}"))?;

    let schema_existed_before = sqlx::query(
        "SELECT EXISTS(SELECT 1 FROM pg_namespace WHERE nspname = $1) AS schema_exists",
    )
    .bind(auth_schema.as_str())
    .fetch_one(&mut *tx)
    .await
    .map_err(|error| format!("failed to inspect schema existence before self-heal: {error}"))?
    .try_get::<bool, _>("schema_exists")
    .map_err(|error| format!("failed to decode schema existence state: {error}"))?;

    let existing_tables_rows = sqlx::query(
        "SELECT table_name::text AS table_name \
         FROM information_schema.tables \
         WHERE table_schema = $1 AND table_type = 'BASE TABLE'",
    )
    .bind(auth_schema.as_str())
    .fetch_all(&mut *tx)
    .await
    .map_err(|error| format!("failed to inspect existing tables before self-heal: {error}"))?;

    let existing_tables: std::collections::BTreeSet<String> = existing_tables_rows
        .into_iter()
        .filter_map(|row| row.try_get::<String, _>("table_name").ok())
        .map(|value| value.to_ascii_lowercase())
        .collect();
    let missing_tables_before: Vec<String> = LOGICAL_TABLE_NAMES
        .iter()
        .map(|name| name.to_ascii_lowercase())
        .filter(|name| !existing_tables.contains(name))
        .collect();

    let ddl_audit_table_existed_before = sqlx::query("SELECT to_regclass($1)::text AS table_ref")
        .bind(format!(
            "{SELF_HEAL_AUDIT_SCHEMA}.{SELF_HEAL_AUDIT_TABLE_DDL}"
        ))
        .fetch_one(&mut *tx)
        .await
        .map_err(|error| format!("failed to inspect self-heal audit table existence: {error}"))?
        .try_get::<Option<String>, _>("table_ref")
        .map_err(|error| {
            format!("failed to decode self-heal audit table existence state: {error}")
        })?
        .is_some();

    for bootstrap_schema in schema_bootstrap_targets(auth_schema.as_str()) {
        ensure_pg_schema_exists(&mut tx, bootstrap_schema).await?;
    }

    let set_search_path_sql = format!(
        "SET LOCAL search_path TO {}",
        quote_pg_identifier(&auth_schema)
    );
    sqlx::query(&set_search_path_sql)
        .execute(&mut *tx)
        .await
        .map_err(|error| {
            format!("failed to set provisioning search_path for '{auth_schema}': {error}")
        })?;

    let provision_sql = auth_provision_sql_for_schema(auth_schema.as_str());
    sqlx::raw_sql(provision_sql.as_str())
        .execute(&mut *tx)
        .await
        .map_err(|error| {
            format!("failed to apply auth provision SQL in schema '{auth_schema}': {error}")
        })?;

    let healing_was_required = !schema_existed_before || !missing_tables_before.is_empty();
    if healing_was_required && ddl_audit_table_existed_before {
        let event = SelfHealingAuditEvent {
            auth_schema: auth_schema.clone(),
            schema_existed_before,
            missing_tables_before,
        };

        if let Err(error) = record_self_heal_audit_log(&mut tx, &event).await {
            tracing::warn!(
                auth_schema = %auth_schema,
                error = %error,
                table = %format!("{SELF_HEAL_AUDIT_SCHEMA}.{SELF_HEAL_AUDIT_TABLE_DDL}"),
                "failed to persist startup self-heal audit row"
            );
        }
    }

    tx.commit()
        .await
        .map_err(|error| format!("failed to commit schema provisioning transaction: {error}"))?;

    tracing::info!(
        auth_schema = %auth_schema,
        "ensured auth schema and required tables are provisioned"
    );
    Ok(())
}

/// Builds a SQLx adapter, optionally forcing tenant-specific schema search path.
async fn build_sqlx_adapter(
    database_url: &str,
    search_path: Option<&SearchPath>,
    auth_schema: &str,
) -> Result<SqlxAdapter, DynErr> {
    let auth_schema = normalize_auth_schema_for_runtime(auth_schema)?;
    if parse_bool_env("ATHENA_AUTH_SKIP_SCHEMA_PROVISIONING") {
        tracing::warn!(
            auth_schema = %auth_schema,
            "skipping auth schema provisioning because ATHENA_AUTH_SKIP_SCHEMA_PROVISIONING is enabled"
        );
    } else {
        ensure_auth_schema_provisioned(database_url, auth_schema.as_str()).await?;
    }

    let pool_config: PoolConfig = PoolConfig::default();
    let mut options: sqlx::pool::PoolOptions<sqlx::Postgres> = PgPoolOptions::new()
        .max_connections(pool_config.max_connections)
        .min_connections(pool_config.min_connections)
        .acquire_timeout(pool_config.acquire_timeout)
        .idle_timeout(pool_config.idle_timeout)
        .max_lifetime(pool_config.max_lifetime);

    let effective_search_path = build_effective_search_path(auth_schema.as_str(), search_path);
    let set_sql: String = format!("SET search_path TO {effective_search_path}");
    options = options.after_connect(move |conn, _meta| {
        let set_sql = set_sql.clone();
        Box::pin(async move {
            sqlx::query(&set_sql).execute(conn).await?;
            Ok(())
        })
    });

    let connect_options: PgConnectOptions =
        parse_pg_connect_options(database_url).map_err(|error| DynErr::from(error))?;
    let pool: sqlx::Pool<sqlx::Postgres> = options
        .connect_with(connect_options)
        .await
        .map_err(|error| format!("could not open PostgreSQL pool: {error}"))?;
    let adapter: SqlxAdapter = SqlxAdapter::from_pool_with_schema(pool, auth_schema.as_str());
    adapter.test_connection().await.map_err(|error| {
        format!("database reachable but health query failed (SELECT 1): {error}")
    })?;
    Ok(adapter)
}

/// Builds a fully configured AthenaAuth runtime with all enabled plugins.
async fn build_auth(
    config: AuthConfig,
    adapter: SqlxAdapter,
) -> Result<Arc<AthenaAuth<SqlxAdapter>>, DynErr> {
    let base_url: String = config.base_url.clone();
    let trusted_origins: Vec<String> = config.trusted_origins.clone();
    let adapter: Arc<SqlxAdapter> = Arc::new(adapter);

    let mut auth_builder: AuthBuilder = AuthBuilder::new(config);
    let mut cors_config = CorsConfig::new();
    let mut cors_origins_seen: std::collections::BTreeSet<String> =
        std::collections::BTreeSet::new();

    if let Ok(parsed_base_url) = Url::parse(&base_url)
        && let Some(host) = parsed_base_url.host_str()
    {
        let mut base_origin = format!("{}://{}", parsed_base_url.scheme(), host);
        if let Some(port) = parsed_base_url.port() {
            base_origin = format!("{base_origin}:{port}");
        }
        if cors_origins_seen.insert(base_origin.clone()) {
            cors_config = cors_config.allowed_origin(base_origin);
        }
    }

    for origin in trusted_origins {
        let trimmed = origin.trim();
        if trimmed.is_empty() {
            continue;
        }
        if cors_origins_seen.insert(trimmed.to_string()) {
            cors_config = cors_config.allowed_origin(trimmed.to_string());
        }
    }

    auth_builder = auth_builder.cors(cors_config);

    match SmtpEmailProvider::try_from_env()
        .map_err(|e| format!("invalid SMTP configuration: {e}"))?
    {
        None => {}
        Some(smtp) => {
            let from_address = env_var_clean("SMTP_FROM_ADDRESS").ok_or({
                "SMTP_FROM_ADDRESS must be set when SMTP_HOST is configured (required for audit logging)"
            })?;
            let from_name = env_var_clean("SMTP_FROM_NAME").filter(|s| !s.is_empty());
            let inner: Arc<dyn EmailProvider> = Arc::new(smtp);
            auth_builder = auth_builder.email_provider(DatabaseLoggingEmailProvider::new(
                adapter.clone(),
                inner,
                from_address,
                from_name,
                Some("smtp".to_string()),
            ));
        }
    }

    let google_client_id = oauth_env_or_default(
        "GOOGLE_CLIENT_ID",
        None,
        "14641082008-2u3231e04vhka4dvpgjus8a7clqes6mb.apps.googleusercontent.com",
    );
    let google_client_secret = oauth_env_or_default(
        "GOOGLE_CLIENT_SECRET",
        None,
        "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",
    );
    let apple_client_id =
        oauth_env_or_default("APPLE_CLIENT_ID", None, "com.example.athena-auth.service");
    let apple_client_secret = oauth_env_or_default(
        "APPLE_CLIENT_SECRET",
        None,
        "replace-with-apple-client-secret-jwt",
    );
    let microsoft_client_id = oauth_env_or_default(
        "MICROSOFT_CLIENT_ID",
        None,
        "00000000-0000-0000-0000-000000000000",
    );
    let microsoft_client_secret = oauth_env_or_default(
        "MICROSOFT_CLIENT_SECRET",
        Some("MICROSOFT_SECRET"),
        "replace-with-microsoft-client-secret",
    );

    let github_client_id = oauth_env_first(
        &[
            "GITHUB_CLIENT_ID",
            "GITHUB_OAUTH_CLIENT_ID",
            "GITHUB_APP_CLIENT_ID",
        ],
        "",
    );
    let github_client_secret = oauth_env_first(
        &[
            "GITHUB_CLIENT_SECRET",
            "GITHUB_OAUTH_CLIENT_SECRET",
            "GITHUB_APP_CLIENT_SECRET",
        ],
        "",
    );
    let cloudflare_client_id =
        oauth_env_first(&["CLOUDFLARE_CLIENT_ID", "CLOUDFLARE_OAUTH_CLIENT_ID"], "");
    let cloudflare_client_secret = oauth_env_first(
        &[
            "CLOUDFLARE_CLIENT_SECRET",
            "CLOUDFLARE_OAUTH_CLIENT_SECRET",
            "CLOUDFLARE_OAUTH_CLIENT_PUBLISHER",
            "cloudflare_oauth_client_publisher",
        ],
        "",
    );

    let oauth_plugin = OAuthPlugin::new()
        .add_provider(
            "google",
            athena_auth::plugins::oauth::OAuthProvider::google(
                &google_client_id,
                &google_client_secret,
            ),
        )
        .add_provider(
            "apple",
            athena_auth::plugins::oauth::OAuthProvider::apple(
                &apple_client_id,
                &apple_client_secret,
            ),
        )
        .add_provider(
            "microsoft",
            athena_auth::plugins::oauth::OAuthProvider::microsoft(
                &microsoft_client_id,
                &microsoft_client_secret,
            ),
        )
        .add_provider(
            "github",
            athena_auth::plugins::oauth::OAuthProvider::github(
                &github_client_id,
                &github_client_secret,
            ),
        )
        .add_provider(
            "cloudflare",
            athena_auth::plugins::oauth::OAuthProvider::cloudflare(
                &cloudflare_client_id,
                &cloudflare_client_secret,
            ),
        );

    let auth = auth_builder
        .database(adapter.as_ref().clone())
        .plugin(EmailPasswordPlugin::new().enable_signup(true))
        .plugin(SessionManagementPlugin::new())
        .plugin(PasswordManagementPlugin::new().require_current_password(true))
        .plugin(AccountManagementPlugin::new())
        .plugin(EmailVerificationPlugin::new())
        .plugin(
            ApiKeyPlugin::builder()
                .key_length(32)
                .prefix("ath_".to_string())
                .enable_session_for_api_keys(true)
                .build(),
        )
        .plugin(oauth_plugin)
        .plugin(TwoFactorPlugin::new())
        .plugin(OrganizationPlugin::new())
        .plugin(passkey_plugin_for(&base_url))
        .plugin(AdminPlugin::new())
        .plugin(
            UserManagementPlugin::new()
                .change_email_enabled(true)
                .delete_user_enabled(true),
        )
        .build()
        .await?;

    Ok(Arc::new(auth))
}

/// Builds the tenant runtime registry for request-time tenant resolution.
async fn build_tenant_registry(
    config: &AuthConfig,
    fallback_database_url: &str,
    file_config: Option<&ServerConfigFile>,
) -> Result<Arc<TenantAuthRegistry>, DynErr> {
    let schema_detection_auth_schema = schema_detection_auth_schema(file_config)?;
    let parsed_multi = ParsedMultiTenancyConfig::from_raw(
        file_config
            .map(|value| value.multitenancy.clone())
            .unwrap_or_default(),
    )?;

    let resolver = HeaderTenantResolver::new(
        parsed_multi.header_name.clone(),
        parsed_multi.default_tenant.clone(),
    );

    let mut auth_by_tenant: std::collections::BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>> =
        std::collections::BTreeMap::new();

    if !parsed_multi.is_multi_tenant_enabled() {
        let auth_schema = schema_detection_auth_schema
            .as_deref()
            .unwrap_or(DEFAULT_AUTH_SCHEMA);
        let adapter = build_sqlx_adapter(fallback_database_url, None, auth_schema).await?;
        let auth = build_auth(config.clone(), adapter).await?;
        auth_by_tenant.insert(parsed_multi.default_tenant.clone(), auth);
        tracing::info!(
            tenant = %parsed_multi.default_tenant,
            auth_schema = %auth_schema,
            "multitenancy disabled; using single default tenant"
        );
    } else {
        for (tenant_id, tenant_config) in &parsed_multi.tenants {
            let auth_schema = schema_detection_auth_schema
                .as_deref()
                .unwrap_or(tenant_config.auth_schema.as_str());
            let database_url = tenant_config.database_url_source.resolve(
                tenant_id,
                fallback_database_url,
                env_var_clean,
            )?;
            let adapter = build_sqlx_adapter(
                &database_url,
                tenant_config.search_path.as_ref(),
                auth_schema,
            )
            .await?;
            let auth: Arc<AthenaAuth<SqlxAdapter>> = build_auth(config.clone(), adapter).await?;
            auth_by_tenant.insert(tenant_id.clone(), auth);
            tracing::info!(
                tenant = %tenant_id,
                auth_schema = %auth_schema,
                search_path = %tenant_config
                    .search_path
                    .as_ref()
                    .map(SearchPath::as_sql_fragment)
                    .unwrap_or("default"),
                schema_mapping_entries = tenant_config.schema_mapping.len(),
                "initialized tenant auth runtime"
            );
        }

        if !auth_by_tenant.contains_key(&parsed_multi.default_tenant) {
            let auth_schema = schema_detection_auth_schema
                .as_deref()
                .unwrap_or(DEFAULT_AUTH_SCHEMA);
            let adapter = build_sqlx_adapter(fallback_database_url, None, auth_schema).await?;
            let auth = build_auth(config.clone(), adapter).await?;
            auth_by_tenant.insert(parsed_multi.default_tenant.clone(), auth);
            tracing::warn!(
                tenant = %parsed_multi.default_tenant,
                auth_schema = %auth_schema,
                "default tenant not present in config; injected fallback tenant using DATABASE_URL"
            );
        }
    }

    Ok(Arc::new(TenantRuntimeRegistry::new(
        resolver,
        auth_by_tenant,
    )?))
}

fn header_value(headers: &HeaderMap, name: &str) -> Option<String> {
    headers
        .get(name)
        .and_then(|header| header.to_str().ok())
        .map(str::to_string)
}

fn header_present(headers: &HeaderMap, name: &str) -> bool {
    headers.get(name).is_some()
}

fn session_cookie_present(headers: &HeaderMap, cookie_name: &str) -> bool {
    let Some(cookie_header) = headers
        .get("cookie")
        .and_then(|header| header.to_str().ok())
    else {
        return false;
    };

    cookie_header.split(';').map(str::trim).any(|entry| {
        let Some((name, value)) = entry.split_once('=') else {
            return false;
        };
        name == cookie_name && !value.is_empty()
    })
}

fn query_param_value(query: Option<&str>, key: &str) -> Option<String> {
    query.and_then(|query| {
        url::form_urlencoded::parse(query.as_bytes())
            .find(|(candidate, _)| candidate == key)
            .map(|(_, value)| value.to_string())
    })
}

fn auth_request_trace_context(req: &Request, session_cookie_name: &str) -> AuthRequestTraceContext {
    AuthRequestTraceContext {
        method: req.method().to_string(),
        path: req.uri().path().to_string(),
        query: req.uri().query().map(str::to_string),
        origin: header_value(req.headers(), "origin"),
        referer: header_value(req.headers(), "referer"),
        authorization_header_present: header_present(req.headers(), "authorization"),
        cookie_header_present: header_present(req.headers(), "cookie"),
        session_cookie_name: session_cookie_name.to_string(),
        session_cookie_present: session_cookie_present(req.headers(), session_cookie_name),
        disable_cookie_cache: query_param_value(req.uri().query(), "disableCookieCache"),
    }
}

fn remote_ip_from_request(req: &Request) -> Option<String> {
    if let Some(connect_info) = req
        .extensions()
        .get::<axum::extract::connect_info::ConnectInfo<SocketAddr>>()
    {
        return Some(connect_info.0.ip().to_string());
    }
    if let Some(addr) = req.extensions().get::<SocketAddr>() {
        return Some(addr.ip().to_string());
    }
    None
}

fn client_ip_for_request(state: &ServerMiddlewareState, req: &Request) -> Option<String> {
    let mut headers = std::collections::HashMap::new();
    for (name, value) in req.headers() {
        if name
            .as_str()
            .eq_ignore_ascii_case(INTERNAL_AXUM_REMOTE_IP_HEADER)
        {
            continue;
        }
        if let Ok(value_str) = value.to_str() {
            headers.insert(name.to_string(), value_str.to_string());
        }
    }

    if let Some(remote_ip) = remote_ip_from_request(req) {
        headers.insert(INTERNAL_AXUM_REMOTE_IP_HEADER.to_string(), remote_ip);
    }

    resolve_client_ip_from_headers(&headers, &state.ip_headers, &state.trusted_proxy_headers)
}

fn blocked_client_ip(state: &ServerMiddlewareState, req: &Request) -> Option<String> {
    let client_ip = client_ip_for_request(state, req)?;
    state
        .blocked_ips
        .contains(client_ip.as_str())
        .then_some(client_ip)
}

fn is_auth_catch_all_route(matched_path: &str) -> bool {
    matched_path == AxumRouterComposer::AXUM08_WILDCARD_ROUTE
        || matched_path.ends_with(AxumRouterComposer::AXUM08_WILDCARD_ROUTE)
}

fn resolve_observed_route(
    state: &ServerMiddlewareState,
    method: &str,
    matched_path: Option<&str>,
    request_path: &str,
) -> String {
    if let Some(matched_path) = matched_path
        && !is_auth_catch_all_route(matched_path)
    {
        return matched_path.to_string();
    }

    state
        .auth_route_labels
        .resolve(method, request_path)
        .map(str::to_string)
        .unwrap_or_else(|| request_path.to_string())
}

fn content_length_bytes(headers: &HeaderMap) -> Option<u64> {
    headers
        .get("content-length")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse::<u64>().ok())
}

fn observed_tenant(headers: &HeaderMap) -> Option<String> {
    header_value_trimmed(headers, DEFAULT_TENANT_HEADER)
        .or_else(|| {
            FALLBACK_TENANT_HEADERS
                .iter()
                .find_map(|header_name| header_value_trimmed(headers, header_name))
        })
        .or_else(|| header_value_trimmed(headers, "x-tenant-id"))
        .or_else(|| header_value_trimmed(headers, "x-tenant"))
}

fn configured_auth_rewrite_prefix() -> Option<String> {
    let raw = env_var_clean("PUBLIC_AUTH_BASE_PATH")
        .or_else(|| {
            env_var_clean("AUTH_BASE_PATH").filter(|value| {
                let trimmed = value.trim();
                !trimmed.is_empty() && trimmed != "/"
            })
        })
        .unwrap_or_else(|| DEFAULT_PUBLIC_AUTH_BASE_PATH_PREFIX.to_string());
    let trimmed = raw.trim();
    if trimmed.is_empty() || trimmed == "/" {
        return None;
    }
    if trimmed.starts_with('/') {
        Some(trimmed.trim_end_matches('/').to_string())
    } else {
        Some(format!("/{}", trimmed.trim_end_matches('/')))
    }
}

fn rewrite_legacy_auth_base_path(req: &mut Request) {
    let path = req.uri().path();
    let Some(rewrite_prefix) = configured_auth_rewrite_prefix() else {
        return;
    };
    let Some(rewritten_path) = path.strip_prefix(rewrite_prefix.as_str()) else {
        return;
    };
    if !rewritten_path.is_empty() && !rewritten_path.starts_with('/') {
        return;
    }

    let normalized_path = if rewritten_path.is_empty() {
        "/"
    } else {
        rewritten_path
    };

    let path_and_query = if let Some(query) = req.uri().query() {
        format!("{normalized_path}?{query}")
    } else {
        normalized_path.to_string()
    };

    let Ok(uri) = Uri::builder()
        .path_and_query(path_and_query.as_str())
        .build()
    else {
        tracing::warn!(
            original_path = %path,
            rewritten_path = %normalized_path,
            "failed to apply legacy auth base-path rewrite shim"
        );
        return;
    };

    *req.uri_mut() = uri;
}

async fn legacy_auth_base_path_rewrite_middleware(
    mut req: axum::extract::Request,
    next: Next,
) -> Response {
    let original_path = req.uri().path().to_string();
    rewrite_legacy_auth_base_path(&mut req);
    if original_path != req.uri().path() {
        tracing::debug!(
            original_path = %original_path,
            rewritten_path = %req.uri().path(),
            "applied legacy auth base-path rewrite shim"
        );
    }

    next.run(req).await
}

/// Resolves the tenant from headers and dispatches the auth request to that runtime.
async fn dispatch_auth_request(
    State(registry): State<Arc<TenantAuthRegistry>>,
    mut req: Request,
) -> Response {
    let trace_id = ensure_request_trace_id(&mut req);
    if let Some(response) =
        maybe_handle_admin_docs_request(registry.as_ref(), req.uri().path(), trace_id.as_str())
            .await
    {
        return response;
    }
    tracing::trace!(
        trace_id = %trace_id,
        method = %req.method(),
        path = %req.uri().path(),
        "dispatching auth request"
    );
    let tenant_result = registry.resolve_tenant_from_headers(&AxumHeaderLookup {
        headers: req.headers(),
    });
    let tenant_id = match tenant_result {
        Ok(value) => value,
        Err(error) => {
            tracing::warn!(
                trace_id = %trace_id,
                method = %req.method(),
                path = %req.uri().path(),
                error = %error,
                "failed to resolve tenant for auth request"
            );
            return json_error_response(
                StatusCode::BAD_REQUEST,
                error.to_string(),
                trace_id.as_str(),
            );
        }
    };
    tracing::trace!(
        trace_id = %trace_id,
        tenant = %tenant_id,
        "resolved tenant for auth request"
    );

    let Some(auth) = registry.runtime(&tenant_id).cloned() else {
        tracing::error!(
            trace_id = %trace_id,
            tenant = %tenant_id,
            path = %req.uri().path(),
            "resolved tenant is missing auth runtime"
        );
        return json_error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("Tenant '{tenant_id}' is registered but auth runtime is missing"),
            trace_id.as_str(),
        );
    };

    let trace_context =
        auth_request_trace_context(&req, auth.config().session.cookie_name.as_str());
    tracing::debug!(
        trace_id = %trace_id,
        tenant = %tenant_id,
        method = %trace_context.method,
        path = %trace_context.path,
        query = ?trace_context.query,
        origin = ?trace_context.origin,
        referer = ?trace_context.referer,
        authorization_header_present = trace_context.authorization_header_present,
        cookie_header_present = trace_context.cookie_header_present,
        session_cookie_name = %trace_context.session_cookie_name,
        session_cookie_present = trace_context.session_cookie_present,
        disable_cookie_cache = ?trace_context.disable_cookie_cache,
        "auth dispatch context"
    );

    if let Ok(header_value) = HeaderValue::from_str(tenant_id.as_str()) {
        req.headers_mut().insert("x-athena-tenant", header_value);
    }

    let mut response = dispatch_auth_axum_request(auth.as_ref(), req).await;
    let status = response.status();

    if status == StatusCode::UNAUTHORIZED || trace_context.path.ends_with("/get-session") {
        let response_trace_id_value =
            response_trace_id(&response).unwrap_or_else(|| trace_id.clone());
        tracing::warn!(
            trace_id = %response_trace_id_value,
            tenant = %tenant_id,
            method = %trace_context.method,
            path = %trace_context.path,
            status = status.as_u16(),
            origin = ?trace_context.origin,
            referer = ?trace_context.referer,
            authorization_header_present = trace_context.authorization_header_present,
            cookie_header_present = trace_context.cookie_header_present,
            session_cookie_name = %trace_context.session_cookie_name,
            session_cookie_present = trace_context.session_cookie_present,
            disable_cookie_cache = ?trace_context.disable_cookie_cache,
            set_cookie = response.headers().contains_key("set-cookie"),
            "auth request completed with session diagnostics"
        );
    }

    if let Ok(header_value) = HeaderValue::from_str(tenant_id.as_str()) {
        response
            .headers_mut()
            .insert("x-athena-tenant", header_value);
    }
    response
}

/// Applies `X-Athena-Time` response timing and records optional HTTP metrics.
async fn athena_time_header_middleware_inner(
    state: &ServerMiddlewareState,
    mut req: axum::extract::Request,
    next: Next,
) -> Response {
    let trace_id = ensure_request_trace_id(&mut req);
    let method = req.method().clone();
    let matched_path = req
        .extensions()
        .get::<MatchedPath>()
        .map(|value| value.as_str().to_string());
    let request_path = req.uri().path().to_string();
    let observed_route = resolve_observed_route(
        state,
        method.as_str(),
        matched_path.as_deref(),
        request_path.as_str(),
    );
    let tenant = observed_tenant(req.headers());
    let host = header_value_trimmed(req.headers(), "host");
    let origin = header_value_trimmed(req.headers(), "origin");
    let referer = header_value_trimmed(req.headers(), "referer");
    let user_agent = header_value_trimmed(req.headers(), "user-agent");
    let query_present = req.uri().query().is_some();
    let disable_cookie_cache = query_param_value(req.uri().query(), "disableCookieCache");
    let authorization_header_present = req.headers().contains_key("authorization");
    let cookie_header_present = req.headers().contains_key("cookie");
    let session_cookie_present =
        session_cookie_present(req.headers(), state.session_cookie_name.as_str());
    let request_body_bytes = content_length_bytes(req.headers());
    let blocked_ip = if state.blocked_ips.is_empty() {
        None
    } else {
        blocked_client_ip(state, &req)
    };
    tracing::trace!(
        request_id = %trace_id,
        trace_id = %trace_id,
        tenant = ?tenant,
        matched_path = ?matched_path,
        route = %observed_route,
        "request route metadata"
    );

    if let Some(metrics) = state.metrics.as_ref() {
        metrics.start_request(method.as_str(), observed_route.as_str());
    }

    let start = Instant::now();
    let mut response = if let Some(blocked_ip) = blocked_ip.as_deref() {
        tracing::warn!(
            request_id = %trace_id,
            trace_id = %trace_id,
            client_ip = %blocked_ip,
            method = %method,
            path = %request_path,
            matched_path = ?matched_path,
            route = %observed_route,
            "blocked request from denylisted IP before auth dispatch"
        );
        json_error_response(
            StatusCode::FORBIDDEN,
            "Access denied from this IP address",
            trace_id.as_str(),
        )
    } else {
        next.run(req).await
    };
    ensure_response_trace_headers(&mut response, trace_id.as_str());
    let response_trace_id_value = response_trace_id(&response).unwrap_or_else(|| trace_id.clone());
    let elapsed = start.elapsed();
    let elapsed_ms = elapsed.as_millis();
    let header_value = HeaderValue::from_str(&format!("{elapsed_ms}ms"))
        .unwrap_or_else(|_| HeaderValue::from_static("0ms"));
    response.headers_mut().insert("X-Athena-Time", header_value);

    let status = response.status();
    let response_body_bytes = content_length_bytes(response.headers());
    let set_cookie = response.headers().contains_key("set-cookie");
    if let Some(metrics) = state.metrics.as_ref() {
        metrics.finish_request(
            HttpRequestObservation::new(
                method.to_string(),
                observed_route.clone(),
                status.as_u16(),
                elapsed,
            )
            .with_request_body_bytes(request_body_bytes)
            .with_response_body_bytes(response_body_bytes),
        );
    }

    emit_request_wide_event(
        &state.logging,
        WideRequestEventInput {
            request_id: trace_id.as_str(),
            trace_id: response_trace_id_value.as_str(),
            method: method.as_str(),
            route: observed_route.as_str(),
            path: request_path.as_str(),
            matched_path: matched_path.as_deref(),
            host: host.as_deref(),
            tenant: tenant.as_deref(),
            origin: origin.as_deref(),
            referer: referer.as_deref(),
            user_agent: user_agent.as_deref(),
            query_present,
            disable_cookie_cache: disable_cookie_cache.as_deref(),
            status: status.as_u16(),
            duration: elapsed,
            request_body_bytes,
            response_body_bytes,
            authorization_header_present,
            cookie_header_present,
            session_cookie_name: state.session_cookie_name.as_str(),
            session_cookie_present,
            set_cookie,
            metrics_enabled: state.metrics.is_some(),
            loki_enabled: state.loki_enabled,
        },
    );
    response
}

/// Thin middleware entrypoint that injects optional metrics state.
async fn athena_time_header_and_metrics_middleware(
    State(state): State<ServerMiddlewareState>,
    req: axum::extract::Request,
    next: Next,
) -> Response {
    athena_time_header_middleware_inner(&state, req, next).await
}

/// Health/status endpoint for server liveliness checks.
async fn status_handler() -> Json<Value> {
    Json(json!({
        "status": "ok",
        "service": "athena-auth",
        "version": env!("CARGO_PKG_VERSION")
    }))
}

#[derive(Clone, Serialize)]
struct AdminDocsConfig {
    base_url: String,
    auth_base_path: String,
    docs_api_origin: String,
    openapi_url: String,
}

fn join_origin_and_path(origin: &str, path: &str) -> String {
    let normalized_origin = origin.trim_end_matches('/');
    if path == "/" {
        return normalized_origin.to_string();
    }
    format!("{normalized_origin}{path}")
}

fn join_auth_base_path(base_path: &str, route_path: &str) -> String {
    let normalized_base = base_path.trim();
    let normalized_route = if route_path.starts_with('/') {
        route_path.to_string()
    } else {
        format!("/{route_path}")
    };

    if normalized_base.is_empty() || normalized_base == "/" {
        return normalized_route;
    }

    format!(
        "{}/{}",
        normalized_base.trim_end_matches('/'),
        normalized_route.trim_start_matches('/')
    )
}

fn build_admin_docs_config(config: &AuthConfig) -> AdminDocsConfig {
    let docs_api_origin = env_var_non_empty("DOCS_API_ORIGIN")
        .unwrap_or_else(|| config.base_url.trim_end_matches('/').to_string());
    let auth_base_path = config.base_path.trim().to_string();
    let openapi_path = join_auth_base_path(config.base_path.as_str(), "/reference/openapi.json");
    let openapi_url = join_origin_and_path(docs_api_origin.as_str(), openapi_path.as_str());

    AdminDocsConfig {
        base_url: config.base_url.clone(),
        auth_base_path,
        docs_api_origin,
        openapi_url,
    }
}

async fn admin_api_config_handler(config: AdminDocsConfig) -> Json<Value> {
    Json(json!({
        "baseUrl": config.base_url,
        "authBasePath": config.auth_base_path,
        "docsApiOrigin": config.docs_api_origin,
        "openapiUrl": config.openapi_url,
    }))
}

async fn admin_docs_html_handler() -> Response {
    const HTML: &str = r#"<!doctype html>
<html lang="en">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <title>Athena Auth API Docs</title>
    <link rel="stylesheet" href="https://unpkg.com/swagger-ui-dist@5/swagger-ui.css" />
    <style>
      html, body {
        margin: 0;
        padding: 0;
        background: #f5f5f4;
      }

      body {
        color: #1c1917;
        font-family: ui-sans-serif, system-ui, sans-serif;
      }

      #swagger-ui {
        min-height: 100vh;
      }
    </style>
  </head>
  <body>
    <div id="swagger-ui"></div>
    <script src="https://unpkg.com/swagger-ui-dist@5/swagger-ui-bundle.js"></script>
    <script>
      async function boot() {
        const response = await fetch('/admin/api-config.json', {
          headers: { 'accept': 'application/json' },
          cache: 'no-store'
        });

        if (!response.ok) {
          throw new Error('Failed to load docs bootstrap config');
        }

        const config = await response.json();
        window.ui = SwaggerUIBundle({
          url: config.openapiUrl,
          dom_id: '#swagger-ui',
          deepLinking: true,
          displayRequestDuration: true,
          persistAuthorization: true,
        });
      }

      boot().catch((error) => {
        const root = document.getElementById('swagger-ui');
        if (root) {
          root.innerHTML = '<pre style="padding:24px;white-space:pre-wrap;">' +
            String(error) +
            '</pre>';
        }
      });
    </script>
  </body>
</html>
"#;

    ([("Content-Type", "text/html; charset=utf-8")], HTML).into_response()
}

fn admin_docs_config_from_registry(
    registry: &TenantAuthRegistry,
) -> Result<AdminDocsConfig, String> {
    let default_tenant = registry.default_tenant();
    let auth = registry
        .runtime(default_tenant)
        .ok_or_else(|| format!("Default tenant '{default_tenant}' is missing auth runtime"))?;
    Ok(build_admin_docs_config(auth.config()))
}

async fn maybe_handle_admin_docs_request(
    registry: &TenantAuthRegistry,
    request_path: &str,
    trace_id: &str,
) -> Option<Response> {
    match request_path {
        "/admin/docs.html" => Some(admin_docs_html_handler().await),
        "/admin/api-config.json" => match admin_docs_config_from_registry(registry) {
            Ok(config) => Some(admin_api_config_handler(config).await.into_response()),
            Err(error) => Some(
                json_error_response(StatusCode::INTERNAL_SERVER_ERROR, error, trace_id)
                    .into_response(),
            ),
        },
        _ => None,
    }
}

#[derive(Default)]
struct DeferredAppState {
    ready_app: RwLock<Option<Router>>,
    startup_error: RwLock<Option<String>>,
}

async fn deferred_app_handler(
    State(state): State<Arc<DeferredAppState>>,
    mut request: Request,
) -> Response {
    let trace_id = ensure_request_trace_id(&mut request);

    if let Some(app) = state.ready_app.read().await.clone() {
        return match app.oneshot(request).await {
            Ok(response) => response,
            Err(error) => json_error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                format!("failed to dispatch initialized app: {error}"),
                trace_id.as_str(),
            )
            .into_response(),
        };
    }

    if let Some(error) = state.startup_error.read().await.clone() {
        return json_error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("Athena Auth startup failed: {error}"),
            trace_id.as_str(),
        )
        .into_response();
    }

    let mut response = (
        StatusCode::SERVICE_UNAVAILABLE,
        Json(json!({
            "status": "starting",
            "service": "athena-auth",
            "version": env!("CARGO_PKG_VERSION"),
            "traceId": trace_id,
        })),
    )
        .into_response();
    ensure_response_trace_headers(&mut response, trace_id.as_str());
    response
}

/// Builds the live tenant schema/table mapping diagnostics payload.
async fn build_schema_debug_payload(
    access: &SchemaDebugAccessEvaluation,
    headers: &HeaderMap,
    runtime: &SchemaDebugRuntime,
) -> Value {
    let request_context = resolve_schema_debug_request_context(runtime, headers);
    let known_tenants: Vec<String> = runtime.tenants.keys().map(ToString::to_string).collect();
    let mut debug_tables: std::collections::BTreeSet<String> = LOGICAL_TABLE_NAMES
        .into_iter()
        .map(str::to_string)
        .collect();
    for tenant in runtime.tenants.values() {
        for logical_table in tenant.configured_mapping.keys() {
            debug_tables.insert(logical_table.clone());
        }
    }
    let table_alias_config: Vec<Value> = debug_tables
        .iter()
        .map(|primary_table| {
            let configured_auth_schemas: Vec<String> = runtime
                .tenants
                .values()
                .map(|tenant| tenant.configured_auth_schema.clone())
                .collect::<std::collections::BTreeSet<_>>()
                .into_iter()
                .collect();
            let secondary_table = secondary_table_name_variant(primary_table.as_str())
                .filter(|value| value != primary_table);
            let primary_is_plural = is_plural_table_name(primary_table.as_str());
            let secondary_is_plural = secondary_table
                .as_ref()
                .map(|table| is_plural_table_name(table.as_str()));
            let secondary_strategy = secondary_table.as_ref().map(|_| {
                if primary_is_plural {
                    "auto_singular"
                } else {
                    "auto_plural"
                }
            });
            let qualified_primary_table_candidates = build_lookup_candidates_for_tables(
                configured_auth_schemas.as_slice(),
                std::slice::from_ref(primary_table),
            );
            let qualified_secondary_table_candidates = secondary_table
                .as_ref()
                .map(|table| {
                    build_lookup_candidates_for_tables(
                        configured_auth_schemas.as_slice(),
                        std::slice::from_ref(table),
                    )
                })
                .unwrap_or_default();

            json!({
                "primary_table": primary_table,
                "secondary_table": secondary_table,
                "primary_is_plural": primary_is_plural,
                "secondary_is_plural": secondary_is_plural,
                "secondary_strategy": secondary_strategy,
                "configured_auth_schema": configured_auth_schemas.first().cloned(),
                "configured_auth_schemas": configured_auth_schemas,
                "qualified_primary_table_candidates": qualified_primary_table_candidates,
                "qualified_secondary_table_candidates": qualified_secondary_table_candidates,
            })
        })
        .collect();

    let tenant_configs: Vec<Value> = runtime
        .tenants
        .iter()
        .map(|(tenant_id, tenant)| {
            let mapping_entries: Vec<Value> = tenant
                .configured_mapping
                .iter()
                .map(|(logical_table, mapping)| {
                    json!({
                        "logical_table": logical_table,
                        "mapped_table": mapping.mapped_table,
                        "source": mapping.source,
                    })
                })
                .collect();

            json!({
                "tenant_id": tenant_id.to_string(),
                "is_default": tenant.is_default,
                "database_source": {
                    "kind": tenant.database_source_kind,
                    "key": tenant.database_source_key,
                    "resolved_url_redacted": tenant.database_url_redacted,
                },
                "configured_search_path": {
                    "sql_fragment": tenant.configured_search_path_sql,
                    "schemas": tenant.configured_search_path_schemas,
                },
                "configured_auth_schema": tenant.configured_auth_schema,
                "configured_mapping": {
                    "entry_count": mapping_entries.len(),
                    "entries": mapping_entries,
                }
            })
        })
        .collect();

    let resolved_tenant: Option<String> = request_context
        .resolved_tenant
        .as_ref()
        .map(ToString::to_string);
    let request_origin: Option<String> = header_value_trimmed(&headers, "origin");
    let request_referer: Option<String> = header_value_trimmed(&headers, "referer");
    let origin_rejection_debug = snapshot_origin_trust_debug(
        runtime.auth_config.as_ref(),
        request_origin.as_deref(),
        request_referer.as_deref(),
    );

    let selected_tenant: Option<TenantId> = request_context
        .resolved_tenant
        .as_ref()
        .filter(|tenant_id| runtime.tenants.contains_key(*tenant_id))
        .cloned();

    let current_tenant_probe: Value = if let Some(tenant_id) = selected_tenant {
        let tenant: Option<&SchemaDebugTenant> = runtime.tenants.get(&tenant_id);
        if let Some(tenant) = tenant {
            let mut logical_tables: std::collections::BTreeSet<String> = LOGICAL_TABLE_NAMES
                .into_iter()
                .map(str::to_string)
                .collect();
            for key in tenant.configured_mapping.keys() {
                logical_tables.insert(key.clone());
            }

            let logical_tables: Vec<String> = logical_tables.into_iter().collect();
            let mut table_names: std::collections::BTreeSet<String> =
                std::collections::BTreeSet::new();
            for logical_table in &logical_tables {
                for variant in table_name_variants_for_debug(logical_table.as_str()) {
                    table_names.insert(variant);
                }
                let mapped_table = tenant
                    .configured_mapping
                    .get(logical_table)
                    .map(|mapping| mapping.mapped_table.clone())
                    .unwrap_or_else(|| logical_table.clone());
                for variant in table_name_variants_for_debug(mapped_table.as_str()) {
                    table_names.insert(variant);
                }
            }
            let table_names: Vec<String> = table_names.into_iter().collect();

            match query_schema_debug_probe(
                tenant.database_url.as_str(),
                tenant.configured_auth_schema.as_str(),
                tenant.configured_search_path_sql.as_deref(),
                &table_names,
            )
            .await
            {
                Ok(probe) => {
                    let effective_search_path = if probe.effective_search_path.is_empty() {
                        tenant.configured_search_path_schemas.clone()
                    } else {
                        probe.effective_search_path.clone()
                    };
                    let runtime_lookup_schemas = vec![tenant.configured_auth_schema.clone()];

                    let mut runtime_ready_count: usize = 0;
                    let mut runtime_missing_count: usize = 0;
                    let mut mapped_target_ready_count: usize = 0;
                    let mut mapping_mismatch_count: usize = 0;

                    let rows: Vec<Value> = logical_tables
                        .iter()
                        .map(|logical_table| {
                            let mapping: Option<&SchemaDebugMapping> = tenant.configured_mapping.get(logical_table);
                            let mapped_table = mapping
                                .map(|value| value.mapped_table.clone())
                                .unwrap_or_else(|| logical_table.clone());
                            let mapping_source = mapping
                                .map(|value| value.source)
                                .unwrap_or("identity");
                            let runtime_lookup_table = logical_table.clone();
                            let runtime_lookup_variants =
                                table_name_variants_for_debug(runtime_lookup_table.as_str());
                            let runtime_secondary_lookup_table =
                                runtime_lookup_variants.get(1).cloned();
                            let mapped_lookup_variants =
                                table_name_variants_for_debug(mapped_table.as_str());
                            let mapped_secondary_lookup_table =
                                mapped_lookup_variants.get(1).cloned();

                            let runtime_schemas_any = probe
                                .table_schemas
                                .get(runtime_lookup_table.as_str())
                                .cloned()
                                .unwrap_or_default();
                            let runtime_secondary_schemas_any = runtime_secondary_lookup_table
                                .as_ref()
                                .and_then(|table| probe.table_schemas.get(table.as_str()).cloned())
                                .unwrap_or_default();
                            let mapped_schemas_any = probe
                                .table_schemas
                                .get(mapped_table.as_str())
                                .cloned()
                                .unwrap_or_default();
                            let mapped_secondary_schemas_any = mapped_secondary_lookup_table
                                .as_ref()
                                .and_then(|table| probe.table_schemas.get(table.as_str()).cloned())
                                .unwrap_or_default();

                            let runtime_schemas_in_search_path = schemas_in_search_path(
                                runtime_schemas_any.as_slice(),
                                effective_search_path.as_slice(),
                            );
                            let runtime_secondary_schemas_in_search_path = schemas_in_search_path(
                                runtime_secondary_schemas_any.as_slice(),
                                effective_search_path.as_slice(),
                            );
                            let mapped_schemas_in_search_path = schemas_in_search_path(
                                mapped_schemas_any.as_slice(),
                                effective_search_path.as_slice(),
                            );
                            let mapped_secondary_schemas_in_search_path = schemas_in_search_path(
                                mapped_secondary_schemas_any.as_slice(),
                                effective_search_path.as_slice(),
                            );

                            let runtime_found = !runtime_schemas_in_search_path.is_empty();
                            let runtime_secondary_found =
                                !runtime_secondary_schemas_in_search_path.is_empty();
                            let runtime_found_any = runtime_found || runtime_secondary_found;

                            let mapped_found = !mapped_schemas_in_search_path.is_empty();
                            let mapped_secondary_found =
                                !mapped_secondary_schemas_in_search_path.is_empty();
                            let mapped_found_any = mapped_found || mapped_secondary_found;
                            let runtime_anywhere = !runtime_schemas_any.is_empty();
                            let runtime_secondary_anywhere =
                                !runtime_secondary_schemas_any.is_empty();
                            let mapped_anywhere = !mapped_schemas_any.is_empty();
                            let mapped_secondary_anywhere = !mapped_secondary_schemas_any.is_empty();

                            if runtime_found_any {
                                runtime_ready_count += 1;
                            } else {
                                runtime_missing_count += 1;
                            }
                            if mapped_found_any {
                                mapped_target_ready_count += 1;
                            }
                            if mapped_table != runtime_lookup_table && !runtime_found_any && mapped_found_any {
                                mapping_mismatch_count += 1;
                            }

                            let overall_status = if runtime_found {
                                if mapped_table != runtime_lookup_table && mapped_found {
                                    "runtime_ready_mapped_also_present"
                                } else {
                                    "runtime_ready"
                                }
                            } else if runtime_secondary_found {
                                "runtime_secondary_ready_primary_missing"
                            } else if !runtime_schemas_any.is_empty() {
                                "runtime_present_outside_search_path"
                            } else if runtime_secondary_anywhere {
                                "runtime_secondary_present_outside_search_path"
                            } else if mapped_table != runtime_lookup_table && mapped_found {
                                "mapping_target_present_runtime_missing"
                            } else if mapped_table != runtime_lookup_table
                                && !mapped_schemas_any.is_empty()
                            {
                                "mapping_target_present_outside_search_path"
                            } else {
                                "runtime_missing"
                            };

                            let note = if mapped_table == runtime_lookup_table {
                                "No explicit mapping for this table; runtime and configured lookup are identical."
                                    .to_string()
                            } else {
                                format!(
                                    "Configured mapping {runtime_lookup_table} -> {mapped_table} ({mapping_source}); runtime lookup currently uses {runtime_lookup_table}."
                                )
                            };

                            let runtime_lookup_candidates: Vec<String> =
                                build_lookup_candidates_for_tables(
                                    runtime_lookup_schemas.as_slice(),
                                    runtime_lookup_variants.as_slice(),
                                );
                            let mapped_lookup_candidates: Vec<String> =
                                build_lookup_candidates_for_tables(
                                    runtime_lookup_schemas.as_slice(),
                                    mapped_lookup_variants.as_slice(),
                                );

                            json!({
                                "logical_table": logical_table,
                                "runtime_lookup_table": runtime_lookup_table,
                                "runtime_secondary_lookup_table": runtime_secondary_lookup_table,
                                "configured_mapped_table": mapped_table,
                                "configured_secondary_mapped_table": mapped_secondary_lookup_table,
                                "mapping_source": mapping_source,
                                "mapping_applied_at_runtime": false,
                                "table_name_format": {
                                    "logical_table": classify_table_name_format(logical_table.as_str()),
                                    "runtime_lookup_table": classify_table_name_format(runtime_lookup_table.as_str()),
                                    "configured_mapped_table": classify_table_name_format(mapped_table.as_str()),
                                },
                                "runtime_status": {
                                    "found_in_effective_search_path": runtime_found,
                                    "found_in_effective_search_path_primary": runtime_found,
                                    "found_in_effective_search_path_secondary": runtime_secondary_found,
                                    "found_in_effective_search_path_any": runtime_found_any,
                                    "found_anywhere_in_database_primary": runtime_anywhere,
                                    "found_anywhere_in_database_secondary": runtime_secondary_anywhere,
                                    "found_anywhere_in_database_any": runtime_anywhere || runtime_secondary_anywhere,
                                    "schemas_in_effective_search_path": runtime_schemas_in_search_path,
                                    "schemas_in_effective_search_path_secondary": runtime_secondary_schemas_in_search_path,
                                    "schemas_anywhere_in_database": runtime_schemas_any,
                                    "schemas_anywhere_in_database_secondary": runtime_secondary_schemas_any,
                                    "lookup_candidates": runtime_lookup_candidates,
                                },
                                "mapped_status": {
                                    "found_in_effective_search_path": mapped_found,
                                    "found_in_effective_search_path_primary": mapped_found,
                                    "found_in_effective_search_path_secondary": mapped_secondary_found,
                                    "found_in_effective_search_path_any": mapped_found_any,
                                    "found_anywhere_in_database_primary": mapped_anywhere,
                                    "found_anywhere_in_database_secondary": mapped_secondary_anywhere,
                                    "found_anywhere_in_database_any": mapped_anywhere || mapped_secondary_anywhere,
                                    "schemas_in_effective_search_path": mapped_schemas_in_search_path,
                                    "schemas_in_effective_search_path_secondary": mapped_secondary_schemas_in_search_path,
                                    "schemas_anywhere_in_database": mapped_schemas_any,
                                    "schemas_anywhere_in_database_secondary": mapped_secondary_schemas_any,
                                    "lookup_candidates": mapped_lookup_candidates,
                                },
                                "overall_status": overall_status,
                                "note": note,
                            })
                        })
                        .collect();

                    json!({
                        "status": "ok",
                        "tenant_id": tenant_id.to_string(),
                        "database_source": {
                            "kind": tenant.database_source_kind,
                            "key": tenant.database_source_key,
                            "resolved_url_redacted": tenant.database_url_redacted,
                        },
                        "search_path": {
                            "configured_auth_schema": tenant.configured_auth_schema,
                            "configured_sql_fragment": tenant.configured_search_path_sql,
                            "configured_schemas": tenant.configured_search_path_schemas,
                            "runtime_current_schema": probe.current_schema,
                            "runtime_effective_schemas": effective_search_path,
                            "runtime_setting_search_path": probe.search_path_setting,
                        },
                        "table_resolution_summary": {
                            "logical_table_count": rows.len(),
                            "runtime_ready_count": runtime_ready_count,
                            "runtime_missing_count": runtime_missing_count,
                            "mapped_target_ready_count": mapped_target_ready_count,
                            "mapping_mismatch_count": mapping_mismatch_count,
                        },
                        "table_resolution": rows,
                    })
                }
                Err(error) => {
                    let rows: Vec<Value> = logical_tables
                        .iter()
                        .map(|logical_table| {
                            let mapping = tenant.configured_mapping.get(logical_table);
                            let mapped_table = mapping
                                .map(|value| value.mapped_table.clone())
                                .unwrap_or_else(|| logical_table.clone());
                            let mapping_source =
                                mapping.map(|value| value.source).unwrap_or("identity");
                            let runtime_secondary_lookup_table =
                                secondary_table_name_variant(logical_table.as_str())
                                    .filter(|value| value != logical_table);
                            let mapped_secondary_lookup_table =
                                secondary_table_name_variant(mapped_table.as_str())
                                    .filter(|value| value != &mapped_table);
                            json!({
                                "logical_table": logical_table,
                                "runtime_lookup_table": logical_table,
                                "runtime_secondary_lookup_table": runtime_secondary_lookup_table,
                                "configured_mapped_table": mapped_table,
                                "configured_secondary_mapped_table": mapped_secondary_lookup_table,
                                "mapping_source": mapping_source,
                                "mapping_applied_at_runtime": false,
                                "runtime_status": {
                                    "found_in_effective_search_path": false,
                                    "found_in_effective_search_path_primary": false,
                                    "found_in_effective_search_path_secondary": false,
                                    "found_in_effective_search_path_any": false,
                                    "found_anywhere_in_database_primary": false,
                                    "found_anywhere_in_database_secondary": false,
                                    "found_anywhere_in_database_any": false,
                                },
                                "mapped_status": {
                                    "found_in_effective_search_path": false,
                                    "found_in_effective_search_path_primary": false,
                                    "found_in_effective_search_path_secondary": false,
                                    "found_in_effective_search_path_any": false,
                                    "found_anywhere_in_database_primary": false,
                                    "found_anywhere_in_database_secondary": false,
                                    "found_anywhere_in_database_any": false,
                                },
                                "overall_status": "probe_unavailable",
                                "note": format!("Probe failed before table verification: {error}"),
                            })
                        })
                        .collect();

                    json!({
                        "status": "probe_error",
                        "tenant_id": tenant_id.to_string(),
                        "error": error,
                        "table_resolution_summary": {
                            "logical_table_count": rows.len(),
                            "runtime_ready_count": 0,
                            "runtime_missing_count": rows.len(),
                            "mapped_target_ready_count": 0,
                            "mapping_mismatch_count": 0,
                        },
                        "table_resolution": rows,
                    })
                }
            }
        } else {
            json!({
                "status": "unknown_tenant",
                "error": "Resolved tenant is not registered in debug runtime.",
            })
        }
    } else {
        json!({
            "status": request_context.resolution_status,
            "error": request_context
                .resolution_error
                .clone()
                .unwrap_or_else(|| "Could not resolve a known tenant from request headers.".to_string()),
        })
    };

    let mut payload = json!({
        "status": "ok",
        "service": "athena-auth",
        "version": env!("CARGO_PKG_VERSION"),
        "debug_endpoint": DEBUG_SCHEMA_ENDPOINT_PATH,
        "generated_at": chrono::Utc::now().to_rfc3339(),
        "privileged_debug": {
            "enabled": access.privileged,
            "access_reason": access.reason,
            "query_param_present": access.query_param_present,
            "unlock_hint": DEBUG_SCHEMA_UNLOCK_HINT,
        },
        "runtime": {
            "multitenancy_enabled": runtime.multitenancy_enabled,
            "tenant_header_name": runtime.header_name,
            "default_tenant": runtime.default_tenant.to_string(),
            "known_tenants": known_tenants,
            "enabled_features": runtime.enabled_features.clone(),
            "mapping_application_mode": "config_only",
            "mapping_application_note": DEBUG_MAPPING_APPLICATION_NOTE,
            "table_alias_config": {
                "entry_count": table_alias_config.len(),
                "entries": table_alias_config,
            },
        },
        "request": {
            "configured_header_name": request_context.configured_header_name,
            "configured_header_value": request_context.configured_header_value,
            "fallback_header_values": request_context.fallback_header_values,
            "resolution_status": request_context.resolution_status,
            "resolution_error": request_context.resolution_error,
            "resolved_tenant": resolved_tenant,
        },
        "origin_rejection_debug": origin_rejection_debug,
        "tenant_configs": tenant_configs,
        "current_tenant_probe": current_tenant_probe,
    });
    if access.privileged {
        payload["privileged_debug"]["details"] = build_schema_debug_privileged_payload(runtime);
    }
    payload
}

/// Exposes live tenant schema/table mapping diagnostics for runtime debugging.
async fn schema_debug_handler(
    query: SchemaDebugQuery,
    headers: HeaderMap,
    runtime: Arc<SchemaDebugRuntime>,
) -> Json<Value> {
    let access = evaluate_schema_debug_access(
        query.jwt_secret.as_deref(),
        runtime.access_gate.jwt_secret.as_deref(),
    );
    Json(build_schema_debug_payload(&access, &headers, runtime.as_ref()).await)
}

/// Renders the schema debug diagnostics as a gated HTML page.
async fn schema_debug_html_handler(
    query: SchemaDebugQuery,
    headers: HeaderMap,
    runtime: Arc<SchemaDebugRuntime>,
) -> Response {
    let access = evaluate_schema_debug_access(
        query.jwt_secret.as_deref(),
        runtime.access_gate.jwt_secret.as_deref(),
    );

    if !access.privileged {
        return (
            StatusCode::FORBIDDEN,
            [("Content-Type", "text/html; charset=utf-8")],
            schema_debug_html::render_forbidden_schema_debug_html(
                access.reason,
                access.query_param_present,
                DEBUG_SCHEMA_UNLOCK_HINT,
            ),
        )
            .into_response();
    }

    let payload = build_schema_debug_payload(&access, &headers, runtime.as_ref()).await;
    (
        [("Content-Type", "text/html; charset=utf-8")],
        schema_debug_html::render_schema_debug_html(&payload),
    )
        .into_response()
}

/// Renders Prometheus metrics as plaintext exposition format.
async fn metrics_handler(metrics: Arc<HttpMetrics>) -> impl IntoResponse {
    match metrics.encode() {
        Ok((body, content_type)) => {
            ([("Content-Type", content_type.to_string())], body).into_response()
        }
        Err(error) => {
            tracing::error!(error = %error, "failed to render Prometheus metrics");
            (
                axum::http::StatusCode::INTERNAL_SERVER_ERROR,
                "metrics encoding error".to_string(),
            )
                .into_response()
        }
    }
}

/// Renders the human-readable metrics dashboard.
async fn metrics_dashboard_handler(
    metrics: Arc<HttpMetrics>,
    prometheus_path: String,
) -> impl IntoResponse {
    (
        [("Content-Type", "text/html; charset=utf-8")],
        metrics.render_dashboard(prometheus_path.as_str()),
    )
        .into_response()
}

/// Expands startup URLs for wildcard host bindings.
fn startup_urls(host: &str, port: u16) -> Vec<String> {
    let mut urls: Vec<String> = vec![format!("http://{host}:{port}")];
    if host == "0.0.0.0" || host == "::" {
        urls.push(format!("http://localhost:{port}"));
        urls.push(format!("http://127.0.0.1:{port}"));
    }
    urls
}

/// Loads `.env` first and then `.env.local` without overriding explicit process env.
fn load_runtime_env_files() -> Result<(), DynErr> {
    let existing_env_keys: BTreeSet<String> = env::vars().map(|(key, _)| key).collect();
    let env_layers: Vec<Vec<(String, String)>> =
        collect_runtime_env_layers(&[".env", ".env.local"])?;

    for (key, value) in layered_env_assignments(&existing_env_keys, &env_layers) {
        // SAFETY: startup runs before worker threads are spawned, so process env mutation is safe here.
        unsafe {
            env::set_var(key, value);
        }
    }

    Ok(())
}

fn collect_runtime_env_layers(paths: &[&str]) -> Result<Vec<Vec<(String, String)>>, DynErr> {
    let mut env_layers: Vec<Vec<(String, String)>> = Vec::with_capacity(paths.len());

    for path in paths {
        match dotenvy::from_filename_iter(path) {
            Ok(iter) => {
                let mut entries: Vec<(String, String)> = Vec::new();
                for item in iter {
                    let (key, value) =
                        item.map_err(|error| format!("failed to parse {path}: {error}"))?;
                    entries.push((key, value));
                }
                env_layers.push(entries);
            }
            Err(error) if error.not_found() => {}
            Err(error) => return Err(format!("failed to load {path}: {error}").into()),
        }
    }

    Ok(env_layers)
}

fn layered_env_assignments(
    existing_env_keys: &BTreeSet<String>,
    env_layers: &[Vec<(String, String)>],
) -> BTreeMap<String, String> {
    let mut assignments: BTreeMap<String, String> = BTreeMap::new();

    for layer in env_layers {
        for (key, value) in layer {
            if existing_env_keys.contains(key) {
                continue;
            }
            assignments.insert(key.clone(), value.clone());
        }
    }

    assignments
}

/// Boots runtime dependencies and starts the Axum HTTP server.
async fn run() -> Result<(), DynErr> {
    load_runtime_env_files()?;
    let config_file: Option<ServerConfigFile> = load_server_config_file()?;
    let logging_runtime =
        LoggingRuntime::from_config(config_file.as_ref().map(|cfg| &cfg.observability.logging))
            .map_err(|error| format!("invalid observability.logging config: {error}"))?;
    let env_filter: EnvFilter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new("trace,tower_http=trace"))
        .add_directive("server=trace".parse().map_err(|error| {
            format!("failed to parse tracing directive 'server=trace': {error}")
        })?)
        .add_directive("athena_auth=trace".parse().map_err(|error| {
            format!("failed to parse tracing directive 'athena_auth=trace': {error}")
        })?);
    let loki_task: Option<tokio::task::JoinHandle<()>> = init_tracing(
        env_filter,
        config_file.as_ref().map(|cfg| &cfg.observability.loki),
        &logging_runtime,
    )?;
    let observability: RuntimeObservability =
        build_observability(config_file.as_ref(), logging_runtime.clone())?;

    if config_file.is_some() {
        tracing::info!(
            config_path = %env::var("ATHENA_AUTH_CONFIG_FILE").unwrap_or_else(|_| DEFAULT_CONFIG_FILE_PATH.to_string()),
            metrics_enabled = observability.metrics.is_some(),
            loki_enabled = observability.loki_enabled,
            logging_format = observability.logging.format.as_str(),
            wide_events_enabled = observability.logging.wide_events_enabled,
            service = %observability.logging.service_name,
            service_version = %observability.logging.service_version,
            environment = %observability.logging.environment,
            region = ?observability.logging.region,
            deployment_id = ?observability.logging.deployment_id,
            instance_id = ?observability.logging.instance_id,
            commit_hash = ?observability.logging.commit_hash,
            "loaded server config from YAML file"
        );
    } else {
        tracing::info!(
            metrics_enabled = false,
            loki_enabled = false,
            logging_format = observability.logging.format.as_str(),
            wide_events_enabled = observability.logging.wide_events_enabled,
            service = %observability.logging.service_name,
            service_version = %observability.logging.service_version,
            environment = %observability.logging.environment,
            region = ?observability.logging.region,
            deployment_id = ?observability.logging.deployment_id,
            instance_id = ?observability.logging.instance_id,
            commit_hash = ?observability.logging.commit_hash,
            "no config.yaml found; observability extras disabled"
        );
    }

    let args: Args = Args::parse();

    let host: String = args
        .host
        .or_else(|| env_var_clean("HOST"))
        .unwrap_or_else(|| "0.0.0.0".to_string());

    let port: u16 = args
        .port
        .or_else(|| env_var_clean("PORT").and_then(|p| p.parse().ok()))
        .unwrap_or(3000);

    let config: AuthConfig = load_config(port, config_file.as_ref())?;
    let blocked_ips = Arc::new(load_blocked_ips(config_file.as_ref())?);
    let schema_debug_auth_config: Arc<AuthConfig> = Arc::new(config.clone());
    let database_url: String = database_url_from_env()
        .ok_or("DATABASE_URL or HYPERDRIVE_THE_ARK_OF_FLORIS must be set")?;
    let addr: SocketAddr = format!("{host}:{port}")
        .parse()
        .map_err(|e| format!("invalid host/port: {e}"))?;
    let startup_state = Arc::new(DeferredAppState::default());
    let bootstrap_app: Router = Router::new()
        .route("/", axum::routing::get(status_handler))
        .route("/health", axum::routing::get(status_handler))
        .route("/ping", axum::routing::get(status_handler))
        .fallback(any(deferred_app_handler))
        .with_state(startup_state.clone());
    let listener: TcpListener = TcpListener::bind(addr).await?;
    tracing::info!(%addr, "listening");
    let server_task = tokio::spawn(async move {
        axum::serve(
            listener,
            bootstrap_app.into_make_service_with_connect_info::<SocketAddr>(),
        )
        .await
    });

    let app_result: Result<(Router, String), DynErr> = async {
        let schema_debug_access_gate: SchemaDebugAccessGate =
            build_schema_debug_access_gate(config_file.as_ref());
        let schema_debug_runtime = build_schema_debug_runtime(
            &database_url,
            config_file.as_ref(),
            schema_debug_access_gate,
            schema_debug_auth_config,
        )?;
        if let Err(error) =
            run_schema_autodetection(config_file.as_ref(), schema_debug_runtime.as_ref()).await
        {
            tracing::warn!(
                error = %error,
                "schema auto-detection failed; server startup will continue"
            );
        }

        let base_path: String = nest_path(&config.base_path);
        let display_base: String = config.base_url.clone();
        let passkey_related_origins = resolve_passkey_related_origins(config_file.as_ref());

        tracing::info!(
            session_intelligence_enabled = config.advanced.session_intelligence.enabled,
            geoip_enabled = config.advanced.session_intelligence.geoip.enabled,
            ip_tracking_disabled = config.advanced.ip_address.disable_ip_tracking,
            trusted_proxy_headers = config.advanced.trusted_proxy_headers.len(),
            blocked_ips = blocked_ips.len(),
            "resolved auth session configuration"
        );

        tracing::info!(
            base_url = %config.base_url,
            auth_base_path = %config.base_path,
            public_auth_base_path = %config.effective_public_base_path(),
            cookie_domain = ?config.effective_session_cookie_domain(),
            trusted_origins = %config.trusted_origins.join(","),
            passkey_related_origins = %passkey_related_origins.join(","),
            "resolved auth web origin configuration"
        );

        tracing::info!(
            listen = %format!("{host}:{port}"),
            auth_base = %base_path,
            public_base = %display_base,
            "starting Athena Auth server",
        );

        let tenant_registry: Arc<TenantRuntimeRegistry<Arc<AthenaAuth<SqlxAdapter>>>> =
            build_tenant_registry(&config, &database_url, config_file.as_ref()).await?;
        let mut known_tenants: Vec<&str> = tenant_registry
            .known_tenants()
            .map(TenantId::as_str)
            .collect();
        known_tenants.sort_unstable();
        tracing::info!(
            tenant_header = %tenant_registry.header_name(),
            default_tenant = %tenant_registry.default_tenant(),
            tenants = %known_tenants.join(","),
            "initialized auth tenant registry"
        );

        tracing::info!("base_path: {:?}", base_path);

        let v07_path_check_policy = if parse_bool_env("AXUM_DISABLE_V07_PATH_CHECKS") {
            V07PathCheckPolicy::Disable
        } else {
            V07PathCheckPolicy::Enforce
        };
        let router_composer = AxumRouterComposer::new(base_path.clone())
            .with_v07_path_check_policy(v07_path_check_policy);
        tracing::info!(
            auth_mount_path = %router_composer.base_path(),
            root_mount = router_composer.is_root_mount(),
            v07_checks_disabled = matches!(v07_path_check_policy, V07PathCheckPolicy::Disable),
            "prepared auth router composition strategy"
        );

        let auth_router: Router<Arc<TenantRuntimeRegistry<Arc<AthenaAuth<SqlxAdapter>>>>> =
            router_composer
                .build_auth_router(any(dispatch_auth_request))
                .layer(axum::middleware::from_fn(
                    legacy_auth_base_path_rewrite_middleware,
                ));
        let middleware_state = ServerMiddlewareState {
            metrics: observability.metrics.clone(),
            logging: observability.logging.clone(),
            loki_enabled: observability.loki_enabled,
            auth_route_labels: build_auth_route_labels(tenant_registry.as_ref()),
            session_cookie_name: config.session.cookie_name.clone(),
            trusted_proxy_headers: config.advanced.trusted_proxy_headers.clone(),
            ip_headers: config.advanced.ip_address.headers.clone(),
            blocked_ips: blocked_ips.clone(),
        };
        let admin_docs_config = build_admin_docs_config(&config);
        let mut base_router: Router<Arc<TenantRuntimeRegistry<Arc<AthenaAuth<SqlxAdapter>>>>> =
            Router::new()
                .route("/", axum::routing::get(status_handler))
                .route("/health", axum::routing::get(status_handler))
                .route("/ping", axum::routing::get(status_handler));
        base_router = base_router.route(
            "/admin/docs.html",
            axum::routing::get(admin_docs_html_handler),
        );
        base_router = base_router.route(
            "/admin/api-config.json",
            axum::routing::get({
                let admin_docs_config = admin_docs_config.clone();
                move || {
                    let admin_docs_config = admin_docs_config.clone();
                    async move { admin_api_config_handler(admin_docs_config).await }
                }
            }),
        );
        let schema_debug_runtime_for_handler = schema_debug_runtime.clone();
        let schema_debug_runtime_for_html_handler = schema_debug_runtime.clone();
        base_router = base_router.route(
            DEBUG_SCHEMA_ENDPOINT_PATH,
            axum::routing::get(move |query: Query<SchemaDebugQuery>, headers: HeaderMap| {
                let runtime = schema_debug_runtime_for_handler.clone();
                async move { schema_debug_handler(query.0, headers, runtime).await }
            }),
        );
        tracing::info!(
            path = DEBUG_SCHEMA_ENDPOINT_PATH,
            "schema debug endpoint enabled"
        );
        base_router = base_router.route(
            DEBUG_SCHEMA_HTML_ENDPOINT_PATH,
            axum::routing::get(move |query: Query<SchemaDebugQuery>, headers: HeaderMap| {
                let runtime = schema_debug_runtime_for_html_handler.clone();
                async move { schema_debug_html_handler(query.0, headers, runtime).await }
            }),
        );
        tracing::info!(
            path = DEBUG_SCHEMA_HTML_ENDPOINT_PATH,
            "schema debug html endpoint enabled"
        );

        tracing::info!("base_router: {:#?}", base_router);

        if !passkey_related_origins.is_empty() {
            let well_known_payload = Arc::new(json!({
                "origins": passkey_related_origins,
            }));
            base_router = base_router.route(
                WELL_KNOWN_WEBAUTHN_PATH,
                axum::routing::get(move || {
                    let payload = well_known_payload.clone();
                    async move { Json((*payload).clone()) }
                }),
            );
            tracing::info!(
                path = WELL_KNOWN_WEBAUTHN_PATH,
                "WebAuthn related-origins endpoint enabled"
            );
        }

        if let (Some(metrics), Some(metrics_path)) = (
            observability.metrics.clone(),
            observability.metrics_path.clone(),
        ) {
            if metrics_path == "/" {
                return Err("observability.metrics.path cannot be '/' because '/' is reserved for server status".into());
            }
            if let Some(metrics_dashboard_path) = observability.metrics_dashboard_path.as_deref()
                && metrics_dashboard_path == metrics_path
            {
                return Err(
                    "observability.metrics.dashboard_path cannot equal observability.metrics.path"
                        .into(),
                );
            }

            tracing::info!(path = %metrics_path, "Prometheus metrics endpoint enabled");
            let metrics_for_handler: Arc<HttpMetrics> = metrics.clone();
            base_router = base_router.route(
                &metrics_path,
                axum::routing::get(move || {
                    let metrics = metrics_for_handler.clone();
                    async move { metrics_handler(metrics).await }
                }),
            );

            if let Some(metrics_dashboard_path) = observability.metrics_dashboard_path.clone() {
                if metrics_dashboard_path == "/" {
                    return Err("observability.metrics.dashboard_path cannot be '/' because '/' is reserved for server status".into());
                }

                tracing::info!(path = %metrics_dashboard_path, "metrics dashboard endpoint enabled");
                let metrics_for_handler: Arc<HttpMetrics> = metrics.clone();
                let prometheus_path = metrics_path.clone();
                base_router = base_router.route(
                    &metrics_dashboard_path,
                    axum::routing::get(move || {
                        let metrics = metrics_for_handler.clone();
                        let prometheus_path = prometheus_path.clone();
                        async move { metrics_dashboard_handler(metrics, prometheus_path).await }
                    }),
                );
            }
        }

        let composed_router: Router<Arc<TenantRuntimeRegistry<Arc<AthenaAuth<SqlxAdapter>>>>> =
            router_composer.mount_auth_router(base_router, auth_router);
        let app: Router = composed_router
            // Accept direct auth-origin requests at both `/api/auth/*` and `/*`.
            // Explicit server/admin routes still win; only misses fall through.
            .fallback(any(dispatch_auth_request))
            .layer(axum::middleware::from_fn_with_state(
                middleware_state,
                athena_time_header_and_metrics_middleware,
            ))
            .with_state(tenant_registry);

        Ok((app, base_path))
    }
    .await;

    match app_result {
        Ok((app, base_path)) => {
            *startup_state.ready_app.write().await = Some(app);
            tracing::info!("----------------------------------------");
            tracing::info!("Athena Auth server booted");
            for url in startup_urls(&host, port) {
                tracing::info!("Hosting now on {url}");
                tracing::info!("Auth routes: {url}{base_path}");
            }
            tracing::info!("----------------------------------------");
        }
        Err(error) => {
            let error_text = error.to_string();
            *startup_state.startup_error.write().await = Some(error_text.clone());
            tracing::error!(
                error = %error_text,
                "Athena Auth startup failed after bootstrap listener bound"
            );
        }
    }
    server_task
        .await
        .map_err(|error| format!("server task join error: {error}"))??;

    if let Some(task) = loki_task {
        task.abort();
    }

    Ok(())
}

/// Program entrypoint with top-level startup error handling.
#[tokio::main]
async fn main() {
    if let Err(e) = run().await {
        tracing::error!(error = %e, "server failed to start");
        eprintln!("Error: {e}");
        let _ = std::io::stdout().flush();
        let _ = std::io::stderr().flush();
        std::process::exit(1);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use athena_auth::{CreateSession, CreateUser, SessionOps, UserOps};
    use axum::body::{Body, to_bytes};
    use axum::extract::connect_info::ConnectInfo;
    use axum::http::{Request, StatusCode};
    use axum::routing::get;
    use chrono::{Duration as ChronoDuration, Utc};
    use std::collections::BTreeMap;
    use std::sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    };
    use tower::ServiceExt;

    fn test_server_middleware_state() -> ServerMiddlewareState {
        ServerMiddlewareState {
            metrics: None,
            logging: LoggingRuntime::from_config(None).expect("defaults should parse"),
            loki_enabled: false,
            auth_route_labels: AuthRouteLabelResolver::default(),
            session_cookie_name: "athena-auth.session-token".to_string(),
            trusted_proxy_headers: Vec::new(),
            ip_headers: Vec::new(),
            blocked_ips: Arc::new(BTreeSet::new()),
        }
    }

    #[test]
    fn rewrite_legacy_auth_base_path_rewrites_prefix_and_query() {
        let mut req = Request::builder()
            .method("POST")
            .uri("/api/auth/sign-in/email?disableCookieCache=true")
            .body(Body::empty())
            .expect("request should build");

        rewrite_legacy_auth_base_path(&mut req);

        assert_eq!(req.method(), axum::http::Method::POST);
        assert_eq!(req.uri().path(), "/sign-in/email");
        assert_eq!(req.uri().query(), Some("disableCookieCache=true"));
    }

    #[test]
    fn rewrite_legacy_auth_base_path_rewrites_exact_base_path_to_root() {
        let mut req = Request::builder()
            .uri("/api/auth")
            .body(Body::empty())
            .expect("request should build");

        rewrite_legacy_auth_base_path(&mut req);

        assert_eq!(req.uri().path(), "/");
        assert_eq!(req.uri().query(), None);
    }

    #[test]
    fn rewrite_legacy_auth_base_path_skips_non_matching_paths() {
        let mut req = Request::builder()
            .uri("/api/authz/sign-in/email?keep=true")
            .body(Body::empty())
            .expect("request should build");

        rewrite_legacy_auth_base_path(&mut req);

        assert_eq!(req.uri().path(), "/api/authz/sign-in/email");
        assert_eq!(req.uri().query(), Some("keep=true"));
    }

    #[test]
    fn admin_docs_config_uses_docs_api_origin_and_auth_base_path() {
        unsafe {
            std::env::set_var("DOCS_API_ORIGIN", "https://docs-api.example.com");
        }

        let config = AuthConfig::new("admin-docs-test-secret-32-characters")
            .base_url("https://auth.example.com")
            .base_path("/api/auth");
        let docs = build_admin_docs_config(&config);

        assert_eq!(docs.base_url, "https://auth.example.com");
        assert_eq!(docs.auth_base_path, "/api/auth");
        assert_eq!(docs.docs_api_origin, "https://docs-api.example.com");
        assert_eq!(
            docs.openapi_url,
            "https://docs-api.example.com/api/auth/reference/openapi.json"
        );

        unsafe {
            std::env::remove_var("DOCS_API_ORIGIN");
        }
    }

    #[tokio::test]
    async fn legacy_auth_base_path_rewrite_middleware_rewrites_request_path() {
        let app = Router::new()
            .route(
                "/{*path}",
                axum::routing::any(|req: axum::extract::Request| async move {
                    format!(
                        "{} {}{}",
                        req.method(),
                        req.uri().path(),
                        req.uri()
                            .query()
                            .map(|value| format!("?{value}"))
                            .unwrap_or_default()
                    )
                }),
            )
            .layer(axum::middleware::from_fn(
                legacy_auth_base_path_rewrite_middleware,
            ));

        let response = app
            .oneshot(
                Request::builder()
                    .method("PATCH")
                    .uri("/api/auth/session?expand=all")
                    .body(Body::empty())
                    .expect("request should build"),
            )
            .await
            .expect("request should succeed");

        assert_eq!(response.status(), StatusCode::OK);
        let body = to_bytes(response.into_body(), usize::MAX)
            .await
            .expect("body should be readable");
        let body = String::from_utf8(body.to_vec()).expect("body should be utf-8");
        assert_eq!(body, "PATCH /session?expand=all");
    }

    #[tokio::test]
    async fn fallback_dispatch_supports_root_and_prefixed_auth_paths() {
        let app =
            Router::new()
                .route("/health", get(|| async { StatusCode::OK }))
                .route(
                    "/api/auth/{*path}",
                    axum::routing::any(|req: axum::extract::Request| async move {
                        format!("prefixed {}", req.uri().path())
                    }),
                )
                .fallback(axum::routing::any(
                    |req: axum::extract::Request| async move {
                        format!("fallback {}", req.uri().path())
                    },
                ));

        let prefixed = app
            .clone()
            .oneshot(
                Request::builder()
                    .uri("/api/auth/get-session")
                    .body(Body::empty())
                    .expect("request should build"),
            )
            .await
            .expect("prefixed request should succeed");
        assert_eq!(prefixed.status(), StatusCode::OK);
        let prefixed_body = to_bytes(prefixed.into_body(), usize::MAX)
            .await
            .expect("body should be readable");
        let prefixed_body =
            String::from_utf8(prefixed_body.to_vec()).expect("body should be utf-8");
        assert_eq!(prefixed_body, "prefixed /api/auth/get-session");

        let root = app
            .clone()
            .oneshot(
                Request::builder()
                    .uri("/get-session")
                    .body(Body::empty())
                    .expect("request should build"),
            )
            .await
            .expect("root request should succeed");
        assert_eq!(root.status(), StatusCode::OK);
        let root_body = to_bytes(root.into_body(), usize::MAX)
            .await
            .expect("body should be readable");
        let root_body = String::from_utf8(root_body.to_vec()).expect("body should be utf-8");
        assert_eq!(root_body, "fallback /get-session");

        let health = app
            .oneshot(
                Request::builder()
                    .uri("/health")
                    .body(Body::empty())
                    .expect("request should build"),
            )
            .await
            .expect("health request should succeed");
        assert_eq!(health.status(), StatusCode::OK);
    }

    struct DbBackedDispatchTenantRuntime {
        auth: Arc<AthenaAuth<SqlxAdapter>>,
        admin_session_token: String,
        auth_schema: String,
    }

    fn sanitized_test_nonce() -> String {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("system time should be after unix epoch")
            .as_nanos();
        format!("{}_{}", std::process::id(), nanos)
    }

    async fn build_test_sqlx_auth_runtime() -> Arc<AthenaAuth<SqlxAdapter>> {
        let config = AuthConfig::new("server-test-secret-key-at-least-32-chars")
            .base_url("http://localhost:3000")
            .trusted_origin("http://localhost:3000");
        let options =
            PgConnectOptions::from_str("postgresql://postgres:postgres@127.0.0.1:5432/athena_auth")
                .expect("test connection URL should parse");
        let pool = PgPoolOptions::new()
            .min_connections(0)
            .max_connections(1)
            .connect_lazy_with(options);
        let adapter = SqlxAdapter::from_pool(adapter_pool_for_tests(pool));

        Arc::new(
            AthenaAuth::<SqlxAdapter>::new(config)
                .database(adapter)
                .plugin(AdminPlugin::new())
                .build()
                .await
                .expect("test auth runtime should build"),
        )
    }

    fn adapter_pool_for_tests(pool: sqlx::Pool<sqlx::Postgres>) -> sqlx::Pool<sqlx::Postgres> {
        pool
    }

    fn build_test_tenant_registry(
        header_name: &str,
        default_tenant: &str,
        runtimes: BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>>,
    ) -> Arc<TenantAuthRegistry> {
        let header = athena_auth_multitenancy::TenantHeaderName::parse(header_name)
            .expect("tenant header name should parse");
        let default = TenantId::parse(default_tenant).expect("default tenant should parse");
        let resolver = HeaderTenantResolver::new(header, default);
        Arc::new(
            TenantRuntimeRegistry::new(resolver, runtimes)
                .expect("tenant registry should include default runtime"),
        )
    }

    async fn build_db_backed_test_sqlx_auth_runtime(
        database_url: &str,
        auth_schema: &str,
        admin_email: &str,
    ) -> DbBackedDispatchTenantRuntime {
        let config = AuthConfig::new("server-test-secret-key-at-least-32-chars")
            .base_url("http://localhost:3000")
            .trusted_origin("http://localhost:3000");
        let adapter = build_sqlx_adapter(database_url, None, auth_schema)
            .await
            .expect("sqlx adapter should build for test schema");
        let auth = Arc::new(
            AthenaAuth::<SqlxAdapter>::new(config)
                .database(adapter)
                .plugin(AdminPlugin::new())
                .build()
                .await
                .expect("db-backed test auth runtime should build"),
        );

        let admin_user = auth
            .database()
            .create_user(
                CreateUser::new()
                    .with_email(admin_email.to_string())
                    .with_name("Tenant Admin")
                    .with_role("admin"),
            )
            .await
            .expect("admin user should be created");
        let admin_session = auth
            .database()
            .create_session(CreateSession {
                user_id: admin_user.id,
                expires_at: Utc::now() + ChronoDuration::hours(24),
                ip_address: None,
                user_agent: None,
                impersonated_by: None,
                active_organization_id: None,
            })
            .await
            .expect("admin session should be created");

        DbBackedDispatchTenantRuntime {
            auth,
            admin_session_token: admin_session.token,
            auth_schema: auth_schema.to_string(),
        }
    }

    async fn drop_test_schema(database_url: &str, auth_schema: &str) {
        let connect_options = parse_pg_connect_options(database_url)
            .expect("test database URL should parse for cleanup");
        let pool = PgPoolOptions::new()
            .min_connections(0)
            .max_connections(1)
            .connect_with(connect_options)
            .await
            .expect("cleanup pool should connect");
        let drop_schema_sql = format!(
            "DROP SCHEMA IF EXISTS {} CASCADE",
            quote_pg_identifier(auth_schema)
        );
        sqlx::query(drop_schema_sql.as_str())
            .execute(&pool)
            .await
            .expect("test schema should drop");
        pool.close().await;
    }

    async fn response_json(response: Response) -> serde_json::Value {
        let body = to_bytes(response.into_body(), usize::MAX)
            .await
            .expect("response body should be readable");
        serde_json::from_slice(&body).expect("response body should be valid JSON")
    }

    async fn response_text(response: Response) -> String {
        let body = to_bytes(response.into_body(), usize::MAX)
            .await
            .expect("response body should be readable");
        String::from_utf8(body.to_vec()).expect("response body should be valid UTF-8")
    }

    fn build_schema_debug_runtime_for_html_tests(secret: Option<&str>) -> Arc<SchemaDebugRuntime> {
        let default_tenant = TenantId::parse("default").expect("default tenant should parse");
        let header = athena_auth_multitenancy::TenantHeaderName::parse("x-athena-tenant")
            .expect("tenant header should parse");
        let resolver = HeaderTenantResolver::new(header, default_tenant.clone());

        let mut tenants = BTreeMap::new();
        tenants.insert(
            default_tenant.clone(),
            SchemaDebugTenant {
                is_default: true,
                database_source_kind: "fallback_database_url",
                database_source_key: None,
                database_url: "postgresql://invalid".to_string(),
                database_url_redacted: "postgresql://invalid".to_string(),
                configured_auth_schema: "athena".to_string(),
                configured_search_path_sql: None,
                configured_search_path_schemas: Vec::new(),
                configured_mapping: BTreeMap::new(),
            },
        );

        Arc::new(SchemaDebugRuntime {
            multitenancy_enabled: false,
            header_name: "x-athena-tenant".to_string(),
            default_tenant,
            resolver,
            tenants,
            access_gate: SchemaDebugAccessGate {
                jwt_secret: secret.map(str::to_string),
                config_snapshot: SchemaDebugConfigSnapshot {
                    path: "config.yaml".to_string(),
                    exists: false,
                    read_status: "not_found",
                    read_error: None,
                    raw_contents: None,
                    resolved_runtime_config: None,
                },
            },
            auth_config: Arc::new(
                AuthConfig::new("schema-debug-html-test-secret-32-chars")
                    .base_url("http://localhost:3000")
                    .trusted_origin("http://localhost:3000"),
            ),
            enabled_features: json!({}),
        })
    }

    #[tokio::test]
    async fn schema_debug_payload_reports_auth_schema_on_table_alias_entries() {
        let runtime =
            build_schema_debug_runtime_for_html_tests(Some("schema-debug-test-secret-32-chars"));
        let access = evaluate_schema_debug_access(
            Some("schema-debug-test-secret-32-chars"),
            runtime.access_gate.jwt_secret.as_deref(),
        );
        let headers = HeaderMap::new();

        let payload = build_schema_debug_payload(&access, &headers, runtime.as_ref()).await;
        let entries = payload["runtime"]["table_alias_config"]["entries"]
            .as_array()
            .expect("table alias entries should be an array");
        let accounts = entries
            .iter()
            .find(|entry| entry["primary_table"] == "accounts")
            .expect("accounts alias entry should exist");

        assert_eq!(accounts["configured_auth_schema"], "athena");
        assert_eq!(
            accounts["configured_auth_schemas"],
            serde_json::json!(["athena"])
        );
        assert_eq!(
            accounts["qualified_primary_table_candidates"],
            serde_json::json!(["athena.accounts"])
        );
        assert_eq!(
            accounts["qualified_secondary_table_candidates"],
            serde_json::json!(["athena.account"])
        );
    }

    #[tokio::test]
    async fn adds_x_athena_time_header() {
        let app: Router = Router::new()
            .route("/ping", get(|| async { "pong" }))
            .layer(axum::middleware::from_fn_with_state(
                test_server_middleware_state(),
                athena_time_header_and_metrics_middleware,
            ));

        let response = app
            .oneshot(
                Request::builder()
                    .uri("/ping")
                    .body(Body::empty())
                    .expect("test request should build"),
            )
            .await
            .expect("request should succeed");

        assert_eq!(response.status(), StatusCode::OK);

        let timing = response
            .headers()
            .get("x-athena-time")
            .and_then(|h| h.to_str().ok())
            .expect("x-athena-time header must be present and valid UTF-8");

        assert!(timing.ends_with("ms"), "timing header should be '<n>ms'");
        let millis = timing.trim_end_matches("ms");
        assert!(
            millis.parse::<u128>().is_ok(),
            "timing header must start with integer milliseconds"
        );
    }

    #[tokio::test]
    async fn middleware_blocks_denylisted_proxy_ip_before_handler() {
        let reached_handler = Arc::new(AtomicBool::new(false));
        let reached_handler_for_route = reached_handler.clone();
        let app: Router = Router::new()
            .route(
                "/ping",
                get(move || {
                    let reached_handler = reached_handler_for_route.clone();
                    async move {
                        reached_handler.store(true, Ordering::SeqCst);
                        "pong"
                    }
                }),
            )
            .layer(axum::middleware::from_fn_with_state(
                ServerMiddlewareState {
                    trusted_proxy_headers: vec!["x-real-ip".to_string()],
                    blocked_ips: Arc::new(BTreeSet::from(["198.51.100.70".to_string()])),
                    ..test_server_middleware_state()
                },
                athena_time_header_and_metrics_middleware,
            ));

        let response = app
            .oneshot(
                Request::builder()
                    .uri("/ping")
                    .header("x-real-ip", "198.51.100.70")
                    .body(Body::empty())
                    .expect("test request should build"),
            )
            .await
            .expect("request should succeed");

        assert_eq!(response.status(), StatusCode::FORBIDDEN);
        assert!(
            !reached_handler.load(Ordering::SeqCst),
            "denylisted requests must not reach the route handler"
        );
        assert!(
            response.headers().contains_key("x-athena-time"),
            "blocked responses should still include timing metadata"
        );

        let body = response_json(response).await;
        assert_eq!(body["message"], "Access denied from this IP address");
        assert!(body["traceId"].as_str().is_some());
    }

    #[tokio::test]
    async fn middleware_blocks_denylisted_socket_ip_without_proxy_headers() {
        let reached_handler = Arc::new(AtomicBool::new(false));
        let reached_handler_for_route = reached_handler.clone();
        let app: Router = Router::new()
            .route(
                "/ping",
                get(move || {
                    let reached_handler = reached_handler_for_route.clone();
                    async move {
                        reached_handler.store(true, Ordering::SeqCst);
                        "pong"
                    }
                }),
            )
            .layer(axum::middleware::from_fn_with_state(
                ServerMiddlewareState {
                    ip_headers: vec!["x-real-ip".to_string()],
                    blocked_ips: Arc::new(BTreeSet::from(["198.51.100.71".to_string()])),
                    ..test_server_middleware_state()
                },
                athena_time_header_and_metrics_middleware,
            ));

        let mut request = Request::builder()
            .uri("/ping")
            .body(Body::empty())
            .expect("test request should build");
        request
            .extensions_mut()
            .insert(ConnectInfo(SocketAddr::from(([198, 51, 100, 71], 443))));

        let response = app.oneshot(request).await.expect("request should succeed");

        assert_eq!(response.status(), StatusCode::FORBIDDEN);
        assert!(
            !reached_handler.load(Ordering::SeqCst),
            "socket-address denylist hits must not reach the route handler"
        );
    }

    #[tokio::test]
    async fn metrics_middleware_collects_request_metrics() {
        let metrics = Arc::new(
            HttpMetrics::new(Some("athena_auth_test")).expect("test metrics should initialize"),
        );
        let app = Router::new()
            .route("/ping", get(|| async { "pong" }))
            .layer(axum::middleware::from_fn_with_state(
                ServerMiddlewareState {
                    metrics: Some(metrics.clone()),
                    ..test_server_middleware_state()
                },
                athena_time_header_and_metrics_middleware,
            ));

        let response = app
            .oneshot(
                Request::builder()
                    .uri("/ping")
                    .body(Body::empty())
                    .expect("test request should build"),
            )
            .await
            .expect("request should succeed");

        assert_eq!(response.status(), StatusCode::OK);

        let (payload, _) = metrics.encode().expect("metrics should encode");
        assert!(
            payload.contains("http_requests_total"),
            "metrics output should include request counter"
        );
        assert!(
            payload.contains("http_request_duration_seconds"),
            "metrics output should include latency histogram"
        );
        assert!(
            payload.contains("http_requests_in_flight"),
            "metrics output should include in-flight gauge"
        );
        assert!(
            payload.contains("http_responses_total"),
            "metrics output should include status-class response counter"
        );
        assert!(
            metrics.render_dashboard("/metrics").contains("/ping"),
            "metrics dashboard should include observed route"
        );
    }

    #[tokio::test]
    async fn metrics_middleware_resolves_auth_catch_all_to_template() {
        let auth = AuthBuilder::new(AuthConfig::new("metrics-route-test-secret-32-characters"))
            .database(athena_auth::MemoryDatabaseAdapter::new())
            .plugin(OAuthPlugin::new())
            .build()
            .await
            .expect("auth should build");
        let mut auth_route_labels = AuthRouteLabelResolver::default();
        auth_route_labels.register_auth(&auth);

        let metrics = Arc::new(
            HttpMetrics::new(Some("athena_auth_test")).expect("test metrics should initialize"),
        );
        let app = Router::new()
            .route("/{*path}", axum::routing::get(|| async { StatusCode::OK }))
            .layer(axum::middleware::from_fn_with_state(
                ServerMiddlewareState {
                    metrics: Some(metrics.clone()),
                    auth_route_labels,
                    ..test_server_middleware_state()
                },
                athena_time_header_and_metrics_middleware,
            ));

        let response = app
            .oneshot(
                Request::builder()
                    .uri("/api/auth/callback/github")
                    .body(Body::empty())
                    .expect("test request should build"),
            )
            .await
            .expect("request should succeed");

        assert_eq!(response.status(), StatusCode::OK);

        let (payload, _) = metrics.encode().expect("metrics should encode");
        assert!(
            payload.contains("/api/auth/callback/{provider}"),
            "metrics output should use the resolved auth route template"
        );
        assert!(
            !payload.contains("/{*path}"),
            "metrics output should not use the catch-all wildcard route"
        );
    }

    #[test]
    fn parse_pg_connect_options_reports_partial_input_on_failure() {
        let error = parse_pg_connect_options("not-a-url").unwrap_err();
        assert!(error.contains("invalid PostgreSQL connection string"));
        assert!(error.contains("preview="));
        assert!(error.contains("expected_scheme=postgresql:// or postgres://"));
        assert!(error.contains("received_prefix='not-a-url'"));
    }

    #[test]
    fn parse_pg_connect_options_trims_surrounding_whitespace() {
        let options = parse_pg_connect_options(
            "  postgresql://postgres:postgres@127.0.0.1:5432/athena_auth  ",
        )
        .expect("trimmed DATABASE_URL should parse");
        assert_eq!(options.get_host(), "127.0.0.1");
    }

    #[test]
    fn parse_pg_connect_options_recovers_from_leading_noise_prefix() {
        let options =
            parse_pg_connect_options("?postgresql://postgres:postgres@127.0.0.1:5432/athena_auth")
                .expect("prefixed DATABASE_URL should parse");
        assert_eq!(options.get_host(), "127.0.0.1");
    }

    #[test]
    fn describe_database_url_for_error_reports_whitespace_and_redacted_preview() {
        let description = describe_database_url_for_error(
            "  postgresql://postgres:secret@localhost:5432/athena_auth  ",
        );
        assert!(description.contains("has_leading_or_trailing_whitespace=true"));
        assert!(description.contains("preview='postgresql://***:***@localhost:5432/athena_auth'"));
        assert!(!description.contains("secret"));
    }

    #[test]
    fn strip_leading_env_noise_recovers_postgres_scheme() {
        assert_eq!(
            strip_leading_env_noise(
                "\u{feff}\0?postgresql://postgres:postgres@localhost:5432/athena_auth"
            ),
            "postgresql://postgres:postgres@localhost:5432/athena_auth"
        );
    }

    #[test]
    fn parse_origins_csv_trims_and_deduplicates() {
        let parsed = parse_origins_csv(
            " https://one.example.com,https://two.example.com,https://one.example.com ,, ",
        );
        assert_eq!(
            parsed,
            vec![
                "https://one.example.com".to_string(),
                "https://two.example.com".to_string(),
            ]
        );
    }

    #[test]
    fn parse_blocked_ips_normalizes_and_deduplicates() {
        let parsed = parse_blocked_ips(
            vec![
                " 198.51.100.80 ".to_string(),
                "198.51.100.80:443".to_string(),
                "for=198.51.100.81;proto=https".to_string(),
            ],
            "test",
        )
        .expect("blocked IPs should parse");

        assert_eq!(
            parsed,
            BTreeSet::from(["198.51.100.80".to_string(), "198.51.100.81".to_string(),])
        );
    }

    #[test]
    fn parse_blocked_ips_rejects_non_ip_literals() {
        let error = parse_blocked_ips(vec!["edge.example.com".to_string()], "test")
            .expect_err("hostnames should be rejected");

        assert!(
            error
                .to_string()
                .contains("must resolve to an IP address literal"),
            "unexpected error: {error}"
        );
    }

    #[tokio::test]
    async fn well_known_webauthn_route_returns_origins_json() {
        let origins = vec![
            "https://shop.example.com".to_string(),
            "https://login.example.co.uk".to_string(),
        ];
        let payload = Arc::new(json!({ "origins": origins }));
        let app = Router::new().route(
            WELL_KNOWN_WEBAUTHN_PATH,
            get(move || {
                let payload = payload.clone();
                async move { Json((*payload).clone()) }
            }),
        );

        let response = app
            .oneshot(
                Request::builder()
                    .uri(WELL_KNOWN_WEBAUTHN_PATH)
                    .body(Body::empty())
                    .expect("request should build"),
            )
            .await
            .expect("request should succeed");
        assert_eq!(response.status(), StatusCode::OK);

        let body = to_bytes(response.into_body(), usize::MAX)
            .await
            .expect("body should be readable");
        let value: serde_json::Value =
            serde_json::from_slice(&body).expect("body should be valid JSON");
        assert_eq!(
            value,
            json!({
                "origins": [
                    "https://shop.example.com",
                    "https://login.example.co.uk"
                ]
            })
        );
    }

    #[tokio::test]
    async fn dispatch_auth_request_sets_tenant_header_for_known_tenant() {
        let auth_default = build_test_sqlx_auth_runtime().await;
        let auth_tenant_a = build_test_sqlx_auth_runtime().await;

        let mut runtimes: BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>> = BTreeMap::new();
        runtimes.insert(
            TenantId::parse("default").expect("default tenant should parse"),
            auth_default,
        );
        runtimes.insert(
            TenantId::parse("tenant-a").expect("tenant-a should parse"),
            auth_tenant_a,
        );

        let registry = build_test_tenant_registry("x-athena-tenant", "default", runtimes);
        let req = Request::builder()
            .method("GET")
            .uri("/admin/email/list")
            .header("x-athena-tenant", "tenant-a")
            .body(Body::empty())
            .expect("request should build");

        let response = dispatch_auth_request(State(registry), req).await;
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
        assert_eq!(
            response
                .headers()
                .get("x-athena-tenant")
                .and_then(|value| value.to_str().ok()),
            Some("tenant-a")
        );
    }

    #[tokio::test]
    async fn dispatch_auth_request_uses_fallback_tenant_header() {
        let auth_default = build_test_sqlx_auth_runtime().await;
        let auth_tenant_a = build_test_sqlx_auth_runtime().await;

        let mut runtimes: BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>> = BTreeMap::new();
        runtimes.insert(
            TenantId::parse("default").expect("default tenant should parse"),
            auth_default,
        );
        runtimes.insert(
            TenantId::parse("tenant-a").expect("tenant-a should parse"),
            auth_tenant_a,
        );

        let registry = build_test_tenant_registry("x-custom-tenant", "default", runtimes);
        let req = Request::builder()
            .method("GET")
            .uri("/admin/email/list")
            .header("x-athena-client", "tenant-a")
            .body(Body::empty())
            .expect("request should build");

        let response = dispatch_auth_request(State(registry), req).await;
        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
        assert_eq!(
            response
                .headers()
                .get("x-athena-tenant")
                .and_then(|value| value.to_str().ok()),
            Some("tenant-a")
        );
    }

    #[tokio::test]
    async fn dispatch_auth_request_rejects_unknown_tenant() {
        let auth_default = build_test_sqlx_auth_runtime().await;

        let mut runtimes: BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>> = BTreeMap::new();
        runtimes.insert(
            TenantId::parse("default").expect("default tenant should parse"),
            auth_default,
        );

        let registry = build_test_tenant_registry("x-athena-tenant", "default", runtimes);
        let req = Request::builder()
            .method("GET")
            .uri("/admin/email/list")
            .header("x-athena-tenant", "tenant-z")
            .body(Body::empty())
            .expect("request should build");

        let response = dispatch_auth_request(State(registry), req).await;
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn dispatch_auth_request_rejects_invalid_tenant_header() {
        let auth_default = build_test_sqlx_auth_runtime().await;

        let mut runtimes: BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>> = BTreeMap::new();
        runtimes.insert(
            TenantId::parse("default").expect("default tenant should parse"),
            auth_default,
        );

        let registry = build_test_tenant_registry("x-athena-tenant", "default", runtimes);
        let req = Request::builder()
            .method("GET")
            .uri("/admin/email/list")
            .header("x-athena-tenant", "tenant bad !")
            .body(Body::empty())
            .expect("request should build");

        let response = dispatch_auth_request(State(registry), req).await;
        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn auth_dispatch_404_response_reuses_generated_request_trace_id() {
        let auth_default = build_test_sqlx_auth_runtime().await;

        let mut runtimes: BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>> = BTreeMap::new();
        runtimes.insert(
            TenantId::parse("default").expect("default tenant should parse"),
            auth_default,
        );

        let registry = build_test_tenant_registry("x-athena-tenant", "default", runtimes);
        let app: Router = Router::new()
            .route("/{*path}", any(dispatch_auth_request))
            .layer(axum::middleware::from_fn_with_state(
                ServerMiddlewareState {
                    auth_route_labels: build_auth_route_labels(registry.as_ref()),
                    ..test_server_middleware_state()
                },
                athena_time_header_and_metrics_middleware,
            ))
            .with_state(registry);

        let response = app
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/definitely-missing")
                    .body(Body::empty())
                    .expect("request should build"),
            )
            .await
            .expect("request should succeed");

        assert_eq!(response.status(), StatusCode::NOT_FOUND);
        let request_id = response
            .headers()
            .get("x-request-id")
            .and_then(|value| value.to_str().ok())
            .expect("response should expose x-request-id")
            .to_string();
        assert_eq!(
            response
                .headers()
                .get("x-athena-trace-id")
                .and_then(|value| value.to_str().ok()),
            Some(request_id.as_str())
        );

        let body = response_json(response).await;
        assert_eq!(body["traceId"], request_id);
        assert_eq!(body["message"], "No handler found for this request");
    }

    #[tokio::test]
    async fn dispatch_auth_request_persists_email_failure_in_resolved_tenant_schema() {
        let Some(database_url) = std::env::var("TEST_DATABASE_URL")
            .ok()
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty())
        else {
            println!("Skipping dispatch persistence test - TEST_DATABASE_URL not set");
            return;
        };

        let nonce = sanitized_test_nonce();
        let default_schema = format!("tenant_dispatch_default_{nonce}");
        let tenant_a_schema = format!("tenant_dispatch_a_{nonce}");
        let default_admin_email = format!("default-admin-{nonce}@example.com");
        let tenant_a_admin_email = format!("tenant-a-admin-{nonce}@example.com");

        let default_runtime = build_db_backed_test_sqlx_auth_runtime(
            database_url.as_str(),
            default_schema.as_str(),
            default_admin_email.as_str(),
        )
        .await;
        let tenant_a_runtime = build_db_backed_test_sqlx_auth_runtime(
            database_url.as_str(),
            tenant_a_schema.as_str(),
            tenant_a_admin_email.as_str(),
        )
        .await;

        let mut runtimes: BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>> = BTreeMap::new();
        runtimes.insert(
            TenantId::parse("default").expect("default tenant should parse"),
            default_runtime.auth.clone(),
        );
        runtimes.insert(
            TenantId::parse("tenant-a").expect("tenant-a should parse"),
            tenant_a_runtime.auth.clone(),
        );
        let registry = build_test_tenant_registry("x-athena-tenant", "default", runtimes);

        let create_payload = serde_json::json!({
            "recipientEmail": "tenant-a@example.com",
            "flow": "tenant_dispatch_flow",
            "provider": "smtp-tenant-a",
            "errorMessage": "smtp timeout",
            "metadata": { "scope": "tenant-a" }
        });
        let create_req = Request::builder()
            .method("POST")
            .uri("/admin/email-failure/create")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::from(create_payload.to_string()))
            .expect("create request should build");
        let create_resp = dispatch_auth_request(State(registry.clone()), create_req).await;
        assert_eq!(create_resp.status(), StatusCode::OK);
        assert_eq!(
            create_resp
                .headers()
                .get("x-athena-tenant")
                .and_then(|value| value.to_str().ok()),
            Some("tenant-a")
        );
        let create_json = response_json(create_resp).await;
        let created_failure_id = create_json
            .pointer("/emailSendFailure/id")
            .and_then(Value::as_str)
            .expect("create response should include emailSendFailure.id")
            .to_string();

        let tenant_list_req = Request::builder()
            .method("GET")
            .uri("/admin/email-failure/list?flow=tenant_dispatch_flow")
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::empty())
            .expect("tenant list request should build");
        let tenant_list_resp =
            dispatch_auth_request(State(registry.clone()), tenant_list_req).await;
        assert_eq!(tenant_list_resp.status(), StatusCode::OK);
        let tenant_list_json = response_json(tenant_list_resp).await;
        let tenant_failures = tenant_list_json
            .pointer("/emailSendFailures")
            .and_then(Value::as_array)
            .expect("tenant list response should include emailSendFailures");
        assert_eq!(tenant_failures.len(), 1);
        assert_eq!(
            tenant_failures[0]["id"],
            Value::String(created_failure_id.clone())
        );
        assert_eq!(tenant_list_json["total"], serde_json::json!(1));

        let default_list_req = Request::builder()
            .method("GET")
            .uri("/admin/email-failure/list?flow=tenant_dispatch_flow")
            .header(
                "authorization",
                format!("Bearer {}", default_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "default")
            .body(Body::empty())
            .expect("default list request should build");
        let default_list_resp =
            dispatch_auth_request(State(registry.clone()), default_list_req).await;
        assert_eq!(default_list_resp.status(), StatusCode::OK);
        let default_list_json = response_json(default_list_resp).await;
        let default_failures = default_list_json
            .pointer("/emailSendFailures")
            .and_then(Value::as_array)
            .expect("default list response should include emailSendFailures");
        assert_eq!(default_failures.len(), 0);
        assert_eq!(default_list_json["total"], serde_json::json!(0));

        let tenant_audit_req = Request::builder()
            .method("GET")
            .uri(format!(
                "/admin/audit-log/list?action=admin.email_failure.create&targetType=email_send_failure&targetId={created_failure_id}&success=true"
            ))
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::empty())
            .expect("tenant audit request should build");
        let tenant_audit_resp =
            dispatch_auth_request(State(registry.clone()), tenant_audit_req).await;
        assert_eq!(tenant_audit_resp.status(), StatusCode::OK);
        let tenant_audit_json = response_json(tenant_audit_resp).await;
        let tenant_audit_logs = tenant_audit_json
            .pointer("/auditLogs")
            .and_then(Value::as_array)
            .expect("tenant audit response should include auditLogs");
        assert_eq!(tenant_audit_logs.len(), 1);
        assert_eq!(tenant_audit_json["total"], serde_json::json!(1));
        assert_eq!(tenant_audit_logs[0]["metadata"]["tenant"], "tenant-a");

        let default_audit_req = Request::builder()
            .method("GET")
            .uri(format!(
                "/admin/audit-log/list?action=admin.email_failure.create&targetType=email_send_failure&targetId={created_failure_id}&success=true"
            ))
            .header(
                "authorization",
                format!("Bearer {}", default_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "default")
            .body(Body::empty())
            .expect("default audit request should build");
        let default_audit_resp = dispatch_auth_request(State(registry), default_audit_req).await;
        assert_eq!(default_audit_resp.status(), StatusCode::OK);
        let default_audit_json = response_json(default_audit_resp).await;
        let default_audit_logs = default_audit_json
            .pointer("/auditLogs")
            .and_then(Value::as_array)
            .expect("default audit response should include auditLogs");
        assert_eq!(default_audit_logs.len(), 0);
        assert_eq!(default_audit_json["total"], serde_json::json!(0));

        tenant_a_runtime.auth.database().close().await;
        default_runtime.auth.database().close().await;
        drop_test_schema(database_url.as_str(), tenant_a_runtime.auth_schema.as_str()).await;
        drop_test_schema(database_url.as_str(), default_runtime.auth_schema.as_str()).await;
    }

    #[tokio::test]
    async fn dispatch_auth_request_enforces_tenant_isolation_for_sent_email_and_templates() {
        let Some(database_url) = std::env::var("TEST_DATABASE_URL")
            .ok()
            .map(|value| value.trim().to_string())
            .filter(|value| !value.is_empty())
        else {
            println!("Skipping dispatch email/template isolation test - TEST_DATABASE_URL not set");
            return;
        };

        let nonce = sanitized_test_nonce();
        let default_schema = format!("tenant_dispatch_crud_default_{nonce}");
        let tenant_a_schema = format!("tenant_dispatch_crud_a_{nonce}");
        let default_admin_email = format!("crud-default-admin-{nonce}@example.com");
        let tenant_a_admin_email = format!("crud-tenant-a-admin-{nonce}@example.com");
        let template_key = format!("tenant_dispatch_template_{nonce}");

        let default_runtime = build_db_backed_test_sqlx_auth_runtime(
            database_url.as_str(),
            default_schema.as_str(),
            default_admin_email.as_str(),
        )
        .await;
        let tenant_a_runtime = build_db_backed_test_sqlx_auth_runtime(
            database_url.as_str(),
            tenant_a_schema.as_str(),
            tenant_a_admin_email.as_str(),
        )
        .await;

        let mut runtimes: BTreeMap<TenantId, Arc<AthenaAuth<SqlxAdapter>>> = BTreeMap::new();
        runtimes.insert(
            TenantId::parse("default").expect("default tenant should parse"),
            default_runtime.auth.clone(),
        );
        runtimes.insert(
            TenantId::parse("tenant-a").expect("tenant-a should parse"),
            tenant_a_runtime.auth.clone(),
        );
        let registry = build_test_tenant_registry("x-athena-tenant", "default", runtimes);

        let create_email_req = Request::builder()
            .method("POST")
            .uri("/admin/email/create")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::from(
                serde_json::json!({
                    "recipientEmail": "tenant-a@example.com",
                    "subject": "Tenant A Subject",
                    "fromAddress": "noreply@example.com",
                    "fromName": "Tenant A",
                    "textBody": "Tenant A Body",
                    "provider": "smtp-tenant-a",
                    "flow": "tenant_dispatch_crud_flow",
                    "metadata": { "scope": "tenant-a", "op": "create-email" }
                })
                .to_string(),
            ))
            .expect("create email request should build");
        let create_email_resp =
            dispatch_auth_request(State(registry.clone()), create_email_req).await;
        assert_eq!(create_email_resp.status(), StatusCode::OK);
        let create_email_json = response_json(create_email_resp).await;
        let sent_email_id = create_email_json
            .pointer("/email/id")
            .and_then(Value::as_str)
            .expect("create email response should include email.id")
            .to_string();

        let get_tenant_email_req = Request::builder()
            .method("GET")
            .uri(format!("/admin/email/get?id={sent_email_id}"))
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::empty())
            .expect("tenant get email request should build");
        let get_tenant_email_resp =
            dispatch_auth_request(State(registry.clone()), get_tenant_email_req).await;
        assert_eq!(get_tenant_email_resp.status(), StatusCode::OK);
        let get_tenant_email_json = response_json(get_tenant_email_resp).await;
        assert_eq!(get_tenant_email_json["email"]["id"], sent_email_id);

        let get_default_email_req = Request::builder()
            .method("GET")
            .uri(format!("/admin/email/get?id={sent_email_id}"))
            .header(
                "authorization",
                format!("Bearer {}", default_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "default")
            .body(Body::empty())
            .expect("default get email request should build");
        let get_default_email_resp =
            dispatch_auth_request(State(registry.clone()), get_default_email_req).await;
        assert_eq!(get_default_email_resp.status(), StatusCode::NOT_FOUND);

        let update_tenant_email_req = Request::builder()
            .method("POST")
            .uri("/admin/email/update")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::from(
                serde_json::json!({
                    "id": sent_email_id,
                    "subject": "Tenant A Subject Updated",
                    "metadata": { "scope": "tenant-a", "op": "update-email" }
                })
                .to_string(),
            ))
            .expect("tenant update email request should build");
        let update_tenant_email_resp =
            dispatch_auth_request(State(registry.clone()), update_tenant_email_req).await;
        assert_eq!(update_tenant_email_resp.status(), StatusCode::OK);
        let update_tenant_email_json = response_json(update_tenant_email_resp).await;
        assert_eq!(
            update_tenant_email_json["email"]["subject"],
            "Tenant A Subject Updated"
        );

        let update_default_email_req = Request::builder()
            .method("POST")
            .uri("/admin/email/update")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", default_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "default")
            .body(Body::from(
                serde_json::json!({
                    "id": sent_email_id,
                    "subject": "Wrong Tenant Update"
                })
                .to_string(),
            ))
            .expect("default update email request should build");
        let update_default_email_resp =
            dispatch_auth_request(State(registry.clone()), update_default_email_req).await;
        assert_eq!(update_default_email_resp.status(), StatusCode::NOT_FOUND);

        let create_template_req = Request::builder()
            .method("POST")
            .uri("/admin/email-template/create")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::from(
                serde_json::json!({
                    "templateKey": template_key,
                    "locale": "en",
                    "subjectTemplate": "Hello {{name}}",
                    "textTemplate": "Hi {{name}}",
                    "variables": ["name"],
                    "isActive": true,
                    "metadata": { "scope": "tenant-a", "op": "create-template" }
                })
                .to_string(),
            ))
            .expect("create template request should build");
        let create_template_resp =
            dispatch_auth_request(State(registry.clone()), create_template_req).await;
        assert_eq!(create_template_resp.status(), StatusCode::OK);
        let create_template_json = response_json(create_template_resp).await;
        let template_id = create_template_json
            .pointer("/id")
            .and_then(Value::as_str)
            .expect("create template response should include id")
            .to_string();

        let list_tenant_templates_req = Request::builder()
            .method("GET")
            .uri(format!(
                "/admin/email-template/list?templateKey={template_key}"
            ))
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::empty())
            .expect("tenant list template request should build");
        let list_tenant_templates_resp =
            dispatch_auth_request(State(registry.clone()), list_tenant_templates_req).await;
        assert_eq!(list_tenant_templates_resp.status(), StatusCode::OK);
        let list_tenant_templates_json = response_json(list_tenant_templates_resp).await;
        let tenant_templates = list_tenant_templates_json
            .pointer("/emailTemplates")
            .and_then(Value::as_array)
            .expect("tenant template response should include emailTemplates");
        assert_eq!(tenant_templates.len(), 1);
        assert_eq!(list_tenant_templates_json["total"], serde_json::json!(1));
        assert_eq!(
            tenant_templates[0]["id"],
            Value::String(template_id.clone())
        );

        let list_default_templates_req = Request::builder()
            .method("GET")
            .uri(format!(
                "/admin/email-template/list?templateKey={template_key}"
            ))
            .header(
                "authorization",
                format!("Bearer {}", default_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "default")
            .body(Body::empty())
            .expect("default list template request should build");
        let list_default_templates_resp =
            dispatch_auth_request(State(registry.clone()), list_default_templates_req).await;
        assert_eq!(list_default_templates_resp.status(), StatusCode::OK);
        let list_default_templates_json = response_json(list_default_templates_resp).await;
        let default_templates = list_default_templates_json
            .pointer("/emailTemplates")
            .and_then(Value::as_array)
            .expect("default template response should include emailTemplates");
        assert_eq!(default_templates.len(), 0);
        assert_eq!(list_default_templates_json["total"], serde_json::json!(0));

        let update_tenant_template_req = Request::builder()
            .method("POST")
            .uri("/admin/email-template/update")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::from(
                serde_json::json!({
                    "id": template_id,
                    "subjectTemplate": "Hello Updated {{name}}",
                    "metadata": { "scope": "tenant-a", "op": "update-template" }
                })
                .to_string(),
            ))
            .expect("tenant update template request should build");
        let update_tenant_template_resp =
            dispatch_auth_request(State(registry.clone()), update_tenant_template_req).await;
        assert_eq!(update_tenant_template_resp.status(), StatusCode::OK);
        let update_tenant_template_json = response_json(update_tenant_template_resp).await;
        assert_eq!(
            update_tenant_template_json["subjectTemplate"],
            "Hello Updated {{name}}"
        );

        let update_default_template_req = Request::builder()
            .method("POST")
            .uri("/admin/email-template/update")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", default_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "default")
            .body(Body::from(
                serde_json::json!({
                    "id": template_id,
                    "subjectTemplate": "Wrong Tenant Template Update"
                })
                .to_string(),
            ))
            .expect("default update template request should build");
        let update_default_template_resp =
            dispatch_auth_request(State(registry.clone()), update_default_template_req).await;
        assert_eq!(update_default_template_resp.status(), StatusCode::NOT_FOUND);

        let tenant_email_audit_req = Request::builder()
            .method("GET")
            .uri(format!(
                "/admin/audit-log/list?action=admin.email.create&targetType=email&targetId={sent_email_id}&success=true"
            ))
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::empty())
            .expect("tenant email audit request should build");
        let tenant_email_audit_resp =
            dispatch_auth_request(State(registry.clone()), tenant_email_audit_req).await;
        assert_eq!(tenant_email_audit_resp.status(), StatusCode::OK);
        let tenant_email_audit_json = response_json(tenant_email_audit_resp).await;
        assert_eq!(tenant_email_audit_json["total"], serde_json::json!(1));

        let default_email_audit_req = Request::builder()
            .method("GET")
            .uri(format!(
                "/admin/audit-log/list?action=admin.email.create&targetType=email&targetId={sent_email_id}&success=true"
            ))
            .header(
                "authorization",
                format!("Bearer {}", default_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "default")
            .body(Body::empty())
            .expect("default email audit request should build");
        let default_email_audit_resp =
            dispatch_auth_request(State(registry.clone()), default_email_audit_req).await;
        assert_eq!(default_email_audit_resp.status(), StatusCode::OK);
        let default_email_audit_json = response_json(default_email_audit_resp).await;
        assert_eq!(default_email_audit_json["total"], serde_json::json!(0));

        let tenant_template_audit_req = Request::builder()
            .method("GET")
            .uri(format!(
                "/admin/audit-log/list?action=admin.email_template.create&targetType=email_template&targetId={template_id}&success=true"
            ))
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::empty())
            .expect("tenant template audit request should build");
        let tenant_template_audit_resp =
            dispatch_auth_request(State(registry.clone()), tenant_template_audit_req).await;
        assert_eq!(tenant_template_audit_resp.status(), StatusCode::OK);
        let tenant_template_audit_json = response_json(tenant_template_audit_resp).await;
        assert_eq!(tenant_template_audit_json["total"], serde_json::json!(1));

        let default_template_audit_req = Request::builder()
            .method("GET")
            .uri(format!(
                "/admin/audit-log/list?action=admin.email_template.create&targetType=email_template&targetId={template_id}&success=true"
            ))
            .header(
                "authorization",
                format!("Bearer {}", default_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "default")
            .body(Body::empty())
            .expect("default template audit request should build");
        let default_template_audit_resp =
            dispatch_auth_request(State(registry.clone()), default_template_audit_req).await;
        assert_eq!(default_template_audit_resp.status(), StatusCode::OK);
        let default_template_audit_json = response_json(default_template_audit_resp).await;
        assert_eq!(default_template_audit_json["total"], serde_json::json!(0));

        let delete_email_req = Request::builder()
            .method("POST")
            .uri("/admin/email/delete")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::from(
                serde_json::json!({ "id": sent_email_id }).to_string(),
            ))
            .expect("delete email request should build");
        let delete_email_resp =
            dispatch_auth_request(State(registry.clone()), delete_email_req).await;
        assert_eq!(delete_email_resp.status(), StatusCode::OK);

        let delete_template_req = Request::builder()
            .method("POST")
            .uri("/admin/email-template/delete")
            .header("content-type", "application/json")
            .header(
                "authorization",
                format!("Bearer {}", tenant_a_runtime.admin_session_token),
            )
            .header("x-athena-tenant", "tenant-a")
            .body(Body::from(
                serde_json::json!({ "id": template_id }).to_string(),
            ))
            .expect("delete template request should build");
        let delete_template_resp =
            dispatch_auth_request(State(registry.clone()), delete_template_req).await;
        assert_eq!(delete_template_resp.status(), StatusCode::OK);

        tenant_a_runtime.auth.database().close().await;
        default_runtime.auth.database().close().await;
        drop_test_schema(database_url.as_str(), tenant_a_runtime.auth_schema.as_str()).await;
        drop_test_schema(database_url.as_str(), default_runtime.auth_schema.as_str()).await;
    }

    #[test]
    fn normalizes_metrics_path() {
        let raw = "metrics/";
        let normalized = normalize_route_path(raw, DEFAULT_METRICS_PATH);
        assert_eq!(normalized, "/metrics");
    }

    #[test]
    fn observed_tenant_prefers_primary_header_then_fallbacks() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "x-athena-client",
            HeaderValue::from_static("tenant-from-fallback"),
        );
        assert_eq!(
            observed_tenant(&headers),
            Some("tenant-from-fallback".to_string())
        );

        headers.insert(
            "x-athena-tenant",
            HeaderValue::from_static("tenant-from-primary"),
        );
        assert_eq!(
            observed_tenant(&headers),
            Some("tenant-from-primary".to_string())
        );
    }

    #[tokio::test]
    async fn schema_debug_html_handler_requires_matching_jwt_secret() {
        let runtime =
            build_schema_debug_runtime_for_html_tests(Some("super-secret-32-chars-minimum-value"));
        let app = Router::new().route(
            DEBUG_SCHEMA_HTML_ENDPOINT_PATH,
            get(move |query: Query<SchemaDebugQuery>, headers: HeaderMap| {
                let runtime = runtime.clone();
                async move { schema_debug_html_handler(query.0, headers, runtime).await }
            }),
        );

        let response = app
            .oneshot(
                Request::builder()
                    .uri(DEBUG_SCHEMA_HTML_ENDPOINT_PATH)
                    .body(Body::empty())
                    .expect("request should build"),
            )
            .await
            .expect("request should succeed");

        assert_eq!(response.status(), StatusCode::FORBIDDEN);
        assert_eq!(
            response
                .headers()
                .get("content-type")
                .and_then(|value| value.to_str().ok()),
            Some("text/html; charset=utf-8")
        );

        let body = response_text(response).await;
        assert!(body.contains("Schema debug HTML is locked"));
        assert!(body.contains("Pass ?jwt_secret="));
        assert!(body.contains("JWT_SECRET"));
    }

    #[tokio::test]
    async fn schema_debug_html_handler_returns_html_when_unlocked() {
        let runtime =
            build_schema_debug_runtime_for_html_tests(Some("super-secret-32-chars-minimum-value"));
        let app = Router::new().route(
            DEBUG_SCHEMA_HTML_ENDPOINT_PATH,
            get(move |query: Query<SchemaDebugQuery>, headers: HeaderMap| {
                let runtime = runtime.clone();
                async move { schema_debug_html_handler(query.0, headers, runtime).await }
            }),
        );

        let response = app
            .oneshot(
                Request::builder()
                    .uri(format!(
                        "{DEBUG_SCHEMA_HTML_ENDPOINT_PATH}?jwt_secret=super-secret-32-chars-minimum-value"
                    ))
                    .body(Body::empty())
                    .expect("request should build"),
            )
            .await
            .expect("request should succeed");

        assert_eq!(response.status(), StatusCode::OK);
        assert_eq!(
            response
                .headers()
                .get("content-type")
                .and_then(|value| value.to_str().ok()),
            Some("text/html; charset=utf-8")
        );

        let body = response_text(response).await;
        assert!(body.contains("Athena Auth Schema Debug"));
        assert!(body.contains("Table resolution"));
        assert!(body.contains("Raw JSON"));
        assert!(body.contains("probe_error"));
    }

    #[test]
    fn schema_debug_access_grants_privileged_payload_on_exact_secret_match() {
        let access = evaluate_schema_debug_access(
            Some("super-secret-32-chars-minimum-value"),
            Some("super-secret-32-chars-minimum-value"),
        );
        assert!(access.privileged);
        assert_eq!(access.reason, "jwt_secret_query_matches_env");
        assert!(access.query_param_present);
    }

    #[test]
    fn schema_debug_access_rejects_missing_query_secret() {
        let access =
            evaluate_schema_debug_access(None, Some("super-secret-32-chars-minimum-value"));
        assert!(!access.privileged);
        assert_eq!(access.reason, "jwt_secret_query_missing");
        assert!(!access.query_param_present);
    }

    #[test]
    fn schema_debug_access_rejects_mismatched_query_secret() {
        let access = evaluate_schema_debug_access(
            Some("wrong-secret"),
            Some("super-secret-32-chars-minimum-value"),
        );
        assert!(!access.privileged);
        assert_eq!(access.reason, "jwt_secret_query_mismatch");
        assert!(access.query_param_present);
    }

    #[test]
    fn schema_debug_access_accepts_wrapped_query_secret() {
        let access = evaluate_schema_debug_access(
            Some("\"super-secret-32-chars-minimum-value\""),
            Some("super-secret-32-chars-minimum-value"),
        );
        assert!(access.privileged);
        assert_eq!(access.reason, "jwt_secret_query_matches_env");
        assert!(access.query_param_present);
    }

    #[test]
    fn schema_debug_access_accepts_plus_preserved_secret_from_query_space() {
        let access =
            evaluate_schema_debug_access(Some("super secret+value"), Some("super+secret+value"));
        assert!(access.privileged);
        assert_eq!(access.reason, "jwt_secret_query_matches_env");
        assert!(access.query_param_present);
    }

    #[test]
    fn schema_debug_oauth_resolution_treats_trusted_app_callback_as_post_login_redirect() {
        let auth_config = AuthConfig::new("super-secret-32-chars-minimum-value")
            .base_url("https://xbp.app")
            .public_base_path("/api")
            .trusted_origins(vec!["https://xbp.app".to_string()]);
        let resolution = simulate_oauth_callback_url_resolution(
            &auth_config,
            "github",
            "https://xbp.app/session",
        );
        assert_eq!(
            resolution.get("outcome").and_then(Value::as_str),
            Some("trusted_app_callback")
        );
        assert_eq!(
            resolution.get("oauth_redirect_uri").and_then(Value::as_str),
            Some("https://xbp.app/api/callback/github")
        );
        assert_eq!(
            resolution.get("app_callback_url").and_then(Value::as_str),
            Some("https://xbp.app/session")
        );
    }

    #[test]
    fn schema_debug_oauth_resolution_accepts_configured_provider_redirect_uri() {
        let auth_config = AuthConfig::new("super-secret-32-chars-minimum-value")
            .base_url("https://xbp.app")
            .public_base_path("/api");
        let resolution = simulate_oauth_callback_url_resolution(
            &auth_config,
            "github",
            "https://xbp.app/api/callback/github",
        );
        assert_eq!(
            resolution.get("outcome").and_then(Value::as_str),
            Some("oauth_redirect_uri")
        );
        assert_eq!(
            resolution.get("oauth_redirect_uri").and_then(Value::as_str),
            Some("https://xbp.app/api/callback/github")
        );
    }

    #[test]
    fn schema_debug_oauth_resolution_rejects_untrusted_callback_url() {
        let auth_config = AuthConfig::new("super-secret-32-chars-minimum-value")
            .base_url("https://xbp.app")
            .public_base_path("/api")
            .trusted_origins(vec!["https://xbp.app".to_string()]);
        let resolution = simulate_oauth_callback_url_resolution(
            &auth_config,
            "github",
            "https://evil.example/session",
        );
        assert_eq!(
            resolution.get("outcome").and_then(Value::as_str),
            Some("rejected")
        );
    }

    #[test]
    fn parses_search_path_fragment_for_debug_output() {
        let schemas = parse_search_path_schemas(Some("\"tenant_a\", \"public\""));
        assert_eq!(schemas, vec!["tenant_a".to_string(), "public".to_string()]);
    }

    #[test]
    fn normalizes_mapping_for_debug_output() {
        let mut raw = std::collections::BTreeMap::new();
        raw.insert(" users ".to_string(), " tenant_users ".to_string());
        raw.insert("".to_string(), "ignored".to_string());

        let normalized = normalize_mapping_for_debug(raw);
        assert_eq!(normalized.get("users"), Some(&"tenant_users".to_string()));
        assert_eq!(normalized.len(), 1);
    }

    #[test]
    fn derives_secondary_table_variants_for_debug() {
        assert_eq!(
            secondary_table_name_variant("users"),
            Some("user".to_string())
        );
        assert_eq!(
            secondary_table_name_variant("passkeys"),
            Some("passkey".to_string())
        );
        assert_eq!(
            secondary_table_name_variant("organization"),
            Some("organizations".to_string())
        );
        assert_eq!(
            secondary_table_name_variant("member"),
            Some("members".to_string())
        );
    }

    #[test]
    fn table_variants_include_primary_then_secondary() {
        assert_eq!(
            table_name_variants_for_debug("users"),
            vec!["users".to_string(), "user".to_string()]
        );
        assert_eq!(
            table_name_variants_for_debug("session_ip_profiles"),
            vec![
                "session_ip_profiles".to_string(),
                "session_ip_profile".to_string(),
            ]
        );
        assert_eq!(
            table_name_variants_for_debug("organization"),
            vec!["organization".to_string(), "organizations".to_string()]
        );
    }

    #[test]
    fn effective_search_path_uses_auth_schema_when_no_search_path_configured() {
        assert_eq!(
            build_effective_search_path("auth_runtime", None),
            "\"auth_runtime\""
        );
    }

    #[test]
    fn effective_search_path_prepends_auth_schema_to_tenant_search_path() {
        let tenant_search_path = SearchPath::parse(Some("tenant_a, public"))
            .expect("search path should parse")
            .expect("search path should be present");

        assert_eq!(
            build_effective_search_path("auth_runtime", Some(&tenant_search_path)),
            "\"auth_runtime\", \"tenant_a\", \"public\""
        );
    }

    #[test]
    fn effective_search_path_falls_back_to_athena_when_schema_is_empty() {
        assert_eq!(build_effective_search_path("", None), "\"athena\"");
    }

    #[test]
    fn effective_search_path_from_sql_prepends_auth_schema() {
        assert_eq!(
            build_effective_search_path_from_sql("auth_runtime", Some("\"tenant_a\", \"public\"")),
            "\"auth_runtime\", \"tenant_a\", \"public\""
        );
    }

    #[test]
    fn effective_search_path_from_sql_uses_athena_fallback() {
        assert_eq!(
            build_effective_search_path_from_sql("", Some("\"tenant_a\"")),
            "\"athena\", \"tenant_a\""
        );
    }

    #[test]
    fn normalize_auth_schema_for_runtime_uses_default_athena() {
        assert_eq!(
            normalize_auth_schema_for_runtime("").expect("empty schema should use default"),
            "athena"
        );
        assert_eq!(
            normalize_auth_schema_for_runtime("   ").expect("blank schema should use default"),
            "athena"
        );
    }

    #[test]
    fn normalize_auth_schema_for_runtime_validates_identifier() {
        let normalized =
            normalize_auth_schema_for_runtime("athena_core").expect("schema should validate");
        assert_eq!(normalized, "athena_core");

        let err = normalize_auth_schema_for_runtime("athena-core")
            .expect_err("invalid schema identifier should fail");
        assert!(
            err.to_string().contains("invalid auth schema value"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn schema_bootstrap_targets_always_include_athena_schema_first() {
        assert_eq!(
            schema_bootstrap_targets("tenant_runtime"),
            vec![DEFAULT_AUTH_SCHEMA, "tenant_runtime"]
        );
    }

    #[test]
    fn schema_bootstrap_targets_deduplicate_default_schema() {
        assert_eq!(
            schema_bootstrap_targets(DEFAULT_AUTH_SCHEMA),
            vec![DEFAULT_AUTH_SCHEMA]
        );
    }

    #[test]
    fn self_heal_audit_schema_matches_default_auth_schema() {
        assert_eq!(SELF_HEAL_AUDIT_SCHEMA, DEFAULT_AUTH_SCHEMA);
    }

    #[test]
    fn bundled_provision_sql_contains_required_tables() {
        let provision_sql = AUTH_PROVISION_SQL.replace("\r\n", "\n");

        assert!(
            provision_sql.contains("CREATE SCHEMA IF NOT EXISTS athena"),
            "provision SQL must create the athena schema"
        );

        for table_name in LOGICAL_TABLE_NAMES {
            let qualified_table = format!("IF NOT EXISTS athena.{table_name}");
            assert!(
                provision_sql.contains(qualified_table.as_str()),
                "provision SQL must include {qualified_table}"
            );
        }

        assert!(
            provision_sql.contains("REFERENCES athena.users"),
            "provision SQL foreign keys must reference athena.users"
        );
        assert!(
            provision_sql.contains("ON athena.users"),
            "provision SQL indexes must target athena.users"
        );
        assert!(
            provision_sql.contains(
                "ALTER TABLE athena.email_send_failures\n    ADD COLUMN IF NOT EXISTS resolved"
            ),
            "provision SQL must self-heal email_send_failures.resolved"
        );
        assert!(
            provision_sql.contains("ALTER TABLE athena.emails\n    ADD COLUMN IF NOT EXISTS flow"),
            "provision SQL must self-heal emails.flow"
        );
        assert!(
            provision_sql.contains(
                "ALTER TABLE athena.api_keys\n    ADD COLUMN IF NOT EXISTS request_count BIGINT"
            ),
            "provision SQL must self-heal api_keys.request_count"
        );
        assert!(
            provision_sql
                .contains("ALTER COLUMN request_count TYPE BIGINT USING request_count::BIGINT"),
            "provision SQL must widen api_keys.request_count for Rust i64 decoding"
        );
    }

    #[test]
    fn auth_provision_sql_for_schema_rewrites_qualified_schema_references() {
        let rewritten = auth_provision_sql_for_schema("tenant_runtime").replace("\r\n", "\n");
        assert!(
            rewritten.contains("CREATE SCHEMA IF NOT EXISTS \"tenant_runtime\";"),
            "rewritten provision SQL must create target schema"
        );
        for table_name in LOGICAL_TABLE_NAMES {
            let qualified_table = format!("IF NOT EXISTS \"tenant_runtime\".{table_name}");
            assert!(
                rewritten.contains(qualified_table.as_str()),
                "rewritten provision SQL must include {qualified_table}"
            );
        }
        assert!(
            rewritten.contains("REFERENCES \"tenant_runtime\".users"),
            "rewritten foreign keys must point at tenant schema users table"
        );
        assert!(
            rewritten.contains("ON \"tenant_runtime\".users"),
            "rewritten indexes must point at tenant schema users table"
        );
        assert!(
            rewritten.contains(
                "ALTER TABLE \"tenant_runtime\".email_send_failures\n    ADD COLUMN IF NOT EXISTS resolved"
            ),
            "rewritten self-heal ALTER TABLE must target tenant schema"
        );
        assert!(
            rewritten.contains(
                "ALTER TABLE \"tenant_runtime\".api_keys\n    ADD COLUMN IF NOT EXISTS request_count BIGINT"
            ),
            "rewritten api_keys self-heal ALTER TABLE must target tenant schema"
        );
        assert!(
            !rewritten.contains("CREATE SCHEMA IF NOT EXISTS athena;"),
            "rewritten SQL must not keep original create schema statement"
        );
    }

    #[test]
    fn config_path_to_env_key_uses_stable_prefix_and_separator() {
        assert_eq!(
            config_path_to_env_key("auth.base_url"),
            "ATHENA_AUTH_CONFIG__AUTH__BASE_URL"
        );
        assert_eq!(
            config_path_to_env_key("auth.public_base_path"),
            "ATHENA_AUTH_CONFIG__AUTH__PUBLIC_BASE_PATH"
        );
        assert_eq!(
            config_path_to_env_key("schema_detection.auth_schema"),
            "ATHENA_AUTH_CONFIG__SCHEMA_DETECTION__AUTH_SCHEMA"
        );
        assert_eq!(
            config_path_to_env_key("auth.password.argon2.memory_cost"),
            "ATHENA_AUTH_CONFIG__AUTH__PASSWORD__ARGON2__MEMORY_COST"
        );
        assert_eq!(
            config_path_to_env_key("auth.session.cookie_domain"),
            "ATHENA_AUTH_CONFIG__AUTH__SESSION__COOKIE_DOMAIN"
        );
    }

    #[test]
    fn config_resolver_collects_leaf_fields() {
        let yaml = r#"
auth:
  base_url: http://localhost:3000
  public_base_path: /api/auth
  trusted_origins:
    - http://localhost:3000
  session:
    expires_in_seconds: 604800
    cookie_domain: .athena-auth.com
schema_detection:
  auth_schema: athena
"#;
        let raw: serde_yaml::Value = serde_yaml::from_str(yaml).expect("yaml should parse");
        let resolver = ConfigResolver::new(env_var_clean);
        let fields = resolver.collect_fields_from_value(&raw);
        let mut paths: Vec<String> = fields.into_iter().map(|field| field.path).collect();
        paths.sort_unstable();
        assert!(paths.contains(&"auth.base_url".to_string()));
        assert!(paths.contains(&"auth.public_base_path".to_string()));
        assert!(paths.contains(&"auth.trusted_origins".to_string()));
        assert!(paths.contains(&"auth.session.expires_in_seconds".to_string()));
        assert!(paths.contains(&"auth.session.cookie_domain".to_string()));
        assert!(paths.contains(&"schema_detection.auth_schema".to_string()));
    }

    #[test]
    fn config_resolver_expands_env_templates() {
        let yaml = r#"
schema_detection:
  auth_schema: ${AUTH_SCHEMA_ENV}
"#;
        let lookup = std::collections::HashMap::from([(
            "AUTH_SCHEMA_ENV".to_string(),
            "tenant_runtime".to_string(),
        )]);
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should parse env template");
        assert_eq!(
            parsed.schema_detection.auth_schema,
            Some("tenant_runtime".to_string())
        );
    }

    #[test]
    fn config_resolver_expands_env_templates_with_default() {
        let yaml = r#"
schema_detection:
  auth_schema: ${AUTH_SCHEMA_ENV:-athena}
"#;
        let lookup = std::collections::HashMap::<String, String>::new();
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should use template default");
        assert_eq!(
            parsed.schema_detection.auth_schema,
            Some("athena".to_string())
        );
    }

    #[test]
    fn config_resolver_errors_when_template_env_is_missing_without_default() {
        let yaml = r#"
schema_detection:
  auth_schema: ${AUTH_SCHEMA_ENV}
"#;
        let lookup = std::collections::HashMap::<String, String>::new();
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let err = resolver
            .resolve_server_config(yaml)
            .expect_err("resolver should fail on unresolved template env");
        assert!(
            err.to_string()
                .contains("references unset env 'AUTH_SCHEMA_ENV'"),
            "unexpected error: {err}"
        );
    }

    #[test]
    fn config_resolver_env_override_wins_over_yaml_value() {
        let yaml = r#"
auth:
  base_url: http://yaml.local
"#;
        let lookup = std::collections::HashMap::from([(
            "ATHENA_AUTH_CONFIG__AUTH__BASE_URL".to_string(),
            "https://env.example.com".to_string(),
        )]);
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should apply env override");
        assert_eq!(
            parsed.auth.base_url,
            Some("https://env.example.com".to_string())
        );
    }

    #[test]
    fn config_resolver_legacy_env_aliases_still_override_yaml() {
        let yaml = r#"
auth:
  base_path: /yaml/path
"#;
        let lookup = std::collections::HashMap::from([(
            "AUTH_BASE_PATH".to_string(),
            "/env/path".to_string(),
        )]);
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should apply legacy alias override");
        assert_eq!(parsed.auth.base_path, Some("/env/path".to_string()));
    }

    #[test]
    fn config_resolver_public_auth_base_path_alias_still_overrides_yaml() {
        let yaml = r#"
auth:
  public_base_path: /yaml/public-auth
"#;
        let lookup = std::collections::HashMap::from([(
            "PUBLIC_AUTH_BASE_PATH".to_string(),
            "/env/public-auth".to_string(),
        )]);
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should apply public auth base path alias override");
        assert_eq!(
            parsed.auth.public_base_path,
            Some("/env/public-auth".to_string())
        );
    }

    #[test]
    fn config_resolver_cookie_domain_alias_still_overrides_yaml() {
        let yaml = r#"
auth:
  session:
    cookie_domain: .yaml.example.com
"#;
        let lookup = std::collections::HashMap::from([(
            "ATHENA_AUTH_COOKIE_DOMAIN".to_string(),
            ".env.example.com".to_string(),
        )]);
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should apply session cookie domain alias override");
        assert_eq!(
            parsed.auth.session.cookie_domain,
            Some(".env.example.com".to_string())
        );
    }

    #[test]
    fn config_resolver_legacy_passkey_related_origins_alias_still_overrides_yaml() {
        let yaml = r#"
auth:
  passkey:
    related_origins:
      - https://yaml-origin.example.com
"#;
        let lookup = std::collections::HashMap::from([(
            "PASSKEY_RELATED_ORIGINS".to_string(),
            "https://one.example.com,https://two.example.com".to_string(),
        )]);
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should apply legacy passkey related origins override");
        assert_eq!(
            parsed.auth.passkey.related_origins,
            vec![
                "https://one.example.com".to_string(),
                "https://two.example.com".to_string()
            ]
        );
    }

    #[test]
    fn config_resolver_sequence_override_accepts_csv() {
        let yaml = r#"
auth:
  trusted_origins:
    - http://yaml.local
"#;
        let lookup = std::collections::HashMap::from([(
            "ATHENA_AUTH_CONFIG__AUTH__TRUSTED_ORIGINS".to_string(),
            "https://one.example.com,https://two.example.com".to_string(),
        )]);
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should apply sequence override");
        assert_eq!(
            parsed.auth.trusted_origins,
            vec![
                "https://one.example.com".to_string(),
                "https://two.example.com".to_string()
            ]
        );
    }

    #[test]
    fn config_resolver_parses_avatar_storage_settings() {
        let yaml = r#"
auth:
  avatar_storage:
    enabled: true
    bucket: avatars
    endpoint: https://storage.example.com
    access_key: ${AVATAR_S3_ACCESS_KEY:-}
    secret_key: ${AVATAR_S3_SECRET_KEY:-}
    url_style: virtual_host
    prefix_path: auth
    file_key_template: "{kind}/{entity_id}/{sha256}.{extension}"
"#;
        let lookup = std::collections::HashMap::from([
            (
                "AVATAR_S3_ACCESS_KEY".to_string(),
                "test-access-key".to_string(),
            ),
            (
                "AVATAR_S3_SECRET_KEY".to_string(),
                "test-secret-key".to_string(),
            ),
        ]);
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should parse avatar storage settings");

        assert!(parsed.auth.avatar_storage.enabled);
        assert_eq!(
            parsed.auth.avatar_storage.bucket,
            Some("avatars".to_string())
        );
        assert_eq!(
            parsed.auth.avatar_storage.endpoint,
            Some("https://storage.example.com".to_string())
        );
        assert_eq!(
            parsed.auth.avatar_storage.access_key,
            Some("test-access-key".to_string())
        );
        assert_eq!(
            parsed.auth.avatar_storage.secret_key,
            Some("test-secret-key".to_string())
        );
        assert_eq!(
            parsed.auth.avatar_storage.url_style,
            Some("virtual_host".to_string())
        );
        assert_eq!(
            parsed.auth.avatar_storage.prefix_path,
            Some("auth".to_string())
        );
        assert_eq!(
            parsed.auth.avatar_storage.file_key_template,
            Some("{kind}/{entity_id}/{sha256}.{extension}".to_string())
        );
    }

    #[test]
    fn config_resolver_parses_password_argon2_settings() {
        let yaml = r#"
auth:
  password:
    argon2:
      memory_cost: 1024
      time_cost: 2
      parallelism: 1
"#;
        let lookup = std::collections::HashMap::<String, String>::new();
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should parse password argon2 settings");
        assert_eq!(parsed.auth.password.argon2.memory_cost, Some(1024));
        assert_eq!(parsed.auth.password.argon2.time_cost, Some(2));
        assert_eq!(parsed.auth.password.argon2.parallelism, Some(1));
    }

    #[test]
    fn config_resolver_parses_repository_config_examples() {
        let example_raw = std::fs::read_to_string("config.yaml.example")
            .expect("config.yaml.example should exist");
        let local_raw = std::fs::read_to_string("config.yaml").expect("config.yaml should exist");

        let empty_lookup = std::collections::HashMap::<String, String>::new();
        let resolver = ConfigResolver::new(|key| empty_lookup.get(key).cloned());

        resolver
            .resolve_server_config(&example_raw)
            .expect("config.yaml.example should parse with env template defaults");
        resolver
            .resolve_server_config(&local_raw)
            .expect("config.yaml should parse with env template defaults");
    }

    #[test]
    fn layered_env_assignments_prefers_env_local_for_missing_process_vars() {
        let existing_env_keys = BTreeSet::<String>::new();
        let env_layers = vec![
            vec![
                (
                    "BASE_URL".to_string(),
                    "http://from-dotenv.example".to_string(),
                ),
                (
                    "PASSWORD_ARGON2_MEMORY_COST".to_string(),
                    "2048".to_string(),
                ),
            ],
            vec![
                (
                    "BASE_URL".to_string(),
                    "http://from-dotenv-local.example".to_string(),
                ),
                (
                    "PASSWORD_ARGON2_MEMORY_COST".to_string(),
                    "1024".to_string(),
                ),
            ],
        ];

        let assignments = layered_env_assignments(&existing_env_keys, &env_layers);

        assert_eq!(
            assignments.get("BASE_URL"),
            Some(&"http://from-dotenv-local.example".to_string())
        );
        assert_eq!(
            assignments.get("PASSWORD_ARGON2_MEMORY_COST"),
            Some(&"1024".to_string())
        );
    }

    #[test]
    fn layered_env_assignments_preserves_existing_process_env() {
        let existing_env_keys = BTreeSet::from([
            "BASE_URL".to_string(),
            "PASSWORD_ARGON2_MEMORY_COST".to_string(),
        ]);
        let env_layers = vec![
            vec![
                (
                    "BASE_URL".to_string(),
                    "http://from-dotenv.example".to_string(),
                ),
                (
                    "PASSWORD_ARGON2_MEMORY_COST".to_string(),
                    "1024".to_string(),
                ),
            ],
            vec![("PASSWORD_ARGON2_TIME_COST".to_string(), "2".to_string())],
        ];

        let assignments = layered_env_assignments(&existing_env_keys, &env_layers);

        assert!(!assignments.contains_key("BASE_URL"));
        assert!(!assignments.contains_key("PASSWORD_ARGON2_MEMORY_COST"));
        assert_eq!(
            assignments.get("PASSWORD_ARGON2_TIME_COST"),
            Some(&"2".to_string())
        );
    }

    #[test]
    fn config_resolver_parses_session_ip_disable_tracking_true() {
        let yaml = r#"
auth:
  session:
    ip:
      disable_tracking: true
"#;
        let lookup = std::collections::HashMap::<String, String>::new();
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should parse disable_tracking=true");
        assert_eq!(parsed.auth.session.ip.disable_tracking, Some(true));
    }

    #[test]
    fn config_resolver_parses_session_ip_disable_tracking_false() {
        let yaml = r#"
auth:
  session:
    ip:
      disable_tracking: false
"#;
        let lookup = std::collections::HashMap::<String, String>::new();
        let resolver = ConfigResolver::new(|key| lookup.get(key).cloned());
        let parsed = resolver
            .resolve_server_config(yaml)
            .expect("resolver should parse disable_tracking=false");
        assert_eq!(parsed.auth.session.ip.disable_tracking, Some(false));
    }

    #[test]
    fn self_heal_audit_insert_sql_builds_for_function_style_columns() {
        let columns: Vec<AuditTableColumn> = vec![
            AuditTableColumn {
                name: "request_id".to_string(),
                data_type: "uuid".to_string(),
                udt_name: "uuid".to_string(),
                is_nullable: false,
                has_default: false,
            },
            AuditTableColumn {
                name: "client_name".to_string(),
                data_type: "text".to_string(),
                udt_name: "text".to_string(),
                is_nullable: false,
                has_default: false,
            },
            AuditTableColumn {
                name: "schema_name".to_string(),
                data_type: "text".to_string(),
                udt_name: "text".to_string(),
                is_nullable: false,
                has_default: false,
            },
            AuditTableColumn {
                name: "function_name".to_string(),
                data_type: "text".to_string(),
                udt_name: "text".to_string(),
                is_nullable: false,
                has_default: false,
            },
            AuditTableColumn {
                name: "action".to_string(),
                data_type: "text".to_string(),
                udt_name: "text".to_string(),
                is_nullable: false,
                has_default: false,
            },
            AuditTableColumn {
                name: "status".to_string(),
                data_type: "text".to_string(),
                udt_name: "text".to_string(),
                is_nullable: false,
                has_default: false,
            },
            AuditTableColumn {
                name: "ddl_sql".to_string(),
                data_type: "text".to_string(),
                udt_name: "text".to_string(),
                is_nullable: false,
                has_default: false,
            },
        ];
        let event = SelfHealingAuditEvent {
            auth_schema: "athena".to_string(),
            schema_existed_before: false,
            missing_tables_before: vec!["users".to_string()],
        };

        let sql: String = build_self_heal_audit_insert_sql(&columns, &event)
            .expect("insert SQL should be generated");

        assert!(sql.contains("INSERT INTO \"athena\".\"audit_log_ddl\""));
        assert!(sql.contains("\"request_id\""));
        assert!(sql.contains("\"client_name\""));
        assert!(sql.contains("startup_self_heal"));
    }

    #[test]
    fn self_heal_audit_insert_sql_returns_none_for_unmapped_required_columns() {
        let columns = vec![AuditTableColumn {
            name: "required_unknown".to_string(),
            data_type: "bytea".to_string(),
            udt_name: "bytea".to_string(),
            is_nullable: false,
            has_default: false,
        }];
        let event = SelfHealingAuditEvent {
            auth_schema: "athena".to_string(),
            schema_existed_before: false,
            missing_tables_before: vec!["users".to_string()],
        };

        let sql = build_self_heal_audit_insert_sql(&columns, &event);
        assert!(
            sql.is_none(),
            "unmapped required columns should skip logging"
        );
    }
}
