konnosDashboard
connection-seconds.ts130 lines · main
1import { env } from '../env.js';
2import { log } from '../lib/logger.js';
3
4/**
5 * Scrape briven_realtime_connection_seconds_total{project=...} from the
6 * realtime service's /metrics endpoint and compute the delta vs the
7 * previous scrape, per project. The cumulative counter resets when the
8 * realtime process restarts; we detect that case (current < previous)
9 * and treat the new value as the delta from zero so a restart costs at
10 * most one scrape window of data.
11 *
12 * State lives in this module — `Map<projectId, lastTotalSeconds>` — so
13 * a fresh api boot loses one hour of connection-seconds data for every
14 * project that was active across the restart. That's the same blast
15 * radius as the realtime side resetting and acceptable at Phase 3
16 * scale; the durable fix is a `scraper_state` row but it's not worth
17 * the table churn until customer count justifies it.
18 *
19 * Authenticates with BRIVEN_RUNTIME_SHARED_SECRET — the realtime
20 * service refuses every other path without it (apps/realtime/src/
21 * index.ts), but /metrics is open today for Prometheus scrape access.
22 * Leaving the auth header in anyway so a future lockdown of /metrics
23 * doesn't break this scraper silently.
24 */
25
26const lastTotalByProject = new Map<string, number>();
27
28/** Visible for tests — pin the parser shape, not the I/O. */
29export function parseConnectionSecondsMetrics(promBody: string): Map<string, number> {
30 const out = new Map<string, number>();
31 const lines = promBody.split('\n');
32 for (const raw of lines) {
33 const line = raw.trim();
34 if (!line || line.startsWith('#')) continue;
35 if (!line.startsWith('briven_realtime_connection_seconds_total')) continue;
36
37 // Format: briven_realtime_connection_seconds_total{project="p_..."} 1234.56
38 const lbraceIdx = line.indexOf('{');
39 const rbraceIdx = line.indexOf('}');
40 if (lbraceIdx === -1 || rbraceIdx === -1 || rbraceIdx < lbraceIdx) continue;
41
42 const labelBody = line.slice(lbraceIdx + 1, rbraceIdx);
43 const projectMatch = labelBody.match(/project="([^"]+)"/);
44 if (!projectMatch) continue;
45 const projectId = projectMatch[1] ?? '';
46 if (!projectId) continue;
47
48 const valueStr = line.slice(rbraceIdx + 1).trim().split(/\s+/)[0] ?? '';
49 const value = Number.parseFloat(valueStr);
50 if (!Number.isFinite(value) || value < 0) continue;
51
52 out.set(projectId, value);
53 }
54 return out;
55}
56
57/**
58 * Fetch /metrics from realtime and parse the connection_seconds gauge.
59 * Returns an empty map on any failure — callers treat it as "no data
60 * this tick" and skip writing rows rather than zeroing valid history.
61 */
62export async function scrapeConnectionSecondsTotals(): Promise<Map<string, number>> {
63 if (!env.BRIVEN_REALTIME_URL) return new Map();
64 try {
65 const res = await fetch(`${env.BRIVEN_REALTIME_URL}/metrics`, {
66 headers: env.BRIVEN_RUNTIME_SHARED_SECRET
67 ? { authorization: `Bearer ${env.BRIVEN_RUNTIME_SHARED_SECRET}` }
68 : {},
69 });
70 if (!res.ok) {
71 log.warn('connection_seconds_scrape_http', {
72 status: res.status,
73 url: env.BRIVEN_REALTIME_URL,
74 });
75 return new Map();
76 }
77 const body = await res.text();
78 return parseConnectionSecondsMetrics(body);
79 } catch (err) {
80 log.warn('connection_seconds_scrape_failed', {
81 url: env.BRIVEN_REALTIME_URL,
82 message: err instanceof Error ? err.message : String(err),
83 });
84 return new Map();
85 }
86}
87
88/**
89 * Compute the delta (in seconds) since the last call, per project. Mutates
90 * the in-memory baseline so successive calls see only the increment.
91 * Visible for tests so the restart-detection branch is pinned.
92 */
93export function diffConnectionSeconds(currentTotals: Map<string, number>): Map<string, number> {
94 const deltas = new Map<string, number>();
95 for (const [projectId, current] of currentTotals) {
96 const previous = lastTotalByProject.get(projectId);
97 let delta: number;
98 if (previous === undefined) {
99 // First scrape after boot — treat the current value as the
100 // baseline. Writing it would over-count seconds that happened
101 // before the api booted; skip this round and start measuring
102 // from the next scrape.
103 delta = 0;
104 } else if (current < previous) {
105 // Counter went backwards — realtime restarted. The "current"
106 // value is the seconds accumulated since that restart; treat
107 // it as the delta.
108 delta = current;
109 } else {
110 delta = current - previous;
111 }
112 lastTotalByProject.set(projectId, current);
113 if (delta > 0) deltas.set(projectId, delta);
114 }
115 return deltas;
116}
117
118/** Test-only — reset the in-memory baseline. */
119export function _resetConnectionSecondsBaseline(): void {
120 lastTotalByProject.clear();
121}
122
123/**
124 * One-shot helper — scrape + diff. The aggregator calls this each hour
125 * and writes the deltas to usage_events for the just-ended hour.
126 */
127export async function collectConnectionSecondsDeltas(): Promise<Map<string, number>> {
128 const totals = await scrapeConnectionSecondsTotals();
129 return diffConnectionSeconds(totals);
130}