backend/

main.rs

//! The backend of `PermaplanT`.

#![recursion_limit = "1024"]
// Enable all lints apart from clippy::restriction by default.
// See https://rust-lang.github.io/rust-clippy/master/index.html#blanket_clippy_restriction_lints for as to why restriction is not enabled.
#![warn(clippy::pedantic)]
#![warn(clippy::nursery)]
#![warn(clippy::cargo)]
// Lints in clippy::restriction which seem useful.
#![warn(
    clippy::clone_on_ref_ptr,
    clippy::empty_structs_with_brackets,
    clippy::exit,
    clippy::expect_used,
    clippy::format_push_string,
    clippy::get_unwrap,
    clippy::if_then_some_else_none,
    clippy::indexing_slicing,
    clippy::integer_division,
    clippy::implicit_clone,
    clippy::large_include_file,
    clippy::missing_docs_in_private_items,
    clippy::mixed_read_write_in_expression,
    clippy::multiple_inherent_impl,
    clippy::mutex_atomic,
    clippy::panic_in_result_fn,
    clippy::partial_pub_fields,
    clippy::print_stderr,
    clippy::print_stdout,
    clippy::rc_buffer,
    clippy::rc_mutex,
    clippy::rest_pat_in_fully_bound_structs,
    clippy::same_name_method,
    clippy::shadow_unrelated,
    clippy::str_to_string,
    clippy::suspicious_xor_used_as_pow,
    clippy::todo,
    clippy::try_err,
    clippy::unimplemented,
    clippy::unnecessary_self_imports,
    clippy::unneeded_field_pattern,
    clippy::unreachable,
    clippy::unseparated_literal_suffix,
    clippy::unwrap_in_result,
    clippy::unwrap_used,
    clippy::use_debug,
    clippy::verbose_file_reads
)]
// Cannot fix some errors because dependencies import them.
#![allow(clippy::multiple_crate_versions)]
// Clippy suggests lots of false "x.get(0)" => "x.first()"
#![allow(clippy::get_first)]
// We often want the same name per module (for instance every enum).
#![allow(clippy::module_name_repetitions)]

use actix_cors::Cors;
use actix_web::{http, middleware::Logger, App, HttpServer};
use config::{api_doc, auth::Config, routes};
use db::{
    connection::Pool,
    cronjobs::{cleanup_layers, cleanup_maps},
};
use std::sync::Arc;

use opentelemetry::propagation::TextMapCompositePropagator;
use opentelemetry::{global, KeyValue};
use opentelemetry_instrumentation_actix_web::{RequestMetrics, RequestTracing};
use opentelemetry_resource_detectors::{
    HostResourceDetector, OsResourceDetector, ProcessResourceDetector,
};
use opentelemetry_sdk::metrics::SdkMeterProvider;
use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator};
use opentelemetry_sdk::trace::SdkTracerProvider;
use opentelemetry_sdk::Resource;

use crate::config::app::Mode;

pub mod config;
pub mod controller;
pub mod db;
pub mod error;
pub mod keycloak_api;
pub mod model;
/// Auto generated by diesel.
#[allow(clippy::wildcard_imports)]
#[allow(clippy::missing_docs_in_private_items)]
pub mod schema;
pub mod service;
pub mod sse;
#[cfg(test)]
pub mod test;

/// OpenTelemetry resource describing this service.
static RESOURCE: std::sync::LazyLock<Resource> = std::sync::LazyLock::new(|| {
    const NAME: &str = env!("CARGO_PKG_NAME");
    const VERSION: &str = env!("CARGO_PKG_VERSION");
    const BUILD_DATE: &str = env!("BUILD_DATE");
    Resource::builder()
        .with_detector(Box::new(OsResourceDetector))
        .with_detector(Box::new(HostResourceDetector::default()))
        .with_detector(Box::new(ProcessResourceDetector))
        .with_attributes(vec![
            KeyValue::new("service.name", NAME),
            KeyValue::new("service.version", VERSION),
            KeyValue::new("service.build_date", BUILD_DATE),
        ])
        .build()
});

/// Main function.
#[actix_web::main]
async fn main() -> std::io::Result<()> {
    env_logger::init();

    log::debug!("Initializing telemetry ...");

    // Initialize OpenTelemetry
    let tracer_provider = init_trace();
    let tracer_provider = match tracer_provider {
        Ok(tp) => tp,
        Err(e) => {
            log::error!("Error initializing OpenTelemetry tracing: {e}");
            std::process::exit(1);
        }
    };

    global::set_tracer_provider(tracer_provider.clone());

    let meter_provider = init_metrics();
    let meter_provider = match meter_provider {
        Ok(mp) => mp,
        Err(e) => {
            log::error!("Error initializing OpenTelemetry metrics: {e}");
            std::process::exit(1);
        }
    };

    global::set_meter_provider(meter_provider.clone());

    opentelemetry_instrumentation_tokio::observe_current_runtime();

    log::debug!("Starting PermaplanT backend...");

    let version = option_env!("CARGO_PKG_VERSION").unwrap_or("unknown");
    let build_date = option_env!("BUILD_DATE").unwrap_or("unknown");

    let config = match config::app::Config::from_env() {
        Ok(config) => config,
        Err(e) => {
            return Err(std::io::Error::other(format!(
                "Error reading configuration: {e}"
            )));
        }
    };

    log::info!("Configuration loaded: {config:#?}");

    Config::init(&config).await;

    log::info!("Version   : {version}");
    log::info!("Build date: {build_date}");

    let data_init = config::data::init(&config);
    let pool = data_init.pool.clone().into_inner();
    start_cronjobs(pool);

    log::info!(
        "Binding to: {}:{}",
        config.bind_address.0,
        config.bind_address.1
    );

    let broadcaster = data_init.broadcaster.clone();

    let mode = config.mode;

    let server = HttpServer::new(move || {
        App::new()
            .wrap(Logger::default())
            .wrap(RequestTracing::new())
            .wrap(RequestMetrics::default())
            .wrap(cors_configuration(mode))
            .app_data(data_init.pool.clone())
            .app_data(data_init.broadcaster.clone())
            .app_data(data_init.http_client.clone())
            .app_data(data_init.keycloak_api.clone())
            .app_data(data_init.mode.clone())
            .configure(routes::config)
            .configure(api_doc::config)
    })
    .disable_signals()
    .shutdown_timeout(5)
    .bind(config.bind_address)?
    .run();

    let srv_handle = server.handle();

    actix_web::rt::spawn(async move {
        if let Err(err) = shutdown_signal().await {
            log::error!("Failed to listen for shutdown signals: {err}");
            return;
        }
        log::info!("Received shutdown signal, notifying all connected clients");
        let action = model::dto::actions::Action::new_shutdown_signal_action(
            "Server is shutting down".to_owned(),
        );
        broadcaster.broadcast_all_maps(action).await;
        log::info!("Shutdown notification sent to all clients");
        srv_handle.stop(true).await;
    });

    server.await?;

    let _ = tracer_provider.shutdown();
    let _ = meter_provider.shutdown();

    Ok(())
}

/// Create a CORS configuration for the server.
fn cors_configuration(mode: Mode) -> Cors {
    let mut cors = Cors::default()
        // 1. Add methods and headers
        .allowed_methods(vec!["GET", "POST", "PUT", "PATCH", "DELETE"])
        .allowed_headers(vec![
            http::header::AUTHORIZATION,
            http::header::ACCEPT,
            http::header::CONTENT_TYPE,
        ])
        .allowed_header("traceparent")
        .allowed_header("tracestate")
        .max_age(3600);

    // 2. Define allowed origin
    match mode {
        Mode::Testing => {
            cors = cors.allow_any_origin();
        }
        Mode::Production => {
            cors = cors
                .allowed_origin("http://localhost:5173")
                .allowed_origin("https://cloud.perma.health");

            // 3. Programmatically add the repetitive subdomains
            let subdomains = ["mr", "dev", "master", "www", "experimental", "telemetry"];
            for sub in subdomains {
                cors = cors.allowed_origin(&format!("https://{sub}.permaplant.net"));
                cors = cors.allowed_origin(&format!("https://{sub}.staging.permaplant.net"));
            }
        }
    }

    cors
}

/// Start all scheduled jobs that get run in the backend.
fn start_cronjobs(pool: Arc<Pool>) {
    tokio::spawn(cleanup_maps(Arc::clone(&pool)));
    tokio::spawn(cleanup_layers(pool));
}

/// Initialize OpenTelemetry tracing.
fn init_trace() -> Result<SdkTracerProvider, Box<dyn std::error::Error + Send + Sync>> {
    let baggage_propagator = BaggagePropagator::new();
    let trace_context_propagator = TraceContextPropagator::new();
    let composite_propagator = TextMapCompositePropagator::new(vec![
        Box::new(baggage_propagator),
        Box::new(trace_context_propagator),
    ]);

    global::set_text_map_propagator(composite_propagator);

    let exporter = opentelemetry_otlp::SpanExporter::builder()
        .with_http()
        .build()?;

    let tr = SdkTracerProvider::builder()
        .with_batch_exporter(exporter)
        .with_resource(RESOURCE.clone())
        .build();
    Ok(tr)
}

/// Initialize OpenTelemetry metrics.
fn init_metrics() -> Result<SdkMeterProvider, Box<dyn std::error::Error + Send + Sync>> {
    let exporter = opentelemetry_otlp::MetricExporter::builder()
        .with_http()
        .build()?;

    let mt = SdkMeterProvider::builder()
        .with_periodic_exporter(exporter)
        .with_resource(RESOURCE.clone())
        .build();
    Ok(mt)
}

/// Wait for shutdown signal.
async fn shutdown_signal() -> std::io::Result<()> {
    let ctrl_c = async {
        tokio::signal::ctrl_c().await?;
        Ok::<(), std::io::Error>(())
    };

    let unix_signal = async {
        use tokio::signal::unix::{signal, SignalKind};

        let mut sigterm = signal(SignalKind::terminate())?;
        let mut sighup = signal(SignalKind::hangup())?;

        tokio::select! {
            _ = sigterm.recv() => {},
            _ = sighup.recv() => {},
        }

        Ok::<(), std::io::Error>(())
    };

    tokio::select! {
        res = ctrl_c => res?,
        res = unix_signal => res?,
    }

    Ok(())
}