//! `/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.

//!

//! The process, the state cache, repositories kept in buckets, OxygenFlow's

//! jobs, runs and runners, the transparency log, the database pool, pushes,

//! backups, mail, and HTTP requests by route class and status with latency

//! histograms. The rest the issue lists comes next.

use crate::{db, Shared};

use axum::extract::State;

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

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

use std::collections::HashMap;

use std::fmt::Write;

use std::sync::atomic::{AtomicU64, Ordering};

use std::sync::Mutex;

/// Messages the hub sent, and failed to send, since it started.

pub static MAIL_SENT: AtomicU64 = AtomicU64::new(0);

pub static MAIL_FAILED: AtomicU64 = AtomicU64::new(0);

/// Latency buckets, in seconds.

const BUCKETS: [f64; 12] = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0];

/// At most this many route classes are kept; more go to `other`.

const MAX_ROUTES: usize = 300;

#[derive(Default)]

struct Hist {

    /// Requests by status class (2xx … 5xx).

    by_status: [u64; 4],

    buckets: [u64; BUCKETS.len()],

    sum: f64,

    count: u64,

}

static HTTP: Mutex<Option<HashMap<(String, String), Hist>>> = Mutex::new(None);

/// Bytes the protocol received and sent, by endpoint, counted before the

/// answer is compressed: what pushes and fetches carry.

static PROTOCOL: Mutex<std::collections::BTreeMap<&'static str, (u64, u64)>> = Mutex::new(std::collections::BTreeMap::new());

/// Count a protocol call's request and response bodies.

pub fn protocol_bytes(endpoint: &'static str, received: u64, sent: u64) {

    let mut g = PROTOCOL.lock().unwrap_or_else(|p| p.into_inner());

    let e = g.entry(endpoint).or_default();

    e.0 += received;

    e.1 += sent;

}

/// A request's route class: a few dozen at most, so a label never holds a

/// user's or repository's name.

pub fn route_class(path: &str) -> String {

    let segs: Vec<&str> = path.trim_start_matches('/').split('/').filter(|s| !s.is_empty()).collect();

    let tab = |s: &str| s.chars().all(|c| c.is_ascii_lowercase() || c == '-' || c == '.') && s.len() <= 24;

    match segs.as_slice() {

        [] => "page:home".into(),

        ["static", ..] => "static".into(),

        ["api", "v1", area, ..] if tab(area) => format!("api:{area}"),

        ["api", ..] => "api".into(),

        ["log", ..] => "tlog".into(),

        [".well-known", ..] => "well-known".into(),

        ["live", ..] => "live".into(),

        ["avatar", ..] => "avatar".into(),

        [_, _, "oxy", "v1", endpoint, ..] if tab(endpoint) => format!("protocol:{endpoint}"),

        [_, _, "oxy", ..] => "protocol".into(),

        [_, _, t, ..] if tab(t) => format!("page:{t}"),

        [_, _] => "page:repository".into(),

        [one] if tab(one) => format!("page:{one}"),

        _ => "page:other".into(),

    }

}

/// Count a request (from the router's outermost timing layer).

pub fn observe(method: &str, path: &str, status: u16, seconds: f64) {

    let mut g = HTTP.lock().unwrap_or_else(|p| p.into_inner());

    let map = g.get_or_insert_with(HashMap::new);

    let mut route = route_class(path);

    if !map.contains_key(&(route.clone(), method.to_string())) && map.len() >= MAX_ROUTES {

        route = "other".into();

    }

    let h = map.entry((route, method.to_string())).or_default();

    let class = (status / 100).clamp(2, 5) as usize - 2;

    h.by_status[class] += 1;

    for (i, b) in BUCKETS.iter().enumerate() {

        if seconds <= *b {

            h.buckets[i] += 1;

        }

    }

    h.sum += seconds;

    h.count += 1;

}

fn http_families(out: &mut String) {

    let g = HTTP.lock().unwrap_or_else(|p| p.into_inner());

    let Some(map) = g.as_ref() else { return };

    let mut keys: Vec<&(String, String)> = map.keys().collect();

    keys.sort();

    let _ = writeln!(out, "# HELP oxyhub_http_requests_total HTTP requests by route class, method and status class.");

    let _ = writeln!(out, "# TYPE oxyhub_http_requests_total counter");

    for k in &keys {

        let h = &map[*k];

        for (i, n) in h.by_status.iter().enumerate().filter(|(_, n)| **n > 0) {

            let _ = writeln!(

                out,

                "oxyhub_http_requests_total{{route=\"{}\",method=\"{}\",status=\"{}xx\"}} {n}",

                label(&k.0),

                label(&k.1),

                i + 2

            );

        }

    }

    let _ = writeln!(out, "# HELP oxyhub_http_request_duration_seconds How long requests took, by route class and method.");

    let _ = writeln!(out, "# TYPE oxyhub_http_request_duration_seconds histogram");

    for k in &keys {

        let h = &map[*k];

        let l = format!("route=\"{}\",method=\"{}\"", label(&k.0), label(&k.1));

        for (i, b) in BUCKETS.iter().enumerate() {

            let _ = writeln!(out, "oxyhub_http_request_duration_seconds_bucket{{{l},le=\"{b}\"}} {}", h.buckets[i]);

        }

        let _ = writeln!(out, "oxyhub_http_request_duration_seconds_bucket{{{l},le=\"+Inf\"}} {}", h.count);

        let _ = writeln!(out, "oxyhub_http_request_duration_seconds_sum{{{l}}} {}", h.sum);

        let _ = writeln!(out, "oxyhub_http_request_duration_seconds_count{{{l}}} {}", h.count);

    }

}

/// 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),

        ],

    );

    let now = db::now_ms();

    family(

        &mut out,

        "oxyhub_flow_runner_last_seen_seconds",

        "Seconds since each registered runner last called the hub.",

        "gauge",

        &runners

            .iter()

            .filter_map(|r| r.last_seen_ms.map(|t| (format!("runner=\"{}\"", label(&r.name)), (now - t).max(0) as f64 / 1000.0)))

            .collect::<Vec<_>>(),

    );

    let since = now - 86_400_000;

    let runs: Vec<(String, i64)> = sqlx::query_as("SELECT status, COUNT(*) FROM flow_runs WHERE finished_ms > $1 GROUP BY status")

        .bind(since)

        .fetch_all(&app.db)

        .await

        .unwrap_or_default();

    family(

        &mut out,

        "oxyhub_flow_runs_last_day",

        "OxygenFlow runs finished in the last 24 hours, by result.",

        "gauge",

        &["passed", "failed", "error"]

            .iter()

            .map(|r| (format!("result=\"{r}\""), runs.iter().find(|(s, _)| s == r).map(|(_, n)| *n).unwrap_or(0) as f64))

            .collect::<Vec<_>>(),

    );

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

    let protocol: Vec<(&'static str, (u64, u64))> =

        PROTOCOL.lock().unwrap_or_else(|p| p.into_inner()).iter().map(|(k, v)| (*k, *v)).collect();

    family(

        &mut out,

        "oxyhub_protocol_bytes_total",

        "Bytes the Oxygen protocol received and sent since the hub started, by endpoint, before compression: push carries pushes, fetch and objects carry fetches.",

        "counter",

        &protocol

            .iter()

            .flat_map(|(e, (rx, tx))| {

                [(format!("endpoint=\"{e}\",direction=\"received\""), *rx as f64), (format!("endpoint=\"{e}\",direction=\"sent\""), *tx as f64)]

            })

            .collect::<Vec<_>>(),

    );

    family(

        &mut out,

        "oxyhub_mail_total",

        "Messages the hub sent and failed to send since it started.",

        "counter",

        &[

            ("result=\"sent\"".into(), MAIL_SENT.load(Ordering::Relaxed) as f64),

            ("result=\"failed\"".into(), MAIL_FAILED.load(Ordering::Relaxed) as f64),

        ],

    );

    family(

        &mut out,

        "oxyhub_db_connections",

        "The database pool's connections, by whether they are idle.",

        "gauge",

        &[

            ("idle=\"true\"".into(), app.db.num_idle() as f64),

            ("idle=\"false\"".into(), (app.db.size() as f64 - app.db.num_idle() as f64).max(0.0)),

        ],

    );

    let (ok, refused): (Option<i64>, Option<i64>) =

        sqlx::query_as("SELECT SUM(CASE WHEN ok THEN 1 ELSE 0 END), SUM(CASE WHEN ok THEN 0 ELSE 1 END) FROM pushes WHERE at_ms > $1")

            .bind(since)

            .fetch_one(&app.db)

            .await

            .unwrap_or((None, None));

    family(

        &mut out,

        "oxyhub_pushes_last_day",

        "Reference updates pushed in the last 24 hours, by whether they were accepted.",

        "gauge",

        &[("result=\"accepted\"".into(), ok.unwrap_or(0) as f64), ("result=\"refused\"".into(), refused.unwrap_or(0) as f64)],

    );

    let backups: Vec<(String, String, i64, Option<i64>)> = sqlx::query_as(

        "SELECT owner, status, started_ms, finished_ms FROM backup_runs WHERE id IN (SELECT MAX(id) FROM backup_runs GROUP BY owner) ORDER BY owner",

    )

    .fetch_all(&app.db)

    .await

    .unwrap_or_default();

    family(

        &mut out,

        "oxyhub_backup_last_run_age_seconds",

        "Seconds since each owner's latest backup run started.",

        "gauge",

        &backups.iter().map(|(o, _, s, _)| (format!("owner=\"{}\"", label(o)), (now - s) as f64 / 1000.0)).collect::<Vec<_>>(),

    );

    family(

        &mut out,

        "oxyhub_backup_last_run_ok",

        "Whether each owner's latest finished backup run succeeded (1), was partial or failed (0); running ones are left out.",

        "gauge",

        &backups

            .iter()

            .filter(|(_, st, _, f)| f.is_some() && st != "running")

            .map(|(o, st, _, _)| (format!("owner=\"{}\"", label(o)), if st == "ok" { 1.0 } else { 0.0 }))

            .collect::<Vec<_>>(),

    );

    flow_timings(app, &mut out, now).await;

    if let Ok((waiting, dead)) = db::queue::counts(&app.db).await {

        family(

            &mut out,

            "oxyhub_queue_items",

            "Items in the durable work queue (#276): waiting (due, backing off or claimed) and dead (failed every attempt).",

            "gauge",

            &[("state=\"waiting\"".into(), waiting as f64), ("state=\"dead\"".into(), dead as f64)],

        );

    }

    http_families(&mut out);

    out

}

/// A finished job's runner, result, note, and when it was ready, started

/// and finished.

type JobTimes = (String, String, String, Option<i64>, Option<i64>, Option<i64>);

/// Bucket bounds for job times, in seconds.

const JOB_SECONDS: [f64; 11] = [1.0, 5.0, 15.0, 30.0, 60.0, 120.0, 300.0, 600.0, 1200.0, 1800.0, 3600.0];

/// Flow's job times over the last day (oxygenflow#35), by runner: how long

/// jobs waited from ready to started (queue time) and ran (run time), and

/// how many finished, by result, and were reused (#265).

async fn flow_timings(app: &Shared, out: &mut String, now: i64) {

    let rows: Vec<JobTimes> = sqlx::query_as(

        "SELECT runner, status, error, COALESCE(ready_ms, queued_ms), started_ms, finished_ms FROM flow_jobs \

         WHERE finished_ms >= $1 AND status IN ('passed', 'failed', 'error', 'cancelled')",

    )

    .bind(now - 86_400_000)

    .fetch_all(&app.db)

    .await

    .unwrap_or_default();

    let mut queue: std::collections::BTreeMap<String, Vec<f64>> = Default::default();

    let mut run: std::collections::BTreeMap<String, Vec<f64>> = Default::default();

    let mut finished: std::collections::BTreeMap<(String, String), f64> = Default::default();

    let mut reused = 0.0;

    for (runner, status, error, ready, started, finished_ms) in &rows {

        if error.starts_with("reused from run #") {

            reused += 1.0;

            continue;

        }

        let r = if runner.is_empty() { "none".to_string() } else { runner.clone() };

        *finished.entry((r.clone(), status.clone())).or_default() += 1.0;

        if let (Some(ready), Some(started)) = (ready, started) {

            queue.entry(r.clone()).or_default().push((started - ready).max(0) as f64 / 1000.0);

        }

        if let (Some(started), Some(f)) = (started, finished_ms) {

            run.entry(r).or_default().push((f - started).max(0) as f64 / 1000.0);

        }

    }

    histogram(

        out,

        "oxyhub_flow_job_queue_seconds",

        "How long finished jobs of the last day waited from ready to started, by runner.",

        &queue,

    );

    histogram(out, "oxyhub_flow_job_run_seconds", "How long finished jobs of the last day ran, by runner.", &run);

    family(

        out,

        "oxyhub_flow_jobs_finished_last_day",

        "Jobs that finished in the last day, by runner and result (none: never placed).",

        "gauge",

        &finished.iter().map(|((r, st), n)| (format!("runner=\"{}\",status=\"{}\"", label(r), label(st)), *n)).collect::<Vec<_>>(),

    );

    family(

        out,

        "oxyhub_flow_jobs_reused_last_day",

        "Jobs of the last day that reused a pass on the same state instead of running (#265).",

        "gauge",

        &[(String::new(), reused)],

    );

}

/// A histogram family from samples (seconds) per runner.

fn histogram(out: &mut String, name: &str, help: &str, by_runner: &std::collections::BTreeMap<String, Vec<f64>>) {

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

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

    for (runner, samples) in by_runner {

        let l = format!("runner=\"{}\"", label(runner));

        for b in JOB_SECONDS {

            let n = samples.iter().filter(|s| **s <= b).count();

            let _ = writeln!(out, "{name}_bucket{{{l},le=\"{b}\"}} {n}");

        }

        let _ = writeln!(out, "{name}_bucket{{{l},le=\"+Inf\"}} {}", samples.len());

        let _ = writeln!(out, "{name}_sum{{{l}}} {}", samples.iter().sum::<f64>());

        let _ = writeln!(out, "{name}_count{{{l}}} {}", samples.len());

    }

}

/// `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();

    }