//! `/metrics` (oxygenhub#212): the hub's numbers in the Prometheus text

//! format, for a monitoring stack (oxygenhub-gcp's Prometheus and Grafana).

//! Private: served only to a bearer of `[server] metrics_token`, and not

//! at all while that is unset.

//!

//! First slice: the process, the state cache, repositories kept in buckets,

//! OxygenFlow's jobs and runners, and the transparency log. Per-route

//! request histograms and the rest the issue lists come next.

use crate::{db, Shared};

use axum::extract::State;

use axum::http::{header, HeaderMap, StatusCode};

use axum::response::{IntoResponse, Response};

use std::fmt::Write;

/// One metric family: its help, type and samples (label set, value).

fn family(out: &mut String, name: &str, help: &str, kind: &str, samples: &[(String, f64)]) {

    let _ = writeln!(out, "# HELP {name} {help}");

    let _ = writeln!(out, "# TYPE {name} {kind}");

    for (labels, v) in samples {

        if labels.is_empty() {

            let _ = writeln!(out, "{name} {v}");

        } else {

            let _ = writeln!(out, "{name}{{{labels}}} {v}");

        }

    }

}

/// A label value, escaped as the text format asks.

fn label(v: &str) -> String {

    v.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n")

}

/// The process's resident memory, from /proc (Linux), in bytes.

fn resident_bytes() -> Option<f64> {

    let statm = std::fs::read_to_string("/proc/self/statm").ok()?;

    let pages: f64 = statm.split_whitespace().nth(1)?.parse().ok()?;

    Some(pages * 4096.0)

}

/// The metrics as text, for whoever is let in.

pub async fn render(app: &Shared) -> String {

    let mut out = String::new();

    family(

        &mut out,

        "oxyhub_build_info",

        "The hub's version and build.",

        "gauge",

        &[(format!("version=\"{}\",build=\"{}\"", env!("CARGO_PKG_VERSION"), label(env!("OXY_BUILD_ID"))), 1.0)],

    );

    family(

        &mut out,

        "oxyhub_uptime_seconds",

        "Seconds since the hub started.",

        "gauge",

        &[(String::new(), app.started.elapsed().as_secs_f64())],

    );

    if let Some(rss) = resident_bytes() {

        family(&mut out, "oxyhub_resident_memory_bytes", "The process's resident memory.", "gauge", &[(String::new(), rss)]);

    }

    let (states, weight, summaries) = app.repos.cache().stats();

    family(&mut out, "oxyhub_state_cache_states", "States the state cache holds.", "gauge", &[(String::new(), states as f64)]);

    family(

        &mut out,

        "oxyhub_state_cache_bytes",

        "What the state cache's states are estimated to weigh.",

        "gauge",

        &[(String::new(), weight as f64)],

    );

    family(

        &mut out,

        "oxyhub_state_cache_budget_bytes",

        "What the state cache may hold.",

        "gauge",

        &[(String::new(), crate::repos::cache::budget() as f64)],

    );

    family(&mut out, "oxyhub_state_cache_summaries", "Summaries the state cache holds.", "gauge", &[(String::new(), summaries as f64)]);

    let strata = oxy_strata::stats();

    family(

        &mut out,

        "oxyhub_bucket_clients_opened_total",

        "Bucket store clients opened.",

        "counter",

        &[(String::new(), strata.opens as f64)],

    );

    family(

        &mut out,

        "oxyhub_bucket_requests_total",

        "Requests to bucket stores, by kind.",

        "counter",

        &strata.ops().map(|(k, n)| (format!("kind=\"{k}\""), n as f64)).collect::<Vec<_>>(),

    );

    family(

        &mut out,

        "oxyhub_bucket_request_seconds_total",

        "Time spent in requests to bucket stores.",

        "counter",

        &[(String::new(), strata.millis as f64 / 1000.0)],

    );

    let jobs: Vec<(String, i64)> =

        sqlx::query_as("SELECT status, COUNT(*) FROM flow_jobs WHERE status IN ('queued', 'running') GROUP BY status")

            .fetch_all(&app.db)

            .await

            .unwrap_or_default();

    let count = |s: &str| jobs.iter().find(|(k, _)| k == s).map(|(_, n)| *n).unwrap_or(0) as f64;

    let waiting: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM flow_jobs WHERE status = 'queued' AND approval = 'waiting'")

        .fetch_one(&app.db)

        .await

        .unwrap_or(0);

    family(

        &mut out,

        "oxyhub_flow_jobs",

        "OxygenFlow jobs now, by state (waiting: for a person's approval).",

        "gauge",

        &[

            ("state=\"queued\"".into(), count("queued") - waiting as f64),

            ("state=\"waiting\"".into(), waiting as f64),

            ("state=\"running\"".into(), count("running")),

        ],

    );

    let runners = db::runners::list(&app.db).await.unwrap_or_default();

    family(

        &mut out,

        "oxyhub_flow_runners",

        "Registered OxygenFlow runners, by whether they reported in the last minutes.",

        "gauge",

        &[

            ("online=\"true\"".into(), runners.iter().filter(|r| r.online()).count() as f64),

            ("online=\"false\"".into(), runners.iter().filter(|r| !r.online()).count() as f64),

        ],

    );

    family(&mut out, "oxyhub_tlog_entries", "Entries in the transparency log.", "gauge", &[(String::new(), app.tlog.size() as f64)]);

    out

}

/// `GET /metrics`.

pub async fn metrics(State(app): State<Shared>, headers: HeaderMap) -> Response {

    let want = &app.config.server.metrics_token;

    if want.is_empty() {

        return (StatusCode::NOT_FOUND, "metrics are off: set [server] metrics_token in hub.toml\n").into_response();

    }

    let given = oxy_server::bearer_token(&headers).unwrap_or("");

    if !crate::auth::mac_equal(given, want) {

        return (StatusCode::UNAUTHORIZED, [(header::WWW_AUTHENTICATE, "Bearer")], "a bearer token is needed\n").into_response();

    }

    ([(header::CONTENT_TYPE, "text/plain; version=0.0.4; charset=utf-8"), (header::CACHE_CONTROL, "no-store")], render(&app).await)

        .into_response()

}