noema-actix-webapi 0.1.0

use std::path::Path;
use std::sync::Arc;

use actix_web::{HttpResponse, web};
use noema::config::ConfigSource;
use noema::resolve;
use tracing_appender::non_blocking::WorkerGuard;
use tracing_subscriber::fmt;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::{EnvFilter, Layer};

use crate::clock::Clock;
use crate::config::{ApplicationConfig, DatabaseConfig, Environment, LogConfig};
use crate::db;
use crate::hasher::Hasher;
use crate::http_client::HttpClient;
use crate::logger::Logger;
use crate::modules::Modules;
use crate::ws::SessionHub;

#[derive(Debug)]
pub enum Error {
    Config(noema::ConfigError),
    Database(sqlx::Error),
    Io(std::io::Error),
}

impl std::fmt::Display for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Config(e) => write!(f, "{e}"),
            Self::Database(e) => write!(f, "{e}"),
            Self::Io(e) => write!(f, "{e}"),
        }
    }
}

impl std::error::Error for Error {}

impl From<noema::ConfigError> for Error {
    fn from(value: noema::ConfigError) -> Self {
        Self::Config(value)
    }
}

impl From<sqlx::Error> for Error {
    fn from(value: sqlx::Error) -> Self {
        Self::Database(value)
    }
}

impl From<std::io::Error> for Error {
    fn from(value: std::io::Error) -> Self {
        Self::Io(value)
    }
}

/// Infra + module catalog. Does not bind or run Actix.
#[derive(Clone)]
pub struct Application {
    modules: Arc<Modules>,
    bind: String,
    health: bool,
    ready: bool,
    _log_guard: Option<Arc<WorkerGuard>>,
}

impl Application {
    /// Load app `config!` types, install [`ApplicationConfig`], Postgres, migrators, infra, logger.
    /// No `HttpServer`. `DATABASE_URL` is required on `source`.
    pub async fn start(
        modules: Modules,
        source: impl ConfigSource + 'static,
        config: ApplicationConfig,
    ) -> Result<Self, Error> {
        let source: std::sync::Arc<dyn ConfigSource> = std::sync::Arc::new(source);
        source.load().await?;
        let url = source
            .get("DATABASE_URL")
            .await?
            .filter(|s| !s.is_empty())
            .ok_or_else(|| noema::ConfigError::MissingKey("DATABASE_URL".into()))?;

        match noema::load(source).await {
            Ok(()) => {}
            Err(noema::ConfigError::AlreadyLoaded(_)) => {}
            Err(err) => return Err(err.into()),
        }

        crate::config::install(config.clone());
        crate::config::install_environment(config.http.env);
        crate::http_client::install(config.http_client.clone());
        crate::cors::install(config.cors.clone());

        let bind = config.http.bind_addr();
        db::start(&DatabaseConfig {
            url,
            max_connections: config.database.max_connections,
            acquire_timeout_ms: config.database.acquire_timeout_ms,
        })
        .await?;
        let pool = resolve::<crate::PgPool>();
        modules.run_migrators(pool.connection()).await?;

        crate::ws::install_origin();

        modules.configure_infrastructure().await;
        let log_guard = init_logging(&config.log)?;
        let _ = resolve::<ApplicationConfig>();
        let _ = resolve::<SessionHub>();
        let _ = resolve::<dyn Logger + Send + Sync>();
        let _ = resolve::<dyn Hasher + Send + Sync>();
        let _ = resolve::<dyn Clock + Send + Sync>();
        let _ = resolve::<dyn HttpClient + Send + Sync>();

        Ok(Self {
            modules: Arc::new(modules),
            bind,
            health: false,
            ready: false,
            _log_guard: log_guard.map(Arc::new),
        })
    }

    /// `GET /health` — process is up.
    pub fn health(mut self) -> Self {
        self.health = true;
        self
    }

    /// `GET /ready` — Postgres `SELECT 1`.
    pub fn ready(mut self) -> Self {
        self.ready = true;
        self
    }

    pub fn bind_addr(&self) -> &str {
        &self.bind
    }

    /// Spawn process-lifetime loops. Does not run Actix.
    pub fn spawn_background_tasks(
        &self,
        tasks: impl IntoIterator<Item = std::sync::Arc<dyn crate::BackgroundTask + Send + Sync>>,
    ) {
        crate::spawn_background_tasks(tasks);
    }

    /// Opt-in WS bus: `broadcast_event` publishes `(room, WsFanoutMessage json)` after local send.
    pub fn on_ws_publish(&self, publish: impl Fn(String, String) + Send + Sync + 'static) {
        crate::ws::on_ws_publish(publish);
    }

    /// Mount module scopes, Swagger UI, and any opted-in probe routes onto **your** `App`.
    pub fn configure(&self, cfg: &mut web::ServiceConfig) {
        if self.health {
            cfg.route("/health", web::get().to(health));
        }
        if self.ready {
            cfg.route("/ready", web::get().to(ready));
        }
        self.modules.configure_presentation(cfg);
        self.modules.mount_swagger(cfg, Arc::clone(&self.modules));
    }
}

/// Same as [`Application::start`].
pub async fn start(
    modules: Modules,
    source: impl ConfigSource + 'static,
    config: ApplicationConfig,
) -> Result<Application, Error> {
    Application::start(modules, source, config).await
}

fn init_logging(settings: &LogConfig) -> Result<Option<WorkerGuard>, Error> {
    let file = settings.file.as_deref().filter(|p| !p.is_empty());
    let stdout = settings.stdout || file.is_none();
    let filter =
        EnvFilter::try_new(settings.level.as_str()).unwrap_or_else(|_| EnvFilter::new("info"));

    let stdout_layer = stdout.then(|| fmt_layer(settings.json, std::io::stdout));

    let mut guard = None;
    let file_layer = if let Some(path) = file {
        let (writer, g) = file_writer(path)?;
        guard = Some(g);
        Some(fmt_layer(settings.json, writer))
    } else {
        None
    };

    let _ = tracing_subscriber::registry()
        .with(filter)
        .with(stdout_layer)
        .with(file_layer)
        .try_init();

    Ok(guard)
}

fn fmt_layer<S, W>(json: bool, writer: W) -> Box<dyn Layer<S> + Send + Sync>
where
    S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
    W: for<'a> tracing_subscriber::fmt::MakeWriter<'a> + Send + Sync + 'static,
{
    let layer = fmt::layer().with_writer(writer);
    if json {
        layer.json().boxed()
    } else {
        layer.boxed()
    }
}

fn file_writer(
    path: &str,
) -> Result<(tracing_appender::non_blocking::NonBlocking, WorkerGuard), Error> {
    let path = Path::new(path);
    let dir = path
        .parent()
        .filter(|p| !p.as_os_str().is_empty())
        .unwrap_or_else(|| Path::new("."));
    let name = path
        .file_name()
        .unwrap_or_else(|| std::ffi::OsStr::new("app.log"));
    std::fs::create_dir_all(dir)?;
    let appender = tracing_appender::rolling::never(dir, name);
    Ok(tracing_appender::non_blocking(appender))
}

async fn health() -> HttpResponse {
    HttpResponse::Ok().body("ok")
}

async fn ready() -> HttpResponse {
    if !db::is_connected() {
        return HttpResponse::ServiceUnavailable().body("no database");
    }
    match db::ping().await {
        Ok(()) => HttpResponse::Ok().body("ready"),
        Err(e) => {
            let body = if Environment::current().exposes_infra_errors() {
                e.to_string()
            } else {
                "not ready".into()
            };
            HttpResponse::ServiceUnavailable().body(body)
        }
    }
}

#[cfg(test)]
pub(crate) fn test_source() -> noema::MapSource {
    let url = std::env::var("DATABASE_URL")
        .unwrap_or_else(|_| "postgres://noema:noema@127.0.0.1:5432/noema".into());
    noema::MapSource::new([("DATABASE_URL", url)])
}

#[cfg(test)]
mod tests {
    use super::*;
    use actix_web::{App, http::StatusCode, test};

    #[actix_web::test]
    async fn probes_are_opt_in() {
        let app = Application::start(Modules::new(), test_source(), ApplicationConfig::default())
            .await
            .expect("start");
        let srv = test::init_service(App::new().configure(|c| app.configure(c))).await;
        let resp =
            test::call_service(&srv, test::TestRequest::get().uri("/health").to_request()).await;
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);

        let app = app.health().ready();
        let srv = test::init_service(App::new().configure(|c| app.configure(c))).await;
        let health =
            test::call_service(&srv, test::TestRequest::get().uri("/health").to_request()).await;
        assert_eq!(health.status(), StatusCode::OK);
        let ready =
            test::call_service(&srv, test::TestRequest::get().uri("/ready").to_request()).await;
        assert_ne!(ready.status(), StatusCode::NOT_FOUND);
    }

    #[actix_web::test]
    async fn swagger_is_always_mounted() {
        let app = Application::start(Modules::new(), test_source(), ApplicationConfig::default())
            .await
            .expect("start");
        let srv = test::init_service(App::new().configure(|c| app.configure(c))).await;
        let resp = test::call_service(
            &srv,
            test::TestRequest::get().uri("/swagger-ui/").to_request(),
        )
        .await;
        assert_ne!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[actix_web::test]
    async fn start_requires_database_url() {
        let err = Application::start(
            Modules::new(),
            noema::MapSource::new(std::iter::empty::<(String, String)>()),
            ApplicationConfig::default(),
        )
        .await
        .err()
        .expect("missing url");
        assert!(err.to_string().contains("DATABASE_URL"));
    }
}