//! OxygenHub: a forge for Oxygen repositories (`docs/spec.md`).

//!

//! One process, or several sharing a database: an axum router serving

//! pages, the JSON API and the wire protocol (the web role); the background

//! tasks (the worker role, singletons under a leader lock: docs/roles.md);

//! SQLite or PostgreSQL for hub state; repositories as bare Oxygen stores.

pub mod api;

pub mod apt;

pub mod assistants;

pub mod auth;

pub mod autoland;

pub mod backups;

pub mod bus;

pub mod config;

pub mod coverage;

pub mod db;

pub mod diffusion;

pub mod events;

pub mod flow;

pub mod headers;

pub mod hooks;

pub mod images;

pub mod insights;

pub mod junit;

pub mod keyless;

pub mod keys;

pub mod leader;

pub mod levels;

pub mod live;

pub mod mail;

pub mod maintenance;

pub mod metrics;

pub mod mirror;

pub mod net;

pub mod notify;

pub mod oidc;

pub mod protocol;

pub mod provenance;

pub mod push_protection;

pub mod render;

pub mod reports;

pub mod repos;

pub mod secrets;

pub mod setup;

pub mod shutdown;

pub mod signin;

pub mod social;

pub mod storage;

pub mod swarm;

pub mod terraform;

pub mod throttle;

pub mod tlog;

pub mod variables;

pub mod web;

pub mod wiki;

use crate::config::Config;

use crate::repos::Repos;

use std::sync::Arc;

/// Everything a request handler can reach.

pub struct App {

    pub config: Config,

    pub db: db::Pool,

    pub repos: Arc<Repos>,

    /// Wikis, one Oxygen repository per repository (hub spec §6.11).

    pub wikis: Arc<Repos>,

    pub server: Arc<oxy_server::Server>,

    pub hub_key: keys::HubKey,

    pub cookie_key: auth::CookieKey,

    /// Seals AI assistants' tokens at rest (hub spec §13.4).

    pub assistant_key: assistants::SecretKey,

    /// Seals Flow secrets at rest (hub spec §13.2, #107).

    pub secret_key: assistants::SecretKey,

    /// The OIDC issuer for jobs (hub spec §13.5), made on first use.

    pub oidc: tokio::sync::OnceCell<oidc::Issuer>,

    /// Which assistants are answering which proposals, and recent calls

    /// per repository (hub spec §13.4).

    pub assistant_gate: std::sync::Mutex<assistants::Gate>,

    /// Failed sign-ins and the like, per account and address (oxygenhub#221).

    pub throttle: std::sync::Mutex<throttle::Throttle>,

    /// Events for live pages and workers, here and in the other processes

    /// sharing the database (docs/roles.md, #273).

    pub bus: bus::Bus,

    /// Who is resolving which conflict together (hub spec §6.10).

    pub rooms: Arc<live::Rooms>,

    /// Outgoing mail: hub.toml's `[email]`, overridden by what the admin

    /// page saved (#101).

    pub mail: std::sync::RwLock<config::Email>,

    /// The image swarm's tracker (oxygenhub#144).

    pub swarm: swarm::Tracker,

    pub started: std::time::Instant,

    /// The transparency log (#238).

    pub tlog: Arc<tlog::Log>,

    /// What this process does: web, worker or both (docs/roles.md, #272).

    pub roles: config::Roles,

    /// The background tasks started here and the leader locks held (#272).

    pub leader: leader::Leader,

}

impl App {

    /// The rootfs images this hub keeps for runners (oxygenflow#39).

    pub fn images(&self) -> images::Store {

        images::Store::new(&self.config.data)

    }

    /// The OIDC issuer, its key loaded or made the first time it is asked for.

    pub async fn oidc(&self) -> anyhow::Result<&oidc::Issuer> {

        let keys = self.config.keys_dir();

        self.oidc.get_or_try_init(|| async move { tokio::task::spawn_blocking(move || oidc::Issuer::load_or_create(&keys)).await? }).await

    }

}

pub type Shared = Arc<App>;

/// Open the data directory and build the application state. `init` must

/// have run once; `serve` and the tests call this.

pub async fn build(config: Config) -> anyhow::Result<Shared> {

    let roles = config.roles()?;

    std::fs::create_dir_all(config.repos_dir())?;

    let db = db::open(&config).await?;

    let repos = Arc::new(Repos::new(config.repos_dir()));

    let wikis = Arc::new(Repos::new(config.data.join("wikis")));

    let hub_key = keys::ensure_identity(&config.keys_dir(), &config.site.name, config.base_url())?;

    let cookie_key = auth::CookieKey::load_or_create(&config.keys_dir())?;

    let assistant_key = assistants::SecretKey::load_or_create(&config.keys_dir())?;

    let secret_key = assistants::SecretKey::load_or_create_named(&config.keys_dir(), "secrets.key")?;

    let backend = protocol::HubBackend::new(

        repos.clone(),

        wikis.clone(),

        db.clone(),

        config.auth.push_tokens.clone(),

        tokio::runtime::Handle::current(),

    )

    .with_acceptor(hub_key.identity.clone(), keys::load_secret(&config.keys_dir())?.1)

    .with_scan_budget(push_protection::Budget {

        bytes: config.limits.scan_limit.0,

        file: config.limits.scan_file_limit.0,

        time: std::time::Duration::from_millis(config.limits.scan_time_ms),

    });

    let slot = backend.app_slot();

    let server = oxy_server::Server::new(backend);

    let mail = mail::load(&db, &config.email).await;

    // The web role appends to the log; a worker alone only reads it (#272).

    let (tlog_dir, origin) = (config.data.join("tlog"), tlog::origin_of(config.base_url()));

    let tlog = Arc::new(if roles.web { tlog::Log::open(&tlog_dir, &origin)? } else { tlog::Log::open_reader(&tlog_dir, &origin)? });

    tlog.configure(&config.tlog)?;

    let app = Arc::new(App {

        mail: std::sync::RwLock::new(mail),

        config,

        db,

        repos,

        wikis,

        server,

        hub_key,

        cookie_key,

        assistant_key,

        secret_key,

        oidc: tokio::sync::OnceCell::new(),

        assistant_gate: std::sync::Mutex::new(assistants::Gate::default()),

        throttle: std::sync::Mutex::new(throttle::Throttle::default()),

        bus: bus::Bus::default(),

        rooms: Arc::new(live::Rooms::default()),

        swarm: Default::default(),

        started: std::time::Instant::now(),

        tlog,

        roles,

        leader: leader::Leader::default(),

    });

    *slot.lock().unwrap() = Some(Arc::downgrade(&app));

    bus::start(&app).await?;

    storage::load(&app).await?;

    coverage::load(&app).await?;

    junit::load(&app).await?;

    reports::load(&app).await?;

    start_background(&app);

    setup::ensure(&app).await?;

    Ok(app)

}

/// Start this process's background tasks (docs/roles.md, #272). The

/// worker role's are singletons across every process sharing the database,

/// each run under its leader lock; the transparency log's checkpointer goes

/// with the web role, which appends to the log, and is a singleton too.

fn start_background(app: &Shared) {

    if app.roles.web {

        // The checkpointer signs the log this process appends to (#238).

        leader::run(app, "tlog.checkpointer", tlog::checkpointer);

    }

    if !app.roles.worker {

        return;

    }

    leader::run(app, "hooks.worker", hooks::worker);

    leader::run(app, "flow.reaper", flow::reaper);

    leader::run(app, "flow.scheduler", flow::scheduler);

    leader::run(app, "flow.placement", flow::placement);

    leader::run(app, "maintenance.nightly", maintenance::nightly);

    leader::run(app, "backups.scheduler", backups::scheduler);

    // The reverse pin index for repositories pushed before it existed (#148):

    // once, by whichever worker leads it first (idempotent if run again).

    leader::run(app, "diffusion.backfill", |app| async move { diffusion::backfill(&app).await });

    // Mend the pin index and the bumps it drives (oxygenhub#210).

    leader::run(app, "diffusion.reconciler", diffusion::reconciler);

    // What each image is for, read or learned once (oxygenhub#173): safe

    // in every worker, so it takes no lock.

    app.leader.note("images.learn_archs");

    tokio::spawn({

        let app = app.clone();

        async move {

            match images::learn_archs(&app).await {

                Ok(n) if n > 0 => tracing::info!(images = n, "read the architecture of kept images"),

                Ok(_) => {}

                Err(e) => tracing::warn!(error = %e, "learning image architectures"),

            }

        }

    });

}

/// The router this process serves: everything for the web role; for a

/// worker alone, `/health` and `/metrics`, so it can be watched.

pub fn router(app: Shared) -> axum::Router {

    if !app.roles.web {

        return worker_router(app);

    }

    full_router(app)

}

/// What a worker-only process answers: its health and its metrics.

fn worker_router(app: Shared) -> axum::Router {

    use axum::routing::get;

    let hsts = headers::hsts(app.config.base_url(), app.config.server.hsts_max_age);

    axum::Router::new()

        .route("/health", get(web::home::health))

        .route("/metrics", get(metrics::metrics))

        .with_state(app)

        .layer(axum::middleware::from_fn_with_state(hsts, headers::security_headers))

}

/// The full router: pages, API, protocol and static assets.

fn full_router(app: Shared) -> axum::Router {

    let https = app_https(&app);

    let hsts = headers::hsts(app.config.base_url(), app.config.server.hsts_max_age);

    web::router(app.clone())

        .layer(axum::middleware::from_fn_with_state(app, agent_requests))

        .layer(axum::middleware::from_fn(private_pages))

        .layer(axum::middleware::from_fn(slow_requests))

        .layer(axum::middleware::from_fn_with_state(https, https_only))

        .layer(axum::middleware::from_fn_with_state(hsts, headers::security_headers))

}

/// The site's https address, when it has one: what [`https_only`] redirects to.

fn app_https(app: &Shared) -> Option<String> {

    let base = app.config.base_url();

    base.starts_with("https://").then(|| base.trim_end_matches('/').to_string())

}

/// The scheme a proxy in front of the hub says the visitor used

/// (`X-Forwarded-Proto`, else Cloudflare's `CF-Visitor`); `None` for a

/// request made to the hub directly (runners and tools on its host).

fn forwarded_scheme(headers: &axum::http::HeaderMap) -> Option<String> {

    if let Some(p) = headers.get("x-forwarded-proto").and_then(|v| v.to_str().ok()) {

        return p.split(',').next().map(|s| s.trim().to_ascii_lowercase());

    }

    let visitor = headers.get("cf-visitor").and_then(|v| v.to_str().ok())?;

    serde_json::from_str::<serde_json::Value>(visitor).ok()?["scheme"].as_str().map(str::to_ascii_lowercase)

}

/// HTTPS only (oxygenhub#218): on a site whose address is https, a request

/// a proxy reports as plain http is sent to the https address (GET and

/// HEAD) or refused (anything else, so a form or a password is never taken

/// over http). Requests made to the hub directly carry no such report and

/// are untouched.

async fn https_only(

    axum::extract::State(https): axum::extract::State<Option<String>>,

    req: axum::extract::Request,

    next: axum::middleware::Next,

) -> axum::response::Response {

    use axum::response::IntoResponse;

    let Some(base) = https else { return next.run(req).await };

    if forwarded_scheme(req.headers()).as_deref() != Some("http") {

        return next.run(req).await;

    }

    let to = format!("{base}{}", req.uri().path_and_query().map(|p| p.as_str()).unwrap_or("/"));

    if matches!(*req.method(), axum::http::Method::GET | axum::http::Method::HEAD) {

        return axum::response::Redirect::permanent(&to).into_response();

    }

    (axum::http::StatusCode::BAD_REQUEST, format!("This site takes requests over HTTPS only: send it to {to}\n")).into_response()

}

/// A request made with a token its owner marked as an agent's runs with

/// that token's name in [`notify::AGENT`], so what the agent does is told

/// to its owner (oxygenhub#183).

async fn agent_requests(

    axum::extract::State(app): axum::extract::State<Shared>,

    req: axum::extract::Request,

    next: axum::middleware::Next,

) -> axum::response::Response {

    let token = req

        .headers()

        .get(axum::http::header::AUTHORIZATION)

        .and_then(|v| v.to_str().ok())

        .and_then(|v| v.strip_prefix("Bearer "))

        .map(|t| t.trim().to_string());

    let agent = match token {

        // A keyless agent's grant token (#253) is always an agent's.

        Some(t) if t.starts_with(keyless::GRANT_TOKEN_PREFIX) => db::accounts::agent_grant_by_token(&app.db, &auth::token_hash(&t))

            .await

            .ok()

            .flatten()

            .filter(|g| g.live(db::now_ms()))

            .map(|g| g.agent),

        Some(t) => db::accounts::token_by_hash(&app.db, &auth::token_hash(&t)).await.ok().flatten().filter(|t| t.agent).map(|t| t.name),

        None => None,

    };

    notify::AGENT.scope(agent, next.run(req)).await

}

/// Requests slower than this are logged, with their path (#145).

const SLOW: std::time::Duration = std::time::Duration::from_secs(2);

/// Log a slow request's method, path and time, so a slow page can be

/// found afterwards. Never the query, which may hold a token.

async fn slow_requests(req: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response {

    let (method, path) = (req.method().clone(), req.uri().path().to_string());

    let start = std::time::Instant::now();

    let res = next.run(req).await;

    let took = start.elapsed();

    metrics::observe(method.as_str(), &path, res.status().as_u16(), took.as_secs_f64());

    if took >= SLOW {

        tracing::warn!("slow request: {method} {path} took {:.1} s ({})", took.as_secs_f64(), res.status().as_u16());

    }

    res

}

/// A page for a signed-in viewer is theirs alone: it may list private

/// repositories, and the dashboard's sky carries their activity (#157). An

/// HTML response to a request with a session cookie, which has not said

/// how to cache itself, says `private, no-store`, so no shared cache (a

/// proxy or a CDN in front of the hub) ever serves it to anyone else.

async fn private_pages(req: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response {

    use axum::http::header;

    let signed_in = req

        .headers()

        .get_all(header::COOKIE)

        .iter()

        .filter_map(|v| v.to_str().ok())

        .any(|c| c.split(';').any(|kv| kv.trim_start().starts_with("oxyhub=")));

    let mut res = next.run(req).await;

    let html = res.headers().get(header::CONTENT_TYPE).and_then(|v| v.to_str().ok()).is_some_and(|t| t.starts_with("text/html"));

    if signed_in && html && !res.headers().contains_key(header::CACHE_CONTROL) {

        res.headers_mut().insert(header::CACHE_CONTROL, header::HeaderValue::from_static("private, no-store"));

    }

    res

}

/// Serve on the configured addresses until the process is told to stop.

pub async fn serve(app: Shared) -> anyhow::Result<()> {

    let router = router(app.clone());

    let http = app.config.bind_addr()?;

    let https = app.config.tls_bind_addr()?;

    if http.is_none() && https.is_none() && app.roles.web {

        anyhow::bail!("nothing to bind: set server.bind or server.tls_bind");

    }

    tracing::info!("roles: {}", app.roles.names().join(", "));

    // Installed first: from here a SIGTERM drains rather than kills.

    let signal = shutdown::on_signal()?;

    let mut tasks = tokio::task::JoinSet::new();

    if let Some(addr) = http {

        let listener = match shutdown::systemd_listener() {

            Some(l) => {

                tracing::info!("http on the socket systemd holds ({})", l.local_addr().map(|a| a.to_string()).unwrap_or_default());

                tokio::net::TcpListener::from_std(l)?

            }

            None => {

                tracing::info!("http on {addr}");

                tokio::net::TcpListener::bind(addr).await?

            }

        };

        let r = router.clone();

        tasks.spawn(async move {

            axum::serve(listener, r.into_make_service_with_connect_info::<std::net::SocketAddr>())

                .with_graceful_shutdown(shutdown::stopping())

                .await

                .map_err(anyhow::Error::from)

        });

    }

    if let Some(addr) = https {

        let _ = rustls::crypto::ring::default_provider().install_default();

        let tls = axum_server::tls_rustls::RustlsConfig::from_pem_file(app.config.tls_cert(), app.config.tls_key())

            .await

            .map_err(|e| anyhow::anyhow!("loading TLS files: {e}"))?;

        tracing::info!("https on {addr}");

        let r = router.clone();

        let handle = axum_server::Handle::new();

        let draining = handle.clone();

        tokio::spawn(async move {

            shutdown::stopping().await;

            draining.graceful_shutdown(Some(shutdown::DRAIN));

        });

        tasks.spawn(async move {

            axum_server::bind_rustls(addr, tls)

                .handle(handle)

                .serve(r.into_make_service_with_connect_info::<std::net::SocketAddr>())

                .await

                .map_err(anyhow::Error::from)

        });

    }

    tokio::select! {

        // A worker with no address to serve on runs until it is stopped.

        r = tasks.join_next(), if !tasks.is_empty() => { r.transpose()?.transpose()?; Ok(()) }

        _ = signal => {

            // The servers stop taking connections and finish what they

            // have; whatever is still running after DRAIN is cut off.

            let drained = tokio::time::timeout(shutdown::DRAIN, async { while tasks.join_next().await.is_some() {} }).await;

            if drained.is_err() {

                tracing::warn!("requests still running after {} s: stopping anyway", shutdown::DRAIN.as_secs());

            }

            tracing::info!("stopped");

            Ok(())

        }

    }

}