//! 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::dev::{Service, ServiceResponse};
use actix_web::http::StatusCode;
use actix_web::http::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 std::env;
use std::io::Error;
use std::io::ErrorKind::Other;
use std::io::Result as IoResult;
use std::net::{SocketAddr, TcpListener};
use std::path::PathBuf;
use std::thread::available_parallelism;
use std::time::{Duration, Instant};
use tracing::info;
use tracing::warn;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::fmt::time::ChronoLocal;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

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

use athena_rs::AppState;
use athena_rs::api::backup;
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::health::{cluster_health, ping, root};
use athena_rs::api::management;
use athena_rs::api::metrics::{prometheus_metrics, record_http_metric};
use athena_rs::api::pipelines::run_pipeline;
use athena_rs::api::provision;
use athena_rs::api::query::sql::sql_query;
use athena_rs::api::registry::{api_registry, api_registry_by_id};
use athena_rs::api::response::{internal_error, not_found};
use athena_rs::api::supabase::ssl_enforcement;
use athena_rs::api::{admin, athena_docs, schema, storage};
use athena_rs::api::{athena_openapi_host, athena_router_registry, athena_wss_openapi_host};
use athena_rs::bootstrap::{Bootstrap, build_shared_state};
use athena_rs::cli::{self, AthenaCli, Command};
use athena_rs::config::{Config, ConfigLocation, DEFAULT_CONFIG_FILE_NAME};
use athena_rs::daemon::spawn_connection_monitor;
use athena_rs::parser::{parse_secs_or_default, parse_usize};
use athena_rs::wss::gateway_wss_info;

#[actix_web::main]
async fn main() -> IoResult<()> {
    dotenv().ok();
    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 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: Config = if config_overridden {
        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)
                ),
            )
        })?
    } else {
        match Config::load_default() {
            Ok(outcome) => {
                if outcome.seeded_default {
                    info!("Seeded default configuration at {}", outcome.path.display());
                }
                outcome.config
            }
            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)
                    ),
                ));
            }
        }
    };

    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).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);
                    athena_rs::cdc::websocket::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).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).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: actix_web::body::MessageBody + 'static,
{
    ErrorHandlers::new()
        .handler(StatusCode::NOT_FOUND, |res: ServiceResponse<B>| {
            let (req, _) = res.into_parts();
            let resp = 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 (req, _) = res.into_parts();
            let resp = internal_error("Internal Server Error", "An unexpected error occurred");
            Ok(ErrorHandlerResponse::Response(
                ServiceResponse::new(req, resp)
                    .map_into_boxed_body()
                    .map_into_right_body(),
            ))
        })
}

async fn run_server(
    config: &Config,
    bootstrap: Bootstrap,
    port_override: Option<u16>,
) -> IoResult<()> {
    let port: u16 = if let Some(port) = port_override {
        port
    } else {
        config
            .get_api()
            .ok_or("No API port configured")
            .and_then(|port_str| port_str.parse().map_err(|_| "Invalid port number"))
            .expect("Failed to parse API port")
    };

    let keep_alive: Duration = parse_secs_or_default(config.get_http_keep_alive_secs(), 15);
    let client_disconnect_timeout: Duration =
        parse_secs_or_default(config.get_client_disconnect_timeout_secs(), 60);
    let client_request_timeout =
        parse_secs_or_default(config.get_client_request_timeout_secs(), 60);
    let worker_count: usize = config
        .get_http_workers()
        .and_then(|v| parse_usize(v.as_str()))
        .unwrap_or_else(|| available_parallelism().map(|n| n.get()).unwrap_or(4));
    let max_connections: usize = config
        .get_http_max_connections()
        .and_then(|v| parse_usize(v.as_str()))
        .unwrap_or(10_000);
    let backlog: usize = config
        .get_http_backlog()
        .and_then(|v| parse_usize(v.as_str()))
        .unwrap_or(2_048);
    let tcp_keepalive: Duration = parse_secs_or_default(config.get_tcp_keepalive_secs(), 75);
    let prometheus_metrics_enabled: bool = config.get_prometheus_metrics_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_keepalive(true)?;
    let keepalive_cfg: TcpKeepalive = TcpKeepalive::new().with_time(tcp_keepalive);
    socket.set_tcp_keepalive(&keepalive_cfg)?;
    socket.bind(&addr.into())?;
    let listen_backlog: i32 = backlog.min(i32::MAX as usize) as i32;
    socket.listen(listen_backlog)?;
    let listener: TcpListener = socket.into();

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

    let cors_allow_any_origin = config.get_cors_allow_any_origin();
    let cors_allowed_origins = config.get_cors_allowed_origins();

    if cors_allow_any_origin {
        warn!(
            "CORS is configured with allow_any_origin=true; consider using api.cors_allowed_origins for tighter access control"
        );
    } else if cors_allowed_origins.is_empty() {
        warn!(
            "CORS allow_any_origin is false but no cors_allowed_origins were configured; cross-origin browser requests will be denied"
        );
    }

    HttpServer::new(move || {
        let mut cors: Cors = Cors::default().allow_any_method().allow_any_header();
        if cors_allow_any_origin {
            cors = cors.allow_any_origin();
        } else {
            for origin in &cors_allowed_origins {
                cors = cors.allowed_origin(origin);
            }
        }
        let mut app = App::new()
            .wrap(default_error_handlers())
            .wrap(cors)
            .wrap_fn(|req, srv| {
                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 started: Instant = Instant::now();
                let fut = srv.call(req);
                async move {
                    let mut res = fut.await?;
                    res.headers_mut()
                        .insert(header::SERVER, "XYLEX/0".parse().unwrap());
                    if let Some(state) = metrics_state {
                        record_http_metric(
                            state.get_ref(),
                            &method,
                            &route,
                            res.status().as_u16(),
                            started.elapsed().as_secs_f64() * 1000.0,
                        );
                    }
                    Ok(res)
                }
            })
            .app_data(app_state.clone())
            .service(root)
            .service(ping)
            .service(cluster_health)
            .service(sql_query)
            .service(fetch_data_route)
            .service(get_data_route)
            .service(proxy_fetch_data_route)
            .service(gateway_update_route)
            .service(gateway_query_route)
            .service(insert_data)
            .service(delete_data)
            .service(postgrest_get_route)
            .service(postgrest_post_route)
            .service(postgrest_patch_route)
            .service(postgrest_delete_route)
            .service(run_pipeline)
            .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_wss_info)
            .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 {
            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.
/// Uses ANSI colors for level/target and forces ANSI on for richer terminal output.
fn init_tracing(enable_sentry_layer: bool) {
    let filter: EnvFilter =
        EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
    let fmt_layer = tracing_subscriber::fmt::layer()
        .with_ansi(true)
        .with_level(true)
        .with_target(true)
        .with_timer(ChronoLocal::new("%H:%M:%S%.3f".to_string()));

    let base = tracing_subscriber::registry().with(filter).with(fmt_layer);

    if enable_sentry_layer {
        base.with(sentry_tracing::layer()).init();
    } else {
        base.init();
    }
}

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

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")
}
