//! OxygenHub: a forge for Oxygen repositories (`docs/spec.md`).
//!
//! One process: an axum router serving pages, the JSON API and the wire
//! protocol; SQLite 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 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 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 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>,
    pub events: events::Events,
    /// 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>,
}

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> {
    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);
    let slot = backend.app_slot();
    let server = oxy_server::Server::new(backend);
    let mail = mail::load(&db, &config.email).await;
    let tlog = Arc::new(tlog::Log::open(&config.data.join("tlog"), &tlog::origin_of(config.base_url()))?);
    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()),
        events: events::Events::default(),
        rooms: Arc::new(live::Rooms::default()),
        swarm: Default::default(),
        started: std::time::Instant::now(),
        tlog,
    });
    *slot.lock().unwrap() = Some(Arc::downgrade(&app));
    storage::load(&app).await?;
    coverage::load(&app).await?;
    junit::load(&app).await?;
    reports::load(&app).await?;
    tokio::spawn(hooks::worker(app.clone()));
    tokio::spawn(tlog::checkpointer(app.clone()));
    tokio::spawn(flow::reaper(app.clone()));
    tokio::spawn(flow::scheduler(app.clone()));
    tokio::spawn(flow::placement(app.clone()));
    tokio::spawn(maintenance::nightly(app.clone()));
    tokio::spawn(backups::scheduler(app.clone()));
    // The reverse pin index for repositories pushed before it existed (#148).
    tokio::spawn({
        let app = app.clone();
        async move { diffusion::backfill(&app).await }
    });
    // Mend the pin index and the bumps it drives (oxygenhub#210).
    tokio::spawn(diffusion::reconciler(app.clone()));
    // What each image is for, read or learned once (oxygenhub#173).
    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"),
            }
        }
    });
    setup::ensure(&app).await?;
    Ok(app)
}

/// The full router: pages, API, protocol and static assets.
pub fn 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() {
        anyhow::bail!("nothing to bind: set server.bind or server.tls_bind");
    }
    // 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! {
        r = tasks.join_next() => { 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(())
        }
    }
}