//! 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::body::{BoxBody, EitherBody};
use actix_web::dev::{Service, ServiceResponse};
use actix_web::http::{StatusCode, header};
use actix_web::{App, HttpRequest, HttpResponse, HttpServer, Responder, get, web};
use clap::Parser;
use dotenv::dotenv;
use moka::future::Cache;
use scylla::client::session::Session;
use scylla::client::session_builder::SessionBuilder;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::env;
use std::future::Future;
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::sync::Arc;
use std::thread::available_parallelism;
use std::time::{Duration, Instant};
use tracing::info;
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::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::pipelines::run_pipeline;
use athena_rs::api::query::sql::sql_query;
use athena_rs::api::registry::{api_registry, api_registry_by_id};
use athena_rs::api::supabase::ssl_enforcement;
use athena_rs::api::{athena_docs, schema};
use athena_rs::api::{athena_openapi_host, athena_router_registry};
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::parser::{parse_secs_or_default, parse_usize};
use athena_rs::utils::request_logging::{log_operation_event, log_request};

#[cfg(test)]
mod tests {
    use super::*;
    use actix_web::web::Data;
    use actix_web::{App, test};
    use athena_rs::drivers::postgresql::sqlx_driver::PostgresClientRegistry;
    use reqwest::Client;
    use serde_json::Value;

    #[actix_web::test]
    async fn test_ping_endpoint() {
        let cache: Arc<Cache<String, Value>> = Arc::new(
            Cache::builder()
                .time_to_live(Duration::from_secs(60))
                .build(),
        );
        let immortal_cache: Arc<Cache<String, Value>> = Arc::new(Cache::builder().build());
        let client = Client::builder()
            .pool_idle_timeout(Duration::from_secs(90))
            .build()
            .expect("Failed to build HTTP client");

        let app_state: AppState = AppState {
            cache,
            immortal_cache,
            client,
            pg_registry: Arc::new(PostgresClientRegistry::empty()),
            gateway_force_camel_case_to_snake_case: false,
            pipeline_registry: None,
            logging_client_name: None,
        };

        let app = test::init_service(App::new().app_data(Data::new(app_state)).service(ping)).await;
        let req = test::TestRequest::get().uri("/").to_request();
        let resp: ServiceResponse = test::call_service(&app, req).await;

        assert!(resp.status().is_success());

        let body: Value = test::read_body_json(resp).await;
        assert_eq!(body["message"], "athena is online");
        assert_eq!(body["cargo_toml_version"], env!("CARGO_PKG_VERSION"));
    }

    #[test]
    async fn test_app_state_creation() {
        let cache: Arc<Cache<String, Value>> = Arc::new(
            Cache::builder()
                .time_to_live(Duration::from_secs(60))
                .build(),
        );
        let immortal_cache: Arc<Cache<String, Value>> = Arc::new(Cache::builder().build());
        let client: Client = Client::builder()
            .pool_idle_timeout(Duration::from_secs(90))
            .build()
            .expect("Failed to build HTTP client");

        let app_state: AppState = AppState {
            cache,
            immortal_cache,
            client,
            pg_registry: Arc::new(PostgresClientRegistry::empty()),
            gateway_force_camel_case_to_snake_case: false,
            pipeline_registry: None,
            logging_client_name: None,
        };

        // Test that the app state was created successfully
        assert!(app_state.cache.weighted_size() == 0);
        assert!(app_state.immortal_cache.weighted_size() == 0);
    }

    #[tokio::test]
    async fn test_cache_expiration() {
        let cache: Arc<Cache<String, Value>> = Arc::new(
            Cache::builder()
                .time_to_live(Duration::from_millis(100))
                .build(),
        );

        let key: String = "expiring_key".to_string();
        let value: Value = json!({"expires": true});

        cache.insert(key.clone(), value.clone()).await;

        // Should exist immediately
        assert!(cache.get(&key).await.is_some());

        // Wait for expiration
        tokio::time::sleep(Duration::from_millis(150)).await;

        // Should be expired now
        assert!(cache.get(&key).await.is_none());
    }
}

/// Builds the canonical route listing surfaced by `/`.
///
/// The list is PostgREST-style and includes the HTTP verbs exposed by Athena so
/// clients can discover endpoints without consulting the OpenAPI spec. The list
/// is embedded in the `/` health response along with the service health bits.
fn api_routes() -> Value {
    json!([
        { "path": "/", "methods": ["GET"], "summary": "API root and route listing" },
        { "path": "/query/sql", "methods": ["POST"], "summary": "Execute SQL using selected driver" },
        { "path": "/gateway/data", "methods": ["POST"], "summary": "Fetch data with conditions (table, columns, conditions)" },
        { "path": "/gateway/fetch", "methods": ["POST"], "summary": "Alias for POST /gateway/data" },
        { "path": "/gateway/update", "methods": ["POST"], "summary": "Fetch/update (same payload as /gateway/data)" },
        { "path": "/data", "methods": ["GET"], "summary": "Fetch data via GET (view, eq_column, eq_value, etc.)" },
        { "path": "/gateway/insert", "methods": ["PUT"], "summary": "Insert a row into a table" },
        { "path": "/gateway/delete", "methods": ["DELETE"], "summary": "Delete a row by table_name and resource_id" },
        { "path": "/gateway/query", "methods": ["POST"], "summary": "Run raw SQL against the selected PostgreSQL pool" },
        { "path": "/pipelines", "methods": ["POST"], "summary": "Run a config-driven pipeline (source → transform → sink)" },
        { "path": "/rest/v1/{table}", "methods": ["GET", "POST", "PATCH", "DELETE"], "summary": "Supabase-compatible REST gateway" },
        { "path": "/router/registry", "methods": ["GET"], "summary": "List Athena router registry entries" },
        { "path": "/registry", "methods": ["GET"], "summary": "List API registry entries" },
        { "path": "/registry/{api_registry_id}", "methods": ["GET"], "summary": "Get API registry entry by id" },
        { "path": "/schema/clients", "methods": ["GET"], "summary": "List configured Postgres clients" },
        { "path": "/schema/tables", "methods": ["GET"], "summary": "List tables (requires X-Athena-Client)" },
        { "path": "/schema/columns", "methods": ["GET"], "summary": "List columns for a table (requires X-Athena-Client, query: table_name)" },
        { "path": "/api/v2/supabase/ssl_enforcement", "methods": ["POST"], "summary": "Toggle Supabase SSL enforcement for a project" }
    ])
}

#[get("/")]
/// Health endpoint that reports service counters, Scylla status, and the route catalogue.
///
/// Responds with `athena is online`, the Cargo package version, and the same route
/// metadata returned by `api_routes`. The handler also logs each request and
/// records a `ping` operation event so tracing tools can correlate health checks.
async fn ping(req: HttpRequest, data: web::Data<AppState>) -> impl Responder {
    let logged_request = log_request(req.clone(), Some(data.get_ref()));
    let operation_start = Instant::now();

    info!("Received health check request");

    let scy_ok: bool = cached_health_status(&data.cache, "scylla_health", || check_scylla()).await;
    let scy_status: &str = if scy_ok { "online" } else { "offline" };
    let cargo_version: &str = env!("CARGO_PKG_VERSION");
    let routes: Value = api_routes();

    log_operation_event(
        Some(data.get_ref()),
        &logged_request,
        "ping",
        None,
        operation_start.elapsed().as_millis(),
        StatusCode::OK,
        Some(json!({
            "route_count": routes.as_array().map(|arr| arr.len()).unwrap_or(0),
            "scylla_status": scy_status
        })),
    );

    HttpResponse::Ok().json(json!({
        "message": "athena is online",
        "version": cargo_version,
        "athena_api": "online",
        "athena_deadpool": "online",
        "athena_scylladb": scy_status,
        "cargo_toml_version": cargo_version,
        "routes": routes
    }))
}

/// Caches expensive boolean checks inside the shared `AppState` cache.
///
/// The closure used to drive the check runs once per cache miss and its boolean
/// result is stored under `cache_key`. Subsequent callers read the cached
/// value until the entry expires.
async fn cached_health_status<F, Fut>(
    cache: &Arc<Cache<String, Value>>,
    cache_key: &str,
    check_fn: F,
) -> bool
where
    F: FnOnce() -> Fut,
    Fut: Future<Output = bool>,
{
    if let Some(Value::Bool(cached)) = cache.get(cache_key).await {
        return cached;
    }

    let status: bool = check_fn().await;
    cache
        .insert(cache_key.to_string(), Value::Bool(status))
        .await;
    status
}

/// Performs a lightweight ScyllaDB connection probe via `SELECT 1`.
///
/// Uses the configured `scylladb` host plus authenticator and returns `true`
/// when the session is established and the probe query succeeds.
async fn check_scylla() -> bool {
    let config: Config = Config::load().expect("Failed to load config.yaml");
    let uri: String = config
        .get_host("scylladb")
        .map(|s| s.clone())
        .unwrap_or_else(|| "127.0.0.1:9042".to_string());
    let empty_map: HashMap<String, String> = HashMap::new();
    let authenticator: &HashMap<String, String> =
        config.get_authenticator("scylladb").unwrap_or(&empty_map);
    let empty_string: String = String::new();
    let session: Session = match SessionBuilder::new()
        .known_node(uri)
        .user(
            authenticator.get("username").unwrap_or(&empty_string),
            authenticator.get("password").unwrap_or(&empty_string),
        )
        .build()
        .await
    {
        Ok(session) => session,
        Err(_) => return false,
    };
    session.query_unpaged("SELECT 1", &[]).await.is_ok()
}

#[actix_web::main]
async fn main() -> IoResult<()> {
    dotenv().ok();
    let sentry_guard = init_sentry();
    init_tracing(sentry_guard.is_some());
    let _sentry_guard = sentry_guard;

    let AthenaCli {
        config_path,
        pipelines_path,
        port: port_override,
        command,
        api_only,
    } = AthenaCli::parse();

    let config_overridden = config_path.is_some();
    let config_path = config_path
        .clone()
        .unwrap_or_else(|| PathBuf::from(DEFAULT_CONFIG_FILE_NAME));
    let pipelines_path = pipelines_path.to_string_lossy().to_string();
    let config = if config_overridden {
        Config::load_from(&config_path).map_err(|err| {
            let attempted_locations = 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 = 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 = 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 = 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::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 => {
            if api_only {
                let 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.",
                ))
            }
        }
    }
}

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 =
        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(parse_usize)
        .unwrap_or_else(|| available_parallelism().map(|n| n.get()).unwrap_or(4));
    let max_connections: usize = config
        .get_http_max_connections()
        .and_then(parse_usize)
        .unwrap_or(10_000);
    let backlog: usize = config
        .get_http_backlog()
        .and_then(parse_usize)
        .unwrap_or(2_048);
    let tcp_keepalive: Duration = parse_secs_or_default(config.get_tcp_keepalive_secs(), 75);

    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: web::Data<AppState> = bootstrap.app_state.clone();

    HttpServer::new(move || {
        let cors: Cors = Cors::default()
            .allow_any_origin()
            .allow_any_method()
            .allow_any_header();
        App::new()
            .wrap(cors)
            .wrap_fn(|req, srv| {
                let fut = srv.call(req);
                async move {
                    let mut res: ServiceResponse<EitherBody<BoxBody>> = fut.await?;
                    res.headers_mut()
                        .insert(header::SERVER, "XYLEX/0".parse().unwrap());
                    Ok(res)
                }
            })
            .app_data(app_state.clone())
            .service(ping)
            .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(api_registry)
            .service(athena_docs)
            .service(api_registry_by_id)
            .configure(schema::services)
            .service(ssl_enforcement)
    })
    .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.
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_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 = env::var("SENTRY_DSN").unwrap_or_else(|_| BETTERSTACK_SENTRY_DSN.to_string());

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

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

    let mut options = 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") {
        if 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")
}
