#![cfg_attr(not(debug_assertions), windows_subsystem = "windows")]
use actix_cors::Cors;
use actix_web::body::MessageBody;
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::{AddrInUse, Other};
use std::io::IsTerminal;
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::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::metrics::prometheus_metrics;
use athena_rs::api::pipelines::{list_pipeline_templates, run_pipeline, simulate_pipeline};
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, 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::daemon::{spawn_connection_monitor, spawn_vacuum_health_collector};
use athena_rs::parser::{parse_secs_or_default, parse_usize};
use athena_rs::utils::pg_tools::ensure_pg_tools;
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);
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.",
))
}
}
}
}
}
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(),
))
}
})
}
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_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_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());
spawn_vacuum_health_collector(app_state.clone());
let cors_allow_any_origin: bool = config.get_cors_allow_any_origin();
let cors_allowed_origins: Vec<String> = 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 request_bytes = req
.headers()
.get(header::CONTENT_LENGTH)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse::<u64>().ok());
let athena_client = req
.headers()
.get("X-Athena-Client")
.and_then(|value| value.to_str().ok())
.map(str::to_string);
let started: Instant = Instant::now();
if let Some(state) = &metrics_state {
state.metrics_state.begin_http_request(&method, &route);
}
let fut = srv.call(req);
async move {
match fut.await {
Ok(mut res) => {
res.headers_mut()
.insert(header::SERVER, "XYLEX/0".parse().unwrap());
if let Some(state) = metrics_state {
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());
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)
}
}
}
})
.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(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_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
}
fn init_tracing(enable_sentry_layer: bool) {
let filter: EnvFilter =
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
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 fmt_layer = tracing_subscriber::fmt::layer()
.with_ansi(ansi)
.with_level(true)
.with_target(true)
.with_timer(ChronoLocal::new("%H:%M:%S%.3f".to_string()));
let base: tracing_subscriber::layer::Layered<
tracing_subscriber::fmt::Layer<
tracing_subscriber::layer::Layered<EnvFilter, tracing_subscriber::Registry>,
tracing_subscriber::fmt::format::DefaultFields,
tracing_subscriber::fmt::format::Format<
tracing_subscriber::fmt::format::Full,
ChronoLocal,
>,
>,
tracing_subscriber::layer::Layered<EnvFilter, tracing_subscriber::Registry>,
> = 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")
}