Live L4: the bell, the inbox and the dashboard (closes #206); /metrics, first slice (refs #212) 8a48dd830178

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

}

    /// The bearer token `/metrics` asks for (oxygenhub#212); empty, the

    /// metrics are off.

    pub metrics_token: String,

            metrics_token: String::new(),

pub mod metrics;

    /// What the cards and the feed listen on (docs/live.md L4): the

    /// viewer's topic and their repositories'.

    live: String,

    let live = std::iter::once(shell.live_user.clone())

        .chain(mine.iter().take(15).map(|r| crate::events::topic::repo(r.id)))

        .filter(|t| !t.is_empty())

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

        .join(" ");

    Ok(Page(Dashboard { shell, mine, reviews, authored, assigned, feed, backdrop, live }).into_response())

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

    /// The signed-in viewer's live topic, `user:<id>` (docs/live.md L4),

    /// or empty.

    pub live_user: String,

            live_user: viewer.account().map(|a| format!("user:{}", a.id)).unwrap_or_default(),

            live_user: String::new(),

/// Live L4 (docs/live.md, oxygenhub#206): a signed-in page's bell listens

/// on the viewer's own topic, the inbox list on it too, and the dashboard's

/// cards and feed on it and the viewer's repositories'; a signed-out page

/// listens on nothing.

#[tokio::test(flavor = "multi_thread")]

async fn the_bell_inbox_and_dashboard_listen_on_the_viewer_s_topics() {

    let h = hub().await;

    let app = &h.app;

    let andy = register(app, "andy", "").await;

    let id = db::accounts::account_by_name(&app.db, "andy").await.unwrap().unwrap().id;

    let topic = format!("user:{id}");

    let page = get(app, "/explore", Some(&andy)).await.body;

    assert!(page.contains(&format!("data-live=\"{topic}\" data-fragment=\"bell\"")), "the bell on every page");

    let inbox = get(app, "/notifications", Some(&andy)).await.body;

    assert!(inbox.contains(&format!("data-live=\"{topic}\" data-fragment=\"inbox\"")), "the inbox list");

    let dash = get(app, "/", Some(&andy)).await.body;

    assert!(dash.contains("data-fragment=\"cards\"") && dash.contains("data-fragment=\"feed\""), "{dash}");

    assert!(dash.contains(&format!("data-live=\"{topic}")), "the dashboard listens on the viewer");

    let anon = get(app, "/explore", None).await.body;

    assert!(!anon.contains("data-fragment=\"bell\""), "nothing to hear signed out");

}

/// `/metrics` (oxygenhub#212): off until a token is set, then served only

/// to its bearer, in the Prometheus text format, with Flow's jobs counted.

#[tokio::test(flavor = "multi_thread")]

async fn metrics_are_private_and_count_flow_jobs() {

    let h = hub().await;

    let (st, _) = call(&h.app, "GET", "/metrics", None, None).await;

    assert_eq!(st, StatusCode::NOT_FOUND, "off by default");

    let text = oxygenhub::metrics::render(&h.app).await;

    for want in [

        "# TYPE oxyhub_uptime_seconds gauge",

        "oxyhub_flow_jobs{state=\"queued\"} 0",

        "oxyhub_bucket_requests_total{kind=\"list\"}",

        "oxyhub_tlog_entries",

    ] {

        assert!(text.contains(want), "missing {want}:\n{text}");

    }

    create(&h.app, "andy", "demo").await;

    let work = tempfile::tempdir().unwrap();

    let c = client(&format!("{}/andy/demo", h.base), work.path());

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(work.path(), "test.sh", "echo ok\n");

    c.snapshot("test").unwrap();

    c.commit("Add a pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    assert_eq!(runs_of(&h.app, "andy", "demo").await.len(), 1);

    let text = oxygenhub::metrics::render(&h.app).await;

    assert!(!text.contains("oxyhub_flow_jobs{state=\"queued\"} 0"), "the run's jobs are queued:\n{text}");

}

- **L4 (done, #206):** the top bar's notification count listens on the

  viewer's `user:<id>` topic on every page (the Shell carries it), the

  inbox's list on it too, and the dashboard's cards and activity feed on

  it and the topics of up to 15 of the viewer's repositories (a stream

  takes 16). Nothing new on the server: the client refetches the page and

  swaps each marked part, so every route already serves them.

### 13.5.5 Metrics (oxygenhub#212, first slice 2026-10-18)

`GET /metrics` serves the hub's numbers in the Prometheus text format, only to a bearer of `[server] metrics_token` in hub.toml. While the token is unset the route answers 404. It reports:

- the build (`oxyhub_build_info`), uptime and resident memory;

- the state cache's states, estimated weight, budget and summaries;

- requests to bucket stores by kind (with pack listings sent and kept apart), and the time spent in them;

- OxygenFlow jobs queued, waiting for approval and running, and runners online and not;

- the transparency log's entries.

Per-route request counts and latency histograms, pushes and fetches, backups, Diffusion, mail and the database pool are the next slices; the runner's own metrics too.

    <a class="bell" href="/notifications" title="Notifications"{% if !shell.live_user.is_empty() %} data-live="{{ shell.live_user }}" data-fragment="bell"{% endif %}>✉{% if shell.unread > 0 %}<span class="count">{{ shell.unread }}</span>{% endif %}</a>

  <div data-live="{{ live }}" data-fragment="cards">

  </div>

    <section class="card-box" data-live="{{ live }}" data-fragment="feed">

<div data-live="{{ shell.live_user }}" data-fragment="inbox">

</div>