Repository Public
1import { nodes } from "./data";2import type { HistoryPoint, HistoryWindow, LiveCluster, LiveNode, LiveSnapshot } from "./live-types";3import type { NodeRecord } from "./schemas";45/**6 * Server-side bridge to the cluster's private metrics collector.7 * The browser never talks to it: this module fetches, strips everything but utilisation figures,8 * re-keys by public record id, and caches, so the collector sees at most one request per cache period9 * whatever the public traffic.10 */11const URL_ = process.env.LIVE_SOURCE_URL ?? "";12const KEY = process.env.LIVE_SOURCE_KEY ?? "";13export const liveConfigured = () => Boolean(URL_ && KEY);1415const SNAPSHOT_TTL = 5_000;16const HISTORY_TTL = 60_000;17const STALE_AFTER = 120_000;18const INTERVAL_S = 20;1920async function source<T>(path: string, timeoutMs = 4000): Promise<T> {21 const res = await fetch(`${URL_}${path}`, { headers: { Authorization: `Bearer ${KEY}` }, cache: "no-store", signal: AbortSignal.timeout(timeoutMs) });22 if (!res.ok) throw new Error(`live source ${res.status}`);23 return (await res.json()) as T;24}2526const num = (v: unknown, fallback = 0): number => (typeof v === "number" && Number.isFinite(v) ? v : fallback);27const pct = (v: unknown): number => Math.min(100, Math.max(0, Math.round(num(v) * 10) / 10));28const round1 = (v: number) => Math.round(v * 10) / 10;2930type RawMetric = Record<string, unknown>;3132/** Pure: one collector record -> one public record. Exported for unit tests. */33export function sanitizeNode(id: string, raw: RawMetric | undefined): LiveNode {34 const online = raw?.status === "online";35 if (!raw || !online) return { id, online: false, cpu: 0, memPct: 0, memUsedGB: 0, memTotalGB: 0, gpu: null, load1: 0, uptimeS: 0, thermal: null, netInKBs: 0, netOutKBs: 0, battery: null };36 const memTotal = num(raw.memTotalMB);37 const memUsed = num(raw.memUsedMB);38 const gpu = num(raw.gpu, -1);39 const battery = num(raw.battery, -1);40 const load = Array.isArray(raw.load) ? num(raw.load[0]) : 0;41 return {42 id,43 online: true,44 cpu: pct(raw.cpu),45 memPct: memTotal > 0 ? pct((memUsed / memTotal) * 100) : 0,46 memUsedGB: round1(memUsed / 1024),47 memTotalGB: Math.round(memTotal / 1024),48 gpu: gpu >= 0 ? pct(gpu) : null,49 load1: Math.round(load * 100) / 100,50 uptimeS: Math.max(0, Math.round(num(raw.uptime))),51 thermal: typeof raw.thermal === "string" && /^[a-z]{1,16}$/i.test(raw.thermal) ? raw.thermal.toLowerCase() : null,52 netInKBs: round1(Math.max(0, num(raw.netInKBs))),53 netOutKBs: round1(Math.max(0, num(raw.netOutKBs))),54 battery: battery >= 0 ? Math.round(battery) : null,55 };56}5758/** Pure: cluster figures over the public records only. CPU and GPU are weighted by each node's core count. */59export function summarize(live: LiveNode[], records: Pick<NodeRecord, "id" | "cpuCores" | "gpuCores">[]): LiveCluster {60 const rec = new Map(records.map((r) => [r.id, r]));61 const on = live.filter((n) => n.online);62 let cpuW = 0, cpuSum = 0, gpuW = 0, gpuSum = 0;63 for (const n of on) {64 const c = rec.get(n.id)?.cpuCores ?? 1;65 cpuW += c;66 cpuSum += n.cpu * c;67 if (n.gpu !== null) {68 const g = rec.get(n.id)?.gpuCores ?? 1;69 gpuW += g;70 gpuSum += n.gpu * g;71 }72 }73 const memUsed = on.reduce((a, n) => a + n.memUsedGB, 0);74 const memTotal = on.reduce((a, n) => a + n.memTotalGB, 0);75 return {76 online: on.length,77 total: live.length,78 cpu: cpuW ? round1(cpuSum / cpuW) : 0,79 memUsedGB: Math.round(memUsed),80 memTotalGB: Math.round(memTotal),81 memPct: memTotal ? round1((memUsed / memTotal) * 100) : 0,82 gpu: gpuW ? round1(gpuSum / gpuW) : null,83 netInKBs: round1(on.reduce((a, n) => a + n.netInKBs, 0)),84 netOutKBs: round1(on.reduce((a, n) => a + n.netOutKBs, 0)),85 coresOnline: cpuW,86 };87}8889const EMPTY: LiveSnapshot = { available: false, ts: 0, intervalS: INTERVAL_S, cluster: null, nodes: [] };90let snap: { at: number; value: LiveSnapshot } | null = null;91let snapPending: Promise<LiveSnapshot> | null = null;9293export async function getSnapshot(): Promise<LiveSnapshot> {94 if (!liveConfigured()) return EMPTY;95 if (snap && Date.now() - snap.at < SNAPSHOT_TTL) return snap.value;96 if (snapPending) return snapPending;97 snapPending = (async () => {98 try {99 const raw = await source<{ ts?: number; metrics?: Record<string, RawMetric> }>("/api/cluster");100 const metrics = raw.metrics ?? {};101 const live = nodes.map((n) => sanitizeNode(n.id, metrics[n.publicName]));102 const ts = Math.round(num(raw.ts) * 1000);103 const value: LiveSnapshot = Date.now() - ts > STALE_AFTER ? EMPTY : { available: true, ts, intervalS: INTERVAL_S, cluster: summarize(live, nodes), nodes: live };104 snap = { at: Date.now(), value };105 return value;106 } catch {107 // Serve the last good reading while it is still recent; otherwise say plainly that telemetry is unavailable.108 if (snap && snap.value.available && Date.now() - snap.value.ts < STALE_AFTER) return snap.value;109 snap = { at: Date.now(), value: EMPTY };110 return EMPTY;111 } finally {112 snapPending = null;113 }114 })();115 return snapPending;116}117118type RawPoint = { ts: number; cpu: number; mem: number; gpu: number };119const histCache = new Map<HistoryWindow, { at: number; value: Map<string, HistoryPoint[]> }>();120const histPending = new Map<HistoryWindow, Promise<Map<string, HistoryPoint[]>>>();121122/** History of every public node for one window (one cached batch feeds the node charts, the cluster chart and the sparklines). */123async function allHistories(window: HistoryWindow): Promise<Map<string, HistoryPoint[]>> {124 const hit = histCache.get(window);125 if (hit && Date.now() - hit.at < HISTORY_TTL) return hit.value;126 const pending = histPending.get(window);127 if (pending) return pending;128 const p = (async () => {129 try {130 const entries = await Promise.all(131 nodes.map(async (n) => {132 const r = await source<{ points?: RawPoint[] }>(`/api/history?window=${window}&node=${encodeURIComponent(n.publicName)}`, 8000).catch(() => ({ points: [] as RawPoint[] }));133 const pts = (r.points ?? []).map((q) => ({ t: num(q.ts) * 1000, cpu: pct(q.cpu), mem: pct(q.mem), gpu: num(q.gpu, -1) >= 0 ? pct(q.gpu) : null }));134 return [n.id, pts] as const;135 }),136 );137 const value = new Map(entries);138 histCache.set(window, { at: Date.now(), value });139 return value;140 } finally {141 histPending.delete(window);142 }143 })();144 histPending.set(window, p);145 return p;146}147148/** Pure: core-weighted cluster series from per-node series aligned on the collector's time buckets. Exported for unit tests. */149export function clusterSeries(byNode: Map<string, HistoryPoint[]>, records: Pick<NodeRecord, "id" | "cpuCores" | "unifiedMemoryGB" | "gpuCores">[]): HistoryPoint[] {150 const acc = new Map<number, { cw: number; cs: number; mw: number; ms: number; gw: number; gs: number }>();151 for (const r of records) {152 const c = r.cpuCores ?? 1, m = r.unifiedMemoryGB ?? 1, g = r.gpuCores ?? 1;153 for (const p of byNode.get(r.id) ?? []) {154 const a = acc.get(p.t) ?? { cw: 0, cs: 0, mw: 0, ms: 0, gw: 0, gs: 0 };155 a.cw += c; a.cs += p.cpu * c; a.mw += m; a.ms += p.mem * m;156 if (p.gpu !== null) { a.gw += g; a.gs += p.gpu * g; }157 acc.set(p.t, a);158 }159 }160 return [...acc.entries()].sort((x, y) => x[0] - y[0]).map(([t, a]) => ({ t, cpu: round1(a.cs / a.cw), mem: round1(a.ms / a.mw), gpu: a.gw ? round1(a.gs / a.gw) : null }));161}162163export async function getHistory(target: string, window: HistoryWindow): Promise<HistoryPoint[] | null> {164 if (!liveConfigured()) return null;165 const all = await allHistories(window);166 if (target === "cluster") return clusterSeries(all, nodes);167 return all.get(target) ?? null;168}169170/** Last hour of CPU for every node, for the sparklines. */171export async function getSpark(): Promise<Record<string, number[]> | null> {172 if (!liveConfigured()) return null;173 const all = await allHistories("1h");174 return Object.fromEntries([...all.entries()].map(([id, pts]) => [id, pts.map((p) => p.cpu)]));175}176