appks/main.rs

use std::sync::Arc;

use actix_web::cookie::Key;
use actix_web::{App, HttpResponse, HttpServer, web};
use appks::api::openapi::OpenApiDoc;
use appks::api::routes::init_routes;
use appks::cache::AppCache;
use appks::cache::redis::AppRedis;
use appks::config::AppConfig;
use appks::error::{AppError, AppResult};
use appks::etcd::EtcdRegistry;
use appks::models::db::AppDatabase;
use appks::queue::NatsQueue;
use appks::service::AppService;
use appks::session::RedisSessionStore;
use appks::storage::s3::AppS3Storage;
use sqlx::Executor;
use utoipa::OpenApi;

#[actix_web::main]
async fn main() -> AppResult<()> {
    tracing_subscriber::fmt::init();

    let config = AppConfig::load();
    validate_session_secret(&config)?;

    tracing::info!("starting AppKS");

    let db = AppDatabase::from_config(&config).await?;
    db.writer().execute("SELECT 1").await?;
    sqlx::migrate!("./migrate")
        .run(db.writer())
        .await
        .map_err(|e| AppError::Config(format!("database migration failed: {e}")))?;

    let redis = AppRedis::from_config(&config).await?;
    let cache = Arc::new(AppCache::from_config(&config).await?);
    let storage = AppS3Storage::from_config(&config).await?;

    let registry = Arc::new(EtcdRegistry::connect(&config).await?);
    registry.start_discovery().await?;
    if config.get_env_or("APP_ETCD_REGISTER_SELF", false)? {
        registry
            .register_self(&config.rpc_self_service_name()?)
            .await?;
    }

    let nats = Arc::new(NatsQueue::connect(&config).await?);

    let service = AppService::new(
        env!("CARGO_PKG_VERSION").to_string(),
        db.clone(),
        redis.clone(),
        cache,
        config.clone(),
        storage,
        registry,
        nats,
    );

    let rpc_host = config.get_env_or::<String>("APP_RPC_SELF_HOST", "0.0.0.0".to_string())?;
    let rpc_port = config.get_env_or::<u16>("APP_RPC_SELF_PORT", 50050)?;
    let rpc_addr: std::net::SocketAddr = format!("{rpc_host}:{rpc_port}").parse()
        .map_err(|e| appks::error::AppError::Config(format!("invalid gRPC address: {e}")))?;
    let grpc_service = service.clone();
    tokio::spawn(async move {
        if let Err(e) = appks::grpc::start_grpc_server(rpc_addr, grpc_service).await {
            tracing::error!(error = %e, "gRPC server failed");
        }
    });

    let host = config.get_env_or::<String>("APP_HTTP_HOST", "0.0.0.0".to_string())?;
    let port = config.get_env_or::<u16>("APP_HTTP_PORT", 8000)?;
    let workers = config.get_env_or::<usize>(
        "APP_HTTP_WORKERS",
        std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(1),
    )?;
    let bind_addr = format!("{host}:{port}");
    let session_key = build_session_key(&config)?;
    let session_cfg = config.session_config()?;

    tracing::info!(addr = %bind_addr, workers, "http server listening");

    HttpServer::new(move || {
        let session_store = RedisSessionStore::new(redis.clone());
        let session_middleware = session_cfg.build_middleware(session_store, session_key.clone());

        App::new()
            .wrap(actix_web::middleware::Logger::default())
            .app_data(web::Data::new(service.clone()))
            .wrap(session_middleware)
            .route("/healthz", web::get().to(healthz))
            .route("/readyz", web::get().to(readyz))
            .route("/openapi.json", web::get().to(openapi_json))
            .configure(init_routes)
    })
    .workers(workers)
    .bind(bind_addr)?
    .run()
    .await?;

    db.close().await;
    Ok(())
}

async fn healthz() -> HttpResponse {
    HttpResponse::Ok().json(serde_json::json!({ "status": "ok" }))
}

async fn readyz(service: web::Data<AppService>) -> Result<HttpResponse, AppError> {
    service.ctx.db.writer().execute("SELECT 1").await?;
    Ok(HttpResponse::Ok().json(serde_json::json!({ "status": "ready" })))
}

async fn openapi_json() -> HttpResponse {
    HttpResponse::Ok().json(OpenApiDoc::openapi())
}

fn build_session_key(config: &AppConfig) -> AppResult<Key> {
    let secret = session_secret(config)?;
    Ok(Key::derive_from(secret.as_bytes()))
}

fn validate_session_secret(config: &AppConfig) -> AppResult<()> {
    session_secret(config).map(|_| ())
}

fn session_secret(config: &AppConfig) -> AppResult<String> {
    let secret = config
        .env
        .get("APP_SESSION_SECRET")
        .map(|s| s.trim())
        .filter(|s| s.len() >= 32)
        .ok_or_else(|| AppError::Config("APP_SESSION_SECRET must be at least 32 bytes".into()))?;
    Ok(secret.to_string())
}