#![cfg_attr(not(debug_assertions), windows_subsystem = "windows")]

//! Athena RS binary.
//!
//! Starts the Actix Web server, wires endpoints, configures CORS and tracing,
//! and exposes convenience endpoints for Scylla demo queries.
//!
//!
use actix_cors::Cors;
use actix_web::HttpMessage;
use actix_web::body::MessageBody;
use actix_web::dev::ServiceResponse;
use actix_web::http::{StatusCode, header};
use actix_web::middleware::{ErrorHandlerResponse, ErrorHandlers};
use actix_web::{App, HttpServer, web};
use sentry::types::Dsn as SentryDsn;

use actix_web::web::Data;
use clap::Parser;
use dotenv::dotenv;
use sentry::ClientInitGuard;
use serde_json::json;
use std::collections::HashMap;
use std::env;
use std::fs::OpenOptions;
use std::io::ErrorKind::{AddrInUse, Other};
use std::io::Result as IoResult;
use std::io::{Error, IsTerminal};
use std::net::{SocketAddr, TcpListener};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::OnceCell;
use tracing::{info, warn};

use tracing_appender::non_blocking::{NonBlocking, WorkerGuard};
use tracing_subscriber::EnvFilter;
use tracing_subscriber::Layer;
use tracing_subscriber::fmt::time::ChronoLocal;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use uuid::Uuid;

use socket2::{Domain, Protocol, Socket, TcpKeepalive, Type};

use athena_rs::AppState;
use athena_rs::api::gateway::deferred::{
    gateway_deferred_requeue, gateway_deferred_status, spawn_deferred_query_worker,
};
use athena_rs::api::gateway::delete::delete_data;
use athena_rs::api::gateway::fetch::{
    fetch_data_route, gateway_update_route, get_data_route, proxy_fetch_data_route,
};
use athena_rs::api::gateway::insert::insert_data;
use athena_rs::api::gateway::postgrest::{
    postgrest_delete_route, postgrest_get_route, postgrest_patch_route, postgrest_post_route,
};
use athena_rs::api::gateway::query::gateway_query_route;
use athena_rs::api::gateway::rpc::{gateway_rpc_route, rpc_get_route, rpc_post_route};
use athena_rs::api::headers::request_context::ResolvedAthenaClient;
use athena_rs::api::health::{cluster_health, ping, root};
use athena_rs::api::host_routing::{
    DEFAULT_WILDCARD_TARGET_BASE_URL, infer_wildcard_host_prefix, wildcard_host_pattern,
};
use athena_rs::api::pipelines::{list_pipeline_templates, run_pipeline, simulate_pipeline};
use athena_rs::api::public_routes::public_route_dispatch;
use athena_rs::api::query::count::sql_count_query;
use athena_rs::api::query::sql::{gateway_sql_query, sql_query};
use athena_rs::api::registry::{api_registry, api_registry_by_id};
use athena_rs::api::response::{internal_error, not_found, too_many_requests};
use athena_rs::api::supabase::ssl_enforcement;
use athena_rs::api::{
    admin, athena_docs, athena_openapi_host, athena_router_registry, athena_wss_openapi_host,
    backup, management, provision, schema, storage,
};
use athena_rs::bootstrap::{Bootstrap, build_shared_state};
use athena_rs::cdc::websocket::websocket_server;
use athena_rs::cli::{self, AthenaCli, Command};
use athena_rs::config::{Config, ConfigLocation, DEFAULT_CONFIG_FILE_NAME};
use athena_rs::config_validation::initialize_runtime_env_settings;
use athena_rs::daemon::{spawn_connection_monitor, spawn_registry_reconnect_worker};
use athena_rs::data::public_routes::{PublicGatewayRouteRecord, get_public_gateway_route_by_key};
#[cfg(feature = "metrics")]
use athena_rs::features::metrics::prometheus_metrics;

#[cfg(feature = "tokio-console")]
use athena_rs::features::tokio_console;

#[cfg(feature = "vacuum_health")]
use athena_rs::features::vacuum_health::spawn_vacuum_health_collector;

use athena_rs::utils::logging_task_limiter::spawn_best_effort_logging_task;
use athena_rs::utils::pg_tools::ensure_pg_tools;
use athena_rs::utils::redis_client::{
    GLOBAL_REDIS, initialize_global_redis_from_env, note_redis_failure_and_start_cooldown,
    note_redis_success, should_bypass_redis_temporarily,
};
use athena_rs::utils::request_logging::{RouteRequestLogContext, log_route_request_event};
use athena_rs::wss::gateway_wss_info;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AdmissionStoreBackend {
    Memory,
    Redis,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AdmissionStoreFailMode {
    FailClosed,
    FailOpen,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AdmissionCheckOutcome {
    Allowed,
    Limited,
    StoreUnavailable,
}

#[derive(Debug, Clone)]
struct AdmissionLimiter {
    enabled: bool,
    window: Duration,
    global_limit_per_window: u64,
    per_client_limit_per_window: u64,
    defer_on_limit_enabled: bool,
    defer_route_prefixes: Vec<String>,
    store_backend: AdmissionStoreBackend,
    store_fail_mode: AdmissionStoreFailMode,
    state: Arc<Mutex<AdmissionLimiterWindow>>,
}

#[derive(Debug)]
/// ## `AdmissionLimiterWindow`
/// A window for the admission limiter.
///
/// # Fields
///
/// * `started_at`: The start time of the window.
/// * `global_count`: The global count of the window.
/// * `by_client`: The by client count of the window.
///
/// # Returns
///
/// A new admission limiter window.
struct AdmissionLimiterWindow {
    started_at: Instant,
    global_count: u64,
    by_client: HashMap<String, u64>,
}

impl AdmissionLimiter {
    /// ## `new`
    /// Create a new admission limiter.
    ///
    /// # Arguments
    ///
    /// * `enabled`: A boolean indicating if the admission limiter is enabled.
    /// * `window`: The window duration in seconds.
    /// * `global_limit_per_window`: The global limit per window.
    /// * `per_client_limit_per_window`: The per client limit per window.
    /// * `defer_on_limit_enabled`: A boolean indicating if the admission limiter should defer on limit.
    /// * `defer_route_prefixes`: The route prefixes that will be deferred when the admission limiter is exceeded.
    /// * `store_backend`: The backend to use for the admission limiter.
    /// * `store_fail_mode`: The fail mode to use for the admission limiter.
    ///
    /// # Returns
    ///
    /// A new admission limiter.
    fn new(
        enabled: bool,
        window: Duration,
        global_limit_per_window: u64,
        per_client_limit_per_window: u64,
        defer_on_limit_enabled: bool,
        defer_route_prefixes: Vec<String>,
        store_backend: AdmissionStoreBackend,
        store_fail_mode: AdmissionStoreFailMode,
    ) -> Self {
        Self {
            enabled,
            window,
            global_limit_per_window,
            per_client_limit_per_window,
            defer_on_limit_enabled,
            defer_route_prefixes,
            store_backend,
            store_fail_mode,
            state: Arc::new(Mutex::new(AdmissionLimiterWindow {
                started_at: Instant::now(),
                global_count: 0,
                by_client: HashMap::new(),
            })),
        }
    }
    /// ## `is_enabled`
    /// Check if the admission limiter is enabled.
    ///
    /// # Returns
    ///
    /// A boolean indicating if the admission limiter is enabled.
    ///
    fn is_enabled(&self) -> bool {
        self.enabled
    }

    /// ## `window_secs`
    /// Get the window duration in seconds.
    ///
    /// # Returns
    ///
    /// A u64 representing the window duration in seconds.
    ///
    fn window_secs(&self) -> u64 {
        self.window.as_secs().max(1)
    }

    /// ## `should_defer_route`
    /// Check if the route should be deferred.
    ///
    /// # Returns
    ///
    /// A boolean indicating if the route should be deferred.
    ///
    fn should_defer_route(&self, route: &str) -> bool {
        if !self.defer_on_limit_enabled {
            return false;
        }

        let normalized_route: &str = route.trim();
        if normalized_route.is_empty() {
            return false;
        }

        self.defer_route_prefixes
            .iter()
            .map(|prefix| prefix.trim())
            .filter(|prefix| !prefix.is_empty())
            .any(|prefix| normalized_route.starts_with(prefix))
    }

    /// ## `fail_open_on_store_unavailable`
    /// Check if the store is unavailable and should fail open.
    ///
    /// # Returns
    ///
    /// A boolean indicating if the store is unavailable and should fail open.
    ///
    fn fail_open_on_store_unavailable(&self) -> bool {
        self.store_fail_mode == AdmissionStoreFailMode::FailOpen
    }

    /// ## `check_allow`
    /// Check if the request should be allowed.
    ///
    /// # Returns
    ///
    /// An `AdmissionCheckOutcome` indicating if the request should be allowed.
    ///
    fn check_allow(&self, client: Option<&str>) -> AdmissionCheckOutcome {
        if !self.enabled {
            return AdmissionCheckOutcome::Allowed;
        }

        match self.store_backend {
            AdmissionStoreBackend::Memory => {
                if self.memory_should_allow(client) {
                    AdmissionCheckOutcome::Allowed
                } else {
                    AdmissionCheckOutcome::Limited
                }
            }
            AdmissionStoreBackend::Redis => self.redis_should_allow(client),
        }
    }

    /// ## `memory_should_allow`
    /// Check if the request should be allowed using memory.
    ///
    /// # Returns
    ///
    /// A boolean indicating if the request should be allowed using memory.
    ///
    fn memory_should_allow(&self, client: Option<&str>) -> bool {
        if !self.enabled {
            return true;
        }

        let Ok(mut guard) = self.state.lock() else {
            // Fail-open to keep availability if the in-process limiter state is poisoned.
            return true;
        };

        if guard.started_at.elapsed() >= self.window {
            guard.started_at = Instant::now();
            guard.global_count = 0;
            guard.by_client.clear();
        }

        if self.global_limit_per_window > 0 && guard.global_count >= self.global_limit_per_window {
            return false;
        }

        if self.per_client_limit_per_window > 0 {
            let client_key: String = client
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .unwrap_or("unknown")
                .to_string();
            let current_client_count: u64 = guard.by_client.get(&client_key).copied().unwrap_or(0);
            if current_client_count >= self.per_client_limit_per_window {
                return false;
            }
            guard
                .by_client
                .insert(client_key, current_client_count.saturating_add(1));
        }

        guard.global_count = guard.global_count.saturating_add(1);
        true
    }

    /// ## `redis_should_allow`
    /// Check if the request should be allowed using Redis.
    ///
    /// # Returns
    ///
    /// An `AdmissionCheckOutcome` indicating if the request should be allowed using Redis.
    ///
    fn redis_should_allow(&self, client: Option<&str>) -> AdmissionCheckOutcome {
        let Some(redis) = GLOBAL_REDIS.get() else {
            return AdmissionCheckOutcome::StoreUnavailable;
        };
        if should_bypass_redis_temporarily() {
            return AdmissionCheckOutcome::StoreUnavailable;
        }

        let client_key: String = client
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .unwrap_or("unknown")
            .to_ascii_lowercase();
        let window_secs: u64 = self.window_secs();
        let bucket: i64 = chrono::Utc::now()
            .timestamp()
            .div_euclid(window_secs as i64);
        let global_key: String = format!("athena:admission:global:{bucket}");
        let client_key: String = format!("athena:admission:client:{client_key}:{bucket}");

        let global_count: u64 =
            match redis.increment_counter_with_ttl_blocking(&global_key, window_secs) {
                Ok(value) => {
                    note_redis_success();
                    value
                }
                Err(_) => {
                    note_redis_failure_and_start_cooldown();
                    return AdmissionCheckOutcome::StoreUnavailable;
                }
            };

        if self.global_limit_per_window > 0 && global_count > self.global_limit_per_window {
            return AdmissionCheckOutcome::Limited;
        }

        let per_client_count: u64 =
            match redis.increment_counter_with_ttl_blocking(&client_key, window_secs) {
                Ok(value) => {
                    note_redis_success();
                    value
                }
                Err(_) => {
                    note_redis_failure_and_start_cooldown();
                    return AdmissionCheckOutcome::StoreUnavailable;
                }
            };
        if self.per_client_limit_per_window > 0
            && per_client_count > self.per_client_limit_per_window
        {
            return AdmissionCheckOutcome::Limited;
        }
        AdmissionCheckOutcome::Allowed
    }
}

/// ## `x_athena_defer_requested`
/// Check if the request should be deferred.
///
/// # Returns
///
/// A boolean indicating if the request should be deferred.
///
fn x_athena_defer_requested(req: &actix_web::dev::ServiceRequest) -> bool {
    req.headers()
        .get("X-Athena-Defer")
        .and_then(|value| value.to_str().ok())
        .map(str::trim)
        .map(|value| matches!(value, "1" | "true" | "TRUE" | "yes" | "YES"))
        .unwrap_or(false)
}

static ADMISSION_EVENT_TABLE_INIT: OnceCell<()> = OnceCell::const_new();
static FILE_LOG_GUARDS: Mutex<Vec<WorkerGuard>> = Mutex::new(Vec::new());

const PRIMARY_TRACING_LOG_DIR: &str = "/var/log/athena";
const FALLBACK_TRACING_LOG_DIR: &str = "./var/log/athena";
const SUCCESS_LOG_FILE_NAME: &str = "success.log";
const ERROR_LOG_FILE_NAME: &str = "error.log";
/// Default tracing filter when `RUST_LOG` is not set.
///
/// `sqlx_core::pool::inner` emits high-volume INFO lines during short-lived
/// network path churn (for example VPN/proxy switches) while stale idle sockets
/// are being discarded. We keep warnings/errors visible but suppress that INFO
/// noise by default.
const DEFAULT_TRACING_FILTER: &str = "info,sqlx_core::pool::inner=warn";

fn as_i64_saturating(value: u64) -> i64 {
    i64::try_from(value).unwrap_or(i64::MAX)
}

/// ## `persist_admission_event`
/// Persist the admission event.
///
/// # Arguments
///
/// * `state`: The application state.
/// * `limiter`: The admission limiter.
/// * `method`: The method of the request.
/// * `route`: The route of the request.
/// * `client_name`: The client name of the request.
/// * `request_bytes`: The request bytes of the request.
/// * `defer_requested`: A boolean indicating if the request should be deferred.
/// * `is_query_route`: A boolean indicating if the request is a query route.
/// * `decision`: The decision of the request.
/// * `reason`: The reason of the request.
/// * `deferred_request_id`: The deferred request id of the request.
/// * `retry_after_seconds`: The retry after seconds of the request.
///
/// # Returns
///
/// A result indicating if the admission event was persisted successfully.
///
async fn persist_admission_event(
    state: &AppState,
    limiter: &AdmissionLimiter,
    method: &str,
    route: &str,
    client_name: Option<&str>,
    request_bytes: Option<u64>,
    defer_requested: bool,
    is_query_route: bool,
    decision: &str,
    reason: &str,
    deferred_request_id: Option<&str>,
    retry_after_seconds: u64,
) -> Result<(), String> {
    let Some(logging_client_name) = state.logging_client_name.as_ref() else {
        return Ok(());
    };

    let Some(pool) = state.pg_registry.get_pool(logging_client_name) else {
        return Ok(());
    };

    let event_id: String = Uuid::new_v4().to_string();
    let request_bytes_i64: Option<i64> = request_bytes.map(as_i64_saturating);
    let retry_after_i32: i32 = retry_after_seconds.min(i32::MAX as u64) as i32;
    let window_i32: i32 = limiter.window_secs().min(i32::MAX as u64) as i32;
    let global_limit_i64: i64 = as_i64_saturating(limiter.global_limit_per_window);
    let per_client_limit_i64: i64 = as_i64_saturating(limiter.per_client_limit_per_window);

    // TODO: remove manual_query
    sqlx::query(
        r#"
        INSERT INTO public.gateway_admission_events (
            event_id,
            method,
            route,
            client_name,
            request_bytes,
            defer_requested,
            is_query_route,
            decision,
            reason,
            deferred_request_id,
            retry_after_seconds,
            window_seconds,
            global_limit_per_window,
            per_client_limit_per_window,
            meta
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
        "#,
    )
    .bind(event_id)
    .bind(method)
    .bind(route)
    .bind(client_name)
    .bind(request_bytes_i64)
    .bind(defer_requested)
    .bind(is_query_route)
    .bind(decision)
    .bind(reason)
    .bind(deferred_request_id)
    .bind(retry_after_i32)
    .bind(window_i32)
    .bind(global_limit_i64)
    .bind(per_client_limit_i64)
    .bind(json!({
        "defer_on_limit_enabled": limiter.defer_on_limit_enabled,
        "defer_route_prefixes": limiter.defer_route_prefixes,
    }))
    .execute(&pool)
    .await
    .map_err(|err| format!("failed to persist admission event: {err}"))?;

    Ok(())
}

/// ## `spawn_admission_event_persist`
/// Spawn a task to persist the admission event.
///
/// # Arguments
///
/// * `app_state`: The application state.
/// * `limiter`: The admission limiter.
/// * `method`: The method of the request.
/// * `route`: The route of the request.
/// * `client_name`: The client name of the request.
/// * `request_bytes`: The request bytes of the request.
/// * `defer_requested`: A boolean indicating if the request should be deferred.
/// * `is_query_route`: A boolean indicating if the request is a query route.
/// * `decision`: The decision of the request.
/// * `reason`: The reason of the request.
/// * `deferred_request_id`: The deferred request id of the request.
/// * `retry_after_seconds`: The retry after seconds of the request.
///
/// # Returns
///
/// A task handle for the persisted admission event.
///
fn spawn_admission_event_persist(
    app_state: Option<Data<AppState>>,
    limiter: AdmissionLimiter,
    method: String,
    route: String,
    client_name: Option<String>,
    request_bytes: Option<u64>,
    defer_requested: bool,
    is_query_route: bool,
    decision: &'static str,
    reason: &'static str,
    deferred_request_id: Option<String>,
    retry_after_seconds: u64,
) {
    let Some(state) = app_state else {
        return;
    };

    let logging_task_limiter: Option<Arc<tokio::sync::Semaphore>> =
        state.logging_task_limiter.clone();
    spawn_best_effort_logging_task(
        logging_task_limiter,
        "admission_event_persist",
        async move {
            if let Err(err) = persist_admission_event(
                state.get_ref(),
                &limiter,
                &method,
                &route,
                client_name.as_deref(),
                request_bytes,
                defer_requested,
                is_query_route,
                decision,
                reason,
                deferred_request_id.as_deref(),
                retry_after_seconds,
            )
            .await
            {
                warn!(
                    error = %err,
                    method = %method,
                    route = %route,
                    decision = %decision,
                    "Failed to persist admission event"
                );
            }
        },
    );
}

#[derive(Debug, Clone)]
struct WildcardHostResolution {
    prefix: Option<String>,
    resolved_client: Option<String>,
    outcome: &'static str,
    error_message: Option<String>,
}

async fn resolve_wildcard_host_client(
    req: &mut actix_web::dev::ServiceRequest,
    app_state: Option<&Data<AppState>>,
) -> WildcardHostResolution {
    let Some(prefix) = infer_wildcard_host_prefix(req.headers()) else {
        return WildcardHostResolution {
            prefix: None,
            resolved_client: None,
            outcome: "not_wildcard_host",
            error_message: None,
        };
    };

    let Some(state) = app_state.map(Data::get_ref) else {
        return WildcardHostResolution {
            prefix: Some(prefix),
            resolved_client: None,
            outcome: "missing_app_state",
            error_message: None,
        };
    };
    let Some(logging_client_name) = state.logging_client_name.as_ref() else {
        return WildcardHostResolution {
            prefix: Some(prefix),
            resolved_client: None,
            outcome: "missing_logging_client",
            error_message: None,
        };
    };
    let Some(pool) = state.pg_registry.get_pool(logging_client_name) else {
        return WildcardHostResolution {
            prefix: Some(prefix),
            resolved_client: None,
            outcome: "missing_logging_pool",
            error_message: None,
        };
    };
    let route_lookup: Result<Option<PublicGatewayRouteRecord>, sqlx::Error> =
        get_public_gateway_route_by_key(&pool, &prefix).await;

    let route: PublicGatewayRouteRecord = match route_lookup {
        Ok(Some(route)) => route,
        Ok(None) => {
            return WildcardHostResolution {
                prefix: Some(prefix),
                resolved_client: None,
                outcome: "route_not_found",
                error_message: None,
            };
        }
        Err(err) => {
            warn!(
                prefix = %prefix,
                error = %err,
                "Wildcard host client lookup failed"
            );
            return WildcardHostResolution {
                prefix: Some(prefix),
                resolved_client: None,
                outcome: "lookup_failed",
                error_message: Some(err.to_string()),
            };
        }
    };

    if !route.is_active {
        return WildcardHostResolution {
            prefix: Some(prefix),
            resolved_client: None,
            outcome: "route_inactive",
            error_message: None,
        };
    }

    let resolved_client = route.client_name.trim().to_string();
    if resolved_client.is_empty() {
        return WildcardHostResolution {
            prefix: Some(prefix),
            resolved_client: None,
            outcome: "empty_client_name",
            error_message: None,
        };
    }

    req.extensions_mut()
        .insert(ResolvedAthenaClient(resolved_client.clone()));
    WildcardHostResolution {
        prefix: Some(prefix),
        resolved_client: Some(resolved_client),
        outcome: "resolved",
        error_message: None,
    }
}

/// ## `admission_and_wildcard_middleware`
/// The admission and wildcard middleware.
///
/// # Arguments
///
/// * `req`: The request.
/// * `next`: The next middleware.
/// * `limiter`: The admission limiter.
///
/// # Returns
///
/// A result indicating if the request was allowed.
///
async fn admission_and_wildcard_middleware<B>(
    mut req: actix_web::dev::ServiceRequest,
    next: actix_web::middleware::Next<B>,
    limiter: AdmissionLimiter,
) -> Result<actix_web::dev::ServiceResponse<actix_web::body::BoxBody>, actix_web::Error>
where
    B: MessageBody + 'static,
{
    let metrics_state: Option<Data<AppState>> = req.app_data::<web::Data<AppState>>().cloned();
    let method: String = req.method().as_str().to_string();
    let route: String = req
        .match_pattern()
        .map(|value| value.to_string())
        .unwrap_or_else(|| req.path().to_string());
    let request_bytes: Option<u64> = req
        .headers()
        .get(header::CONTENT_LENGTH)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse::<u64>().ok());
    let host = req
        .headers()
        .get(header::HOST)
        .and_then(|value| value.to_str().ok())
        .map(str::to_string);
    let athena_defer: bool = x_athena_defer_requested(&req);
    let started: Instant = Instant::now();

    let wildcard_resolution: WildcardHostResolution =
        resolve_wildcard_host_client(&mut req, metrics_state.as_ref()).await;

    if let Some(resolved_client) = wildcard_resolution.resolved_client.as_deref() {
        info!(
            host_pattern = %wildcard_host_pattern(),
            wildcard_target = %DEFAULT_WILDCARD_TARGET_BASE_URL,
            resolved_client = %resolved_client,
            "Resolved X-Athena-Client from wildcard host route"
        );
    }
    if wildcard_resolution.prefix.is_some() {
        log_route_request_event(
            metrics_state.as_ref().map(Data::get_ref),
            RouteRequestLogContext {
                source: Some("wildcard_host".to_string()),
                outcome: Some(wildcard_resolution.outcome.to_string()),
                method: Some(method.clone()),
                path: Some(route.clone()),
                host,
                wildcard_host_pattern: Some(wildcard_host_pattern()),
                wildcard_prefix: wildcard_resolution.prefix.clone(),
                route_key: wildcard_resolution.prefix.clone(),
                resolved_client: wildcard_resolution.resolved_client.clone(),
                error_message: wildcard_resolution.error_message.clone(),
                metadata: Some(json!({
                    "wildcard_target": DEFAULT_WILDCARD_TARGET_BASE_URL
                })),
                ..Default::default()
            },
        );
    }

    let athena_client: Option<String> = req
        .extensions()
        .get::<ResolvedAthenaClient>()
        .map(|value| value.0.clone())
        .or_else(|| {
            req.headers()
                .get("X-Athena-Client")
                .and_then(|value| value.to_str().ok())
                .map(str::to_string)
        });

    if let Some(state) = &metrics_state {
        state.metrics_state.begin_http_request(&method, &route);
    }
    let is_query_route: bool = route.starts_with("/gateway/query");
    let limiter_for_check: AdmissionLimiter = limiter.clone();
    let check: AdmissionCheckOutcome = limiter_for_check.check_allow(athena_client.as_deref());
    let mut allow_request: bool = matches!(check, AdmissionCheckOutcome::Allowed);
    let store_unavailable: bool = matches!(check, AdmissionCheckOutcome::StoreUnavailable);
    if store_unavailable {
        allow_request = limiter_for_check.fail_open_on_store_unavailable();
    }
    if is_query_route && athena_defer {
        allow_request = true;
    }

    if !allow_request {
        let should_defer: bool = limiter.should_defer_route(&route);
        let mut status_code: u16 = 429;
        let decision: &'static str = "rejected";
        let mut reason: &'static str = "admission_limit_exceeded";
        let deferred_request_id_for_log: Option<String> = None;
        let mut resp: actix_web::HttpResponse = if store_unavailable {
            status_code = 503;
            reason = "admission_store_unavailable";
            actix_web::HttpResponse::ServiceUnavailable().json(json!({
                "status": "error",
                "code": "admission_store_unavailable",
                "message": "Admission limiter unavailable",
                "error": "Admission store backend is unavailable. Retry shortly."
            }))
        } else if is_query_route && should_defer {
            reason = "gateway_query_requires_explicit_defer_header";
            too_many_requests(
                "Too Many Requests",
                "Gateway query overload requires explicit deferred mode. Retry with header X-Athena-Defer: true to queue the query.",
            )
        } else {
            too_many_requests(
                "Too Many Requests",
                "Request rate exceeded gateway admission limit",
            )
        };

        let retry_after_seconds: String = limiter.window_secs().to_string();
        let retry_after_seconds_u64: u64 = limiter.window_secs();
        if let Ok(retry_after_value) = retry_after_seconds.parse() {
            resp.headers_mut()
                .insert(header::RETRY_AFTER, retry_after_value);
        }
        resp.headers_mut()
            .insert(header::SERVER, "XYLEX/0".parse().unwrap());

        spawn_admission_event_persist(
            metrics_state.clone(),
            limiter.clone(),
            method.clone(),
            route.clone(),
            athena_client.clone(),
            request_bytes,
            athena_defer,
            is_query_route,
            decision,
            reason,
            deferred_request_id_for_log,
            retry_after_seconds_u64,
        );

        if let Some(state) = metrics_state {
            state.metrics_state.finish_http_request(
                &method,
                &route,
                status_code,
                started.elapsed().as_secs_f64(),
                request_bytes,
                None,
                athena_client.as_deref(),
            );
        }

        let response: ServiceResponse = req.into_response(resp).map_into_boxed_body();
        return Ok(response);
    }

    match next.call(req).await {
        Ok(mut res) => {
            let response_bytes: Option<u64> = res
                .headers()
                .get(header::CONTENT_LENGTH)
                .and_then(|value| value.to_str().ok())
                .and_then(|value| value.parse::<u64>().ok());
            res.headers_mut()
                .insert(header::SERVER, "XYLEX/0".parse().unwrap());
            let res = res.map_into_boxed_body();
            if let Some(state) = metrics_state {
                state.metrics_state.finish_http_request(
                    &method,
                    &route,
                    res.status().as_u16(),
                    started.elapsed().as_secs_f64(),
                    request_bytes,
                    response_bytes,
                    athena_client.as_deref(),
                );
            }
            Ok(res)
        }
        Err(err) => {
            if let Some(state) = metrics_state {
                state.metrics_state.record_http_handler_error(
                    &method,
                    &route,
                    started.elapsed().as_secs_f64(),
                    request_bytes,
                    athena_client.as_deref(),
                );
            }
            Err(err)
        }
    }
}

/// Load `.env` from the directory containing the active `config.yaml`.
///
/// [`dotenv`] already ran from the process working directory; this picks up secrets when the
/// resolved config lives elsewhere (for example `%APPDATA%/athena/config.yaml` with `.env` beside it,
/// or `--config path/to/config.yaml` while the shell cwd is not the repo root).
fn load_dotenv_beside_config_file(config_file: &Path) {
    if let Some(dir) = config_file.parent() {
        let env_path: PathBuf = dir.join(".env");
        if env_path.is_file() {
            match dotenv::from_path(&env_path) {
                Ok(()) => {
                    info!(path = %env_path.display(), "Loaded supplementary .env beside config file");
                }
                Err(err) => {
                    warn!(
                        path = %env_path.display(),
                        error = %err,
                        "Found .env beside config file but failed to parse it; ignoring \
                         (quoted Windows paths: use C:/... not C:\\... inside \"...\")"
                    );
                }
            }
        }
    }
}

#[actix_web::main]
async fn main() -> IoResult<()> {
    if let Err(err) = dotenv() {
        warn!(
            error = %err,
            "Failed to parse `.env` in the current directory; variables from that file are skipped \
             (in quoted values use forward slashes for Windows paths, e.g. C:/Program Files/..., or escape each \\)"
        );
    }
    initialize_global_redis_from_env();
    let sentry_guard: Option<ClientInitGuard> = init_sentry();
    init_tracing(sentry_guard.is_some());
    let _sentry_guard: Option<ClientInitGuard> = sentry_guard;

    let cli: AthenaCli = AthenaCli::parse();
    let config_path: Option<PathBuf> = cli.config_path.clone();
    let pipelines_path: PathBuf = cli.pipelines_path.clone();
    let port_override: Option<u16> = cli.port;
    let max_workers_override: Option<usize> = cli.max_workers;
    let command: Option<Command> = cli.command;
    let api_only: bool = cli.api_only;
    #[cfg(feature = "cdc")]
    let cdc_only: bool = cli.cdc_only;

    let config_overridden: bool = config_path.is_some();
    let config_path: PathBuf = config_path
        .clone()
        .unwrap_or_else(|| PathBuf::from(DEFAULT_CONFIG_FILE_NAME));
    let pipelines_path: String = pipelines_path.to_string_lossy().to_string();
    let (config, resolved_config_path): (Config, PathBuf) = if config_overridden {
        let cfg: Config = Config::load_from(&config_path).map_err(|err| {
            let attempted_locations: Vec<ConfigLocation> = vec![ConfigLocation::new(
                "explicit config path".to_string(),
                config_path.clone(),
            )];
            Error::new(
                Other,
                format!(
                    "Failed to load config '{}': {}. Looked in:\n{}",
                    config_path.display(),
                    err,
                    format_attempted_locations(&attempted_locations)
                ),
            )
        })?;
        (cfg, config_path.clone())
    } else {
        match Config::load_default() {
            Ok(outcome) => {
                let path: PathBuf = outcome.path.clone();
                if outcome.seeded_default {
                    info!(
                        path = %path.display(),
                        "Seeded default configuration (first run)"
                    );
                } else {
                    info!(path = %path.display(), "Loaded configuration");
                }
                (outcome.config, path)
            }
            Err(err) => {
                let attempts: Vec<ConfigLocation> = err.attempted_locations.clone();
                return Err(Error::new(
                    Other,
                    format!(
                        "Failed to load config '{}': {}. Looked in:\n{}",
                        config_path.display(),
                        err,
                        format_attempted_locations(&attempts)
                    ),
                ));
            }
        }
    };

    load_dotenv_beside_config_file(&resolved_config_path);
    initialize_runtime_env_settings(&config.validation_ranges);

    match command {
        Some(Command::Server) => {
            let bootstrap: Bootstrap = build_shared_state(&config, &pipelines_path)
                .await
                .map_err(|err| Error::new(Other, err.to_string()))?;
            run_server(&config, bootstrap, port_override, max_workers_override).await
        }
        Some(Command::Pipeline(args)) => {
            let bootstrap: Bootstrap = build_shared_state(&config, &pipelines_path)
                .await
                .map_err(|err| Error::new(Other, err.to_string()))?;
            cli::run_pipeline_command(&bootstrap, args)
                .await
                .map_err(|err| Error::new(Other, err.to_string()))
        }
        Some(Command::Clients { command }) => {
            cli::run_clients_command(command).map_err(|err| Error::new(Other, err.to_string()))
        }
        Some(Command::Fetch(cmd)) => cli::run_fetch_command(cmd)
            .await
            .map_err(|err| Error::new(Other, err.to_string())),
        Some(Command::Provision(args)) => cli::run_provision_command(args)
            .await
            .map_err(|err| Error::new(Other, err.to_string())),
        #[cfg(feature = "cdc")]
        Some(Command::Cdc { command }) => cli::run_cdc_command(&config, command)
            .await
            .map_err(|err| Error::new(Other, err.to_string())),
        Some(Command::Diag) => {
            cli::run_diag_command().map_err(|err| Error::new(Other, err.to_string()))
        }
        Some(Command::Version) => {
            cli::run_version_command();
            Ok(())
        }
        None => {
            #[cfg(feature = "cdc")]
            {
                if cdc_only {
                    let port: u16 = port_override.unwrap_or(4053);
                    websocket_server(port)
                        .await
                        .map_err(|err| Error::new(Other, err.to_string()))
                } else if api_only {
                    let bootstrap: Bootstrap = build_shared_state(&config, &pipelines_path)
                        .await
                        .map_err(|err| Error::new(Other, err.to_string()))?;
                    run_server(&config, bootstrap, port_override, max_workers_override).await
                } else {
                    Err(Error::new(
                        Other,
                        "No command provided; pass --api-only to boot the API, --cdc-only for the CDC WebSocket server, or specify a subcommand.",
                    ))
                }
            }
            #[cfg(not(feature = "cdc"))]
            {
                if api_only {
                    let bootstrap: Bootstrap = build_shared_state(&config, &pipelines_path)
                        .await
                        .map_err(|err| Error::new(Other, err.to_string()))?;
                    run_server(&config, bootstrap, port_override, max_workers_override).await
                } else {
                    Err(Error::new(
                        Other,
                        "No command provided; pass --api-only to boot the API or specify a subcommand.",
                    ))
                }
            }
        }
    }
}

/// Registers default error handlers so 404 and all 5xx responses use the standard
/// API envelope (status, message, error). Ensures every request path can result
/// in at least a 2xx or a 5xx (or 4xx) with a consistent JSON body.
fn default_error_handlers<B>() -> ErrorHandlers<B>
where
    B: MessageBody + 'static,
{
    ErrorHandlers::new()
        .handler(StatusCode::NOT_FOUND, |res: ServiceResponse<B>| {
            let has_json_body: bool = res
                .headers()
                .get(header::CONTENT_TYPE)
                .and_then(|v| v.to_str().ok())
                .is_some_and(|ct| ct.contains("application/json"));

            if has_json_body {
                Ok(ErrorHandlerResponse::Response(
                    res.map_into_boxed_body().map_into_right_body(),
                ))
            } else {
                let (req, _) = res.into_parts();
                let resp: actix_web::HttpResponse =
                    not_found("Not Found", "The requested resource was not found");
                Ok(ErrorHandlerResponse::Response(
                    ServiceResponse::new(req, resp)
                        .map_into_boxed_body()
                        .map_into_right_body(),
                ))
            }
        })
        .default_handler_server(|res: ServiceResponse<B>| {
            let has_json_body: bool = res
                .headers()
                .get(header::CONTENT_TYPE)
                .and_then(|v| v.to_str().ok())
                .is_some_and(|ct| ct.contains("application/json"));

            if has_json_body {
                Ok(ErrorHandlerResponse::Response(
                    res.map_into_boxed_body().map_into_right_body(),
                ))
            } else {
                let (req, _) = res.into_parts();
                let resp: actix_web::HttpResponse =
                    internal_error("Internal Server Error", "An unexpected error occurred");
                Ok(ErrorHandlerResponse::Response(
                    ServiceResponse::new(req, resp)
                        .map_into_boxed_body()
                        .map_into_right_body(),
                ))
            }
        })
}

/// ## `run_server`
/// Run the server.
///
/// # Arguments
///
/// * `config`: The configuration.
/// * `bootstrap`: The bootstrap.
/// * `port_override`: The port override.
///
/// # Returns
///
/// A result indicating if the server was run successfully.
///
async fn run_server(
    config: &Config,
    bootstrap: Bootstrap,
    port_override: Option<u16>,
    max_workers_override: Option<usize>,
) -> IoResult<()> {
    let port: u16 = port_override.unwrap_or_else(|| config.get_api_port());

    let keep_alive: Duration = Duration::from_secs(config.get_http_keep_alive_timeout_secs());
    let client_disconnect_timeout: Duration =
        Duration::from_secs(config.get_client_disconnect_timeout_value_secs());
    let client_request_timeout: Duration =
        Duration::from_secs(config.get_client_request_timeout_value_secs());
    let worker_count: usize =
        max_workers_override.unwrap_or_else(|| config.get_http_worker_count());
    if worker_count == 0 {
        return Err(Error::new(
            Other,
            "Invalid --max-workers value: must be greater than 0.",
        ));
    }
    let max_connections: usize = config.get_http_max_connections_value();
    let backlog: usize = config.get_http_backlog_value();
    let tcp_keepalive: Duration = Duration::from_secs(config.get_tcp_keepalive_timeout_secs());
    let prometheus_metrics_enabled: bool = config.get_prometheus_metrics_enabled();
    let admission_window_secs: u64 = config.get_gateway_admission_window_secs().max(1);
    let admission_store_backend: AdmissionStoreBackend =
        match config.get_gateway_admission_store_backend().as_str() {
            "memory" => AdmissionStoreBackend::Memory,
            _ => AdmissionStoreBackend::Redis,
        };
    let admission_store_fail_mode: AdmissionStoreFailMode =
        match config.get_gateway_admission_store_fail_mode().as_str() {
            "fail_open" => AdmissionStoreFailMode::FailOpen,
            _ => AdmissionStoreFailMode::FailClosed,
        };
    let admission_limiter: AdmissionLimiter = AdmissionLimiter::new(
        config.get_gateway_admission_limit_enabled(),
        Duration::from_secs(admission_window_secs),
        config.get_gateway_admission_global_requests_per_window(),
        config.get_gateway_admission_per_client_requests_per_window(),
        config.get_gateway_admission_defer_on_limit_enabled(),
        config.get_gateway_admission_defer_route_prefixes(),
        admission_store_backend,
        admission_store_fail_mode,
    );

    if admission_limiter.is_enabled() {
        info!(
            admission_window_secs = admission_limiter.window_secs(),
            admission_global_requests_per_window =
                config.get_gateway_admission_global_requests_per_window(),
            admission_per_client_requests_per_window =
                config.get_gateway_admission_per_client_requests_per_window(),
            admission_defer_on_limit_enabled =
                config.get_gateway_admission_defer_on_limit_enabled(),
            admission_store_backend = ?admission_store_backend,
            admission_store_fail_mode = ?admission_store_fail_mode,
            "Gateway admission limiter enabled"
        );
    }

    let addr: SocketAddr = SocketAddr::from(([0, 0, 0, 0], port));
    let socket: Socket = Socket::new(Domain::IPV4, Type::STREAM, Some(Protocol::TCP))?;

    socket.set_nonblocking(true)?;
    socket.set_reuse_address(true)?;
    socket.set_keepalive(true)?;
    let keepalive_cfg: TcpKeepalive = TcpKeepalive::new().with_time(tcp_keepalive);
    socket.set_tcp_keepalive(&keepalive_cfg)?;
    socket.bind(&addr.into()).map_err(|e| {
        let msg = if e.kind() == AddrInUse {
            format!(
                "Address already in use (port {}). Stop the process using this port or choose a different one (e.g. --port).",
                port
            )
        } else {
            format!("Failed to bind to {}: {}", addr, e)
        };
        Error::new(e.kind(), msg)
    })?;
    let listen_backlog: i32 = backlog.min(i32::MAX as usize) as i32;
    socket
        .listen(listen_backlog)
        .map_err(|e| Error::new(e.kind(), format!("Failed to listen on {}: {}", addr, e)))?;
    let listener: TcpListener = socket.into();

    // Ensure tools exist (version selection happens at backup/restore runtime).
    // This keeps startup cheap while still surfacing missing binaries early.
    ensure_pg_tools()
        .await
        .map_err(|err| Error::new(Other, format!("PostgreSQL tools unavailable: {err}")))?;

    let app_state: Data<AppState> = bootstrap.app_state.clone();

    spawn_connection_monitor(app_state.clone());

    #[cfg(feature = "vacuum_health")]
    spawn_vacuum_health_collector(app_state.clone());
    spawn_registry_reconnect_worker(app_state.clone());

    let deferred_worker_poll_ms: u64 = config.get_gateway_deferred_query_worker_poll_ms().max(100);

    spawn_deferred_query_worker(
        app_state.clone(),
        config.get_gateway_deferred_query_worker_enabled(),
        Duration::from_millis(deferred_worker_poll_ms),
    );

    warn!("CORS is temporarily disabled: allowing any origin");

    HttpServer::new(move || {
        let limiter: AdmissionLimiter = admission_limiter.clone();
        let cors: Cors = Cors::default()
            .allow_any_origin()
            .allow_any_method()
            .allow_any_header();

        let mut app = App::new()
            .wrap(default_error_handlers())
            .wrap(cors)
            .wrap(actix_web::middleware::from_fn(move |req, next| {
                admission_and_wildcard_middleware(req, next, limiter.clone())
            }))
            .app_data(app_state.clone())
            .service(root)
            .service(ping)
            .service(cluster_health)
            .service(sql_query)
            .service(gateway_sql_query)
            .service(sql_count_query)
            .service(fetch_data_route)
            .service(get_data_route)
            .service(proxy_fetch_data_route)
            .service(gateway_update_route)
            .service(gateway_query_route)
            .service(gateway_rpc_route)
            .service(insert_data)
            .service(delete_data)
            .service(rpc_get_route)
            .service(rpc_post_route)
            .service(postgrest_get_route)
            .service(postgrest_post_route)
            .service(postgrest_patch_route)
            .service(postgrest_delete_route)
            .service(run_pipeline)
            .service(simulate_pipeline)
            .service(list_pipeline_templates)
            .service(athena_router_registry)
            .service(athena_openapi_host)
            .service(athena_wss_openapi_host)
            .service(api_registry)
            .service(athena_docs)
            .service(api_registry_by_id)
            .service(gateway_deferred_status)
            .service(gateway_deferred_requeue)
            .service(gateway_wss_info)
            .service(public_route_dispatch)
            .configure(admin::services)
            .configure(backup::services)
            .configure(provision::services)
            .configure(management::services)
            .configure(schema::services)
            .configure(storage::services)
            .service(ssl_enforcement);
        if prometheus_metrics_enabled {
            #[cfg(feature = "metrics")]
            {
                app = app.service(prometheus_metrics);
            }
        }
        app
    })
    .workers(worker_count)
    .keep_alive(keep_alive)
    .client_disconnect_timeout(client_disconnect_timeout)
    .client_request_timeout(client_request_timeout)
    .max_connections(max_connections)
    .backlog(backlog as u32)
    .listen(listener)?
    .run()
    .await
}

/// Configures tracing with chrono timestamps and an environment-configurable filter.
/// ANSI colors are enabled only when stderr is a TTY (or when ATHENA_ANSI=1).
/// Set NO_COLOR or ATHENA_ANSI=0 to disable colors even in a TTY.
fn ensure_log_destination(dir: &Path) -> IoResult<()> {
    std::fs::create_dir_all(dir)?;
    OpenOptions::new()
        .create(true)
        .append(true)
        .open(dir.join(SUCCESS_LOG_FILE_NAME))?;
    OpenOptions::new()
        .create(true)
        .append(true)
        .open(dir.join(ERROR_LOG_FILE_NAME))?;
    Ok(())
}

fn resolve_tracing_log_directory() -> (Option<PathBuf>, Option<String>) {
    let preferred_dir: PathBuf = PathBuf::from(PRIMARY_TRACING_LOG_DIR);
    match ensure_log_destination(&preferred_dir) {
        Ok(()) => (Some(preferred_dir), None),
        Err(preferred_err) => {
            let fallback_dir: PathBuf = PathBuf::from(FALLBACK_TRACING_LOG_DIR);
            match ensure_log_destination(&fallback_dir) {
                Ok(()) => (
                    Some(fallback_dir.clone()),
                    Some(format!(
                        "unable to use {}; falling back to {} ({})",
                        preferred_dir.display(),
                        fallback_dir.display(),
                        preferred_err
                    )),
                ),
                Err(fallback_err) => (
                    None,
                    Some(format!(
                        "unable to initialize tracing log files at {} ({}) or {} ({}); continuing without file log sinks",
                        preferred_dir.display(),
                        preferred_err,
                        fallback_dir.display(),
                        fallback_err
                    )),
                ),
            }
        }
    }
}

fn is_error_level(metadata: &tracing::Metadata<'_>) -> bool {
    *metadata.level() == tracing::Level::ERROR
}

/// Stderr `fmt` layer; generic so it can stack above `console_subscriber` + `EnvFilter`.
fn athena_tracing_stdout_fmt_layer<S>(
    ansi: bool,
) -> impl tracing_subscriber::layer::Layer<S> + Send + Sync + 'static
where
    S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
{
    tracing_subscriber::fmt::layer()
        .with_ansi(ansi)
        .with_level(true)
        .with_target(true)
        .with_timer(ChronoLocal::new("%H:%M:%S%.3f".to_string()))
}

/// File `fmt` layer (success vs error split); generic for the same reason as [`athena_tracing_stdout_fmt_layer`].
fn athena_tracing_file_layer<S>(
    writer: NonBlocking,
    errors_only: bool,
) -> impl tracing_subscriber::layer::Layer<S> + Send + Sync + 'static
where
    S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
{
    tracing_subscriber::fmt::layer()
        .with_ansi(false)
        .with_level(true)
        .with_target(true)
        .with_timer(ChronoLocal::new("%H:%M:%S%.3f".to_string()))
        .with_writer(writer)
        .with_filter(tracing_subscriber::filter::filter_fn(move |metadata| {
            if errors_only {
                is_error_level(metadata)
            } else {
                !is_error_level(metadata)
            }
        }))
}

#[cfg(feature = "tokio-console")]
fn tokio_console_env_opt_in() -> bool {
    tokio_console::env_opt_in()
}

/// Tokio console layer with env-based tuning, but completed-task retention never below one hour.
#[cfg(feature = "tokio-console")]
fn tokio_console_subscriber_layer<S>()
-> impl tracing_subscriber::layer::Layer<S> + Send + Sync + 'static
where
    S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
{
    tokio_console::subscriber_layer()
}

#[cfg(feature = "tokio-console")]
fn tokio_console_trace_targets_filter() -> tracing_subscriber::filter::Targets {
    tokio_console::trace_targets_filter()
}

fn init_tracing(enable_sentry_layer: bool) {
    #[cfg(feature = "tokio-console")]
    let console_opt_in: bool = tokio_console_env_opt_in();

    let stdout_filter: EnvFilter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new(DEFAULT_TRACING_FILTER));
    let no_color: bool = env::var("NO_COLOR").is_ok();
    let ansi: bool = !no_color
        && env::var("ATHENA_ANSI")
            .ok()
            .and_then(|v| match v.as_str() {
                "1" | "true" | "yes" => Some(true),
                "0" | "false" | "no" => Some(false),
                _ => None,
            })
            .unwrap_or_else(|| std::io::stderr().is_terminal());

    let (log_dir, file_log_warning) = resolve_tracing_log_directory();
    let mut file_log_guards: Vec<WorkerGuard> = Vec::new();
    let file_writers: Option<(NonBlocking, NonBlocking)> = if let Some(log_dir) = log_dir.as_ref() {
        let success_appender: tracing_appender::rolling::RollingFileAppender =
            tracing_appender::rolling::never(log_dir, SUCCESS_LOG_FILE_NAME);
        let (success_writer, success_guard) = tracing_appender::non_blocking(success_appender);
        file_log_guards.push(success_guard);

        let error_appender: tracing_appender::rolling::RollingFileAppender =
            tracing_appender::rolling::never(log_dir, ERROR_LOG_FILE_NAME);
        let (error_writer, error_guard) = tracing_appender::non_blocking(error_appender);
        file_log_guards.push(error_guard);

        Some((success_writer, error_writer))
    } else {
        None
    };

    if !file_log_guards.is_empty() {
        if let Ok(mut guards) = FILE_LOG_GUARDS.lock() {
            guards.extend(file_log_guards);
        } else {
            eprintln!("failed to retain tracing file log worker guards");
        }
    }

    #[cfg(feature = "tokio-console")]
    {
        if console_opt_in {
            match file_writers {
                Some((sw, ew)) => {
                    tracing_subscriber::registry()
                        .with(
                            tokio_console_subscriber_layer()
                                .with_filter(tokio_console_trace_targets_filter()),
                        )
                        .with(
                            athena_tracing_stdout_fmt_layer(ansi)
                                .with_filter(stdout_filter.clone()),
                        )
                        .with(
                            athena_tracing_file_layer(sw, false).with_filter(stdout_filter.clone()),
                        )
                        .with(
                            athena_tracing_file_layer(ew, true).with_filter(stdout_filter.clone()),
                        )
                        .with(
                            enable_sentry_layer.then_some(
                                sentry_tracing::layer().with_filter(stdout_filter.clone()),
                            ),
                        )
                        .init();
                }
                None => {
                    tracing_subscriber::registry()
                        .with(
                            tokio_console_subscriber_layer()
                                .with_filter(tokio_console_trace_targets_filter()),
                        )
                        .with(
                            athena_tracing_stdout_fmt_layer(ansi)
                                .with_filter(stdout_filter.clone()),
                        )
                        .with(
                            enable_sentry_layer.then_some(
                                sentry_tracing::layer().with_filter(stdout_filter.clone()),
                            ),
                        )
                        .init();
                }
            }
            info!(
                message = "tokio-console subscriber active (ATHENA_BENCHMARK_CLIENT=true); optional TOKIO_CONSOLE_BIND for listen address (default 127.0.0.1:6669); TOKIO_CONSOLE_RETENTION is floored at 1h for completed tasks; stdout/file logs use RUST_LOG (default info); Grafana URL is typically http://<host>:6669"
            );
        } else {
            match file_writers {
                Some((sw, ew)) => {
                    tracing_subscriber::registry()
                        .with(
                            athena_tracing_stdout_fmt_layer(ansi)
                                .with_filter(stdout_filter.clone()),
                        )
                        .with(
                            athena_tracing_file_layer(sw, false).with_filter(stdout_filter.clone()),
                        )
                        .with(
                            athena_tracing_file_layer(ew, true).with_filter(stdout_filter.clone()),
                        )
                        .with(
                            enable_sentry_layer.then_some(
                                sentry_tracing::layer().with_filter(stdout_filter.clone()),
                            ),
                        )
                        .init();
                }
                None => {
                    tracing_subscriber::registry()
                        .with(
                            athena_tracing_stdout_fmt_layer(ansi)
                                .with_filter(stdout_filter.clone()),
                        )
                        .with(
                            enable_sentry_layer.then_some(
                                sentry_tracing::layer().with_filter(stdout_filter.clone()),
                            ),
                        )
                        .init();
                }
            }
        }
    }
    #[cfg(not(feature = "tokio-console"))]
    {
        match file_writers {
            Some((sw, ew)) => {
                tracing_subscriber::registry()
                    .with(athena_tracing_stdout_fmt_layer(ansi).with_filter(stdout_filter.clone()))
                    .with(athena_tracing_file_layer(sw, false).with_filter(stdout_filter.clone()))
                    .with(athena_tracing_file_layer(ew, true).with_filter(stdout_filter.clone()))
                    .with(
                        enable_sentry_layer
                            .then_some(sentry_tracing::layer().with_filter(stdout_filter.clone())),
                    )
                    .init();
            }
            None => {
                tracing_subscriber::registry()
                    .with(athena_tracing_stdout_fmt_layer(ansi).with_filter(stdout_filter.clone()))
                    .with(
                        enable_sentry_layer
                            .then_some(sentry_tracing::layer().with_filter(stdout_filter.clone())),
                    )
                    .init();
            }
        }
    }

    if let Some(message) = file_log_warning {
        warn!(message = %message, "tracing file logging status");
    }
}

const BETTERSTACK_SENTRY_DSN: &str =
    "https://mMR1Bs5K6vSzXT8YGYyUxQSE@s1741777.eu-fsn-3.betterstackdata.com/1741777";

/// `init_sentry` initializes the Sentry client.
///
/// # Returns   
/// A `sentry::ClientInitGuard` if the Sentry client is initialized successfully, otherwise `None`.
fn init_sentry() -> Option<sentry::ClientInitGuard> {
    let dsn_source: String =
        env::var("SENTRY_DSN").unwrap_or_else(|_| BETTERSTACK_SENTRY_DSN.to_string());

    if dsn_source.trim().is_empty() {
        return None;
    }

    let dsn: SentryDsn = match dsn_source.parse::<SentryDsn>() {
        Ok(dsn) => dsn,
        Err(err) => {
            eprintln!("failed to parse Sentry DSN: {err}");
            return None;
        }
    };

    let mut options: sentry::ClientOptions = sentry::ClientOptions {
        dsn: Some(dsn),
        release: Some(env!("CARGO_PKG_VERSION").into()),
        environment: env::var("SENTRY_ENVIRONMENT").ok().map(Into::into),
        attach_stacktrace: true,
        ..Default::default()
    };

    if let Ok(value) = env::var("SENTRY_SAMPLE_RATE")
        && let Ok(parsed) = value.parse::<f32>()
    {
        options.sample_rate = parsed;
    }

    Some(sentry::init(options))
}

fn format_attempted_locations(locations: &[ConfigLocation]) -> String {
    locations
        .iter()
        .map(|location| format!("- {}", location.describe()))
        .collect::<Vec<_>>()
        .join("\n")
}
