Repository Public
1// The ONE alert dispatcher.2//3// How events become alerts — and why exactly once4// ------------------------------------------------5// The money code is not touched: it already records everything important in three append-only tables (audit_events — the hash6// chain —, deriv_risk_events, research_log). The dispatcher TAILS those tables by id. One pass is one database transaction that7// 1. takes an advisory lock (one dispatcher at a time, even with two API processes),8// 2. reads the rows after the stored cursor,9// 3. inserts the alerts they produce,10// 4. advances the cursor,11// and commits all of it together. A crash anywhere before the commit leaves the cursor where it was and no alert written: the12// next pass redoes the same rows. A crash after the commit has both the alerts and the new cursor. There is no window in which13// an event is consumed without its alert, or an alert written without the event being consumed — no in-memory queue, no hook14// that could fire inside a money transaction that later rolls back, nothing lost while the process is down (the rows wait).15//16// audit_events numbers its rows under a lock held until commit, so its ids become visible in order and a plain cursor is exact.17// The two bigserial tables can commit out of order (a transaction that took id 5 may commit after id 6 is visible): for them the18// pass re-reads an overlap window behind the cursor and records the ids it handled in `alert_seen`, in the same transaction, so19// a late row is picked up once and an already handled row is skipped. A unique index on (source, source_id, kind) is the last20// line: the same event cannot produce the same alert twice whatever happens above.21//22// Delivery (push, webhook) is a second step on rows marked `pending`. A row is claimed (pending → sent) BEFORE the network call:23// a crash in between loses that notification rather than repeating it — the alert itself is in the app either way. At most once.24import { alertDef, safeParams, type AlertParams, type AlertSeverity, type AlertSubject } from '@spw/core';25import { sql } from '../db.ts';26import { diskFree, lastBackup } from '../monitor/health.ts';27import { conditionDrafts, type ConditionDeps } from './conditions.ts';28import { deriveAlerts, pinDraft, pinOf, type AlertEvent, type Draft } from './rules.ts';29import { vapidKeys, vapidSubject, webhookSecret } from './secrets.ts';30import { alertSettings, getState, migrateAlerts, putState, type AlertRow, type AlertSettings } from './store.ts';31import { sendWebhook } from './webhook.ts';32import { sendPush, type PushOutcome } from './webpush.ts';3334type Q = typeof sql;35export interface DispatcherDeps extends ConditionDeps {36 fetch: typeof fetch;37 origin: string;38 /** audit rows younger than this wait one pass (a route writes its audit row a moment before the rows the rule reads) */39 settleMs: number;40 /** tests: webhook to a local http receiver */41 allowHttpWebhook: boolean;42 /** tests: fault injection inside the tailing transaction (throwing here is a crash before the commit) */43 fault?: (stage: 'before_cursor', source: string) => void;44}45interface Source { name: 'audit' | 'risk' | 'research'; overlap: number; max(q: Q): Promise<number>; read(q: Q, after: number, limit: number): Promise<AlertEvent[]> }46const OVERLAP = 200, BATCH = 500;47const SOURCES: Source[] = [48 { name: 'audit', overlap: 0,49 max: async (q) => Number((await q<{ m: string | null }[]>`select max(seq) as m from audit_events`)[0]?.m ?? 0),50 read: async (q, after, limit) => (await q<{ seq: string; ts: string; type: string; actor: string; data: Record<string, unknown> }[]>`select seq, ts, type, actor, data from audit_events where seq > ${after} order by seq limit ${limit}`)51 .map((r) => ({ source: 'audit' as const, id: Number(r.seq), ts: new Date(r.ts), type: r.type, actor: r.actor, data: r.data ?? {} })) },52 { name: 'risk', overlap: OVERLAP,53 max: async (q) => Number((await q<{ m: string | null }[]>`select max(id) as m from deriv_risk_events`)[0]?.m ?? 0),54 read: async (q, after, limit) => (await q<{ id: string; account_id: string; ts: Date; kind: string; code: string | null; message: string; data: Record<string, unknown> }[]>`55 select id, account_id, ts, kind, code, message, data from deriv_risk_events where id > ${after} order by id limit ${limit}`)56 .map((r) => ({ source: 'risk' as const, id: Number(r.id), ts: r.ts, type: r.kind, code: r.code, message: r.message, accountId: r.account_id, data: r.data ?? {} })) },57 { name: 'research', overlap: OVERLAP,58 max: async (q) => Number((await q<{ m: string | null }[]>`select max(id) as m from research_log`)[0]?.m ?? 0),59 read: async (q, after, limit) => (await q<{ id: string; agent_id: string | null; ts: Date; kind: string; title: string; body: string }[]>`60 select id, agent_id, ts, kind, title, body from research_log where id > ${after} order by id limit ${limit}`)61 .map((r) => ({ source: 'research' as const, id: Number(r.id), ts: r.ts, type: r.kind, title: r.title, message: r.body, agentId: r.agent_id, data: {} })) },62];6364/** local HH:MM of an instant in a time zone */65export function localMinutes(now: number, tz: string): number {66 const parts = new Intl.DateTimeFormat('en-GB', { timeZone: tz, hour: '2-digit', minute: '2-digit', hourCycle: 'h23' }).formatToParts(new Date(now));67 return Number(parts.find((p) => p.type === 'hour')!.value) * 60 + Number(parts.find((p) => p.type === 'minute')!.value);68}69export function inQuietHours(now: number, q: AlertSettings['quiet']): boolean {70 if (!q.enabled) return false;71 const m = (s: string) => Number(s.slice(0, 2)) * 60 + Number(s.slice(3, 5)), a = m(q.start), b = m(q.end), t = localMinutes(now, q.tz);72 if (a === b) return false;73 return a < b ? t >= a && t < b : t >= a || t < b; // a window that crosses midnight74}7576export class AlertDispatcher {77 readonly deps: DispatcherDeps;78 private readonly listeners = new Set<() => void>();79 private lastConditions = 0;80 constructor(deps: DispatcherDeps) { this.deps = deps; }8182 subscribe(fn: () => void): () => void { this.listeners.add(fn); return () => { this.listeners.delete(fn); }; }83 changed(): void { for (const l of [...this.listeners]) { try { l(); } catch { /* a listener must not break the others */ } } }8485 /** Insert one alert (or bump a digest). Returns true when a row was created or bumped. Honors the owner's switch for the kind. */86 async raise(draft: Draft, o: { q?: Q; ts?: Date; source?: string; sourceId?: string; settings?: AlertSettings } = {}): Promise<boolean> {87 const def = alertDef(draft.kind), q = o.q ?? sql, s = o.settings ?? await alertSettings();88 if (!def) return false;89 if (!def.fixed && s.rules[draft.kind] === false) return false;90 const severity: AlertSeverity = draft.severity ?? def.severity, ts = o.ts ?? new Date(this.deps.now());91 const params: AlertParams = Object.fromEntries(Object.entries(draft.params).filter(([, v]) => v !== undefined));92 const subject: AlertSubject | null = draft.subject ?? null;93 const ins = await q`insert into alerts (ts, kind, severity, params, subject, dedupe_key, source, source_id)94 values (${ts}, ${draft.kind}, ${severity}, ${q.json(params as never)}, ${subject ? q.json(subject as never) : null}, ${draft.dedupe ?? null}, ${o.source ?? 'rule'}, ${o.sourceId ?? null})95 on conflict do nothing returning id`;96 if (ins.length) return true;97 if (!draft.bump || !draft.dedupe) return false;98 // digest: the alert of this window already exists — count this event on it (and show it as unread again)99 const up = await q`update alerts set count = count + 1, params = (params || ${q.json(params as never)}) || jsonb_build_object('count', count + 1), ts = ${ts}, read_at = null100 where dedupe_key = ${draft.dedupe} returning id`;101 return up.length > 0;102 }103104 /** events → alerts, one transaction (see the header). Returns the number of alerts created or bumped. */105 async tail(): Promise<number> {106 const settings = await alertSettings(), settleBefore = this.deps.now() - this.deps.settleMs;107 const n = await sql.begin(async (tx) => {108 const q = tx as unknown as Q;109 await tx`select pg_advisory_xact_lock(7707002)`;110 let made = 0;111 for (const src of SOURCES) {112 const [c] = await tx<{ cursor: string; floor: string }[]>`select cursor, floor from alert_cursors where source = ${src.name} for update`;113 if (!c) { // first run: start from the present, history is not replayed114 const m = await src.max(q);115 await tx`insert into alert_cursors (source, cursor, floor) values (${src.name}, ${m}, ${m})`;116 continue;117 }118 const cursor = Number(c.cursor), floor = Number(c.floor);119 let rows = await src.read(q, Math.max(floor, cursor - src.overlap), BATCH);120 if (src.name === 'audit') { const k = rows.findIndex((r) => r.ts.getTime() > settleBefore); if (k >= 0) rows = rows.slice(0, k); }121 if (!rows.length) continue;122 let fresh = rows;123 if (src.overlap > 0) {124 const claimed = await tx<{ id: string }[]>`insert into alert_seen ${tx(rows.map((r) => ({ source: src.name, id: r.id })))} on conflict do nothing returning id`;125 const set = new Set(claimed.map((r) => Number(r.id)));126 fresh = rows.filter((r) => set.has(r.id));127 }128 for (const ev of fresh) {129 // a rule that fails on one event must not stop the others: its work is rolled back to a savepoint and the event is skipped130 try {131 await tx.savepoint(async (sp) => {132 const sq = sp as unknown as Q;133 for (const d of await deriveAlerts(ev, { q: sq, settings })) if (await this.raise(d, { q: sq, ts: ev.ts, source: ev.source, sourceId: String(ev.id), settings })) made += 1;134 });135 } catch (e) { console.error(`[alerts] rule failed on ${ev.source} #${ev.id} (${ev.type}): ${(e as Error).message.slice(0, 200)}`); }136 }137 this.deps.fault?.('before_cursor', src.name);138 const next = Math.max(cursor, rows[rows.length - 1]!.id);139 await tx`update alert_cursors set cursor = ${next} where source = ${src.name}`;140 if (src.overlap > 0) await tx`delete from alert_seen where source = ${src.name} and id <= ${next - src.overlap}`;141 }142 return made;143 });144 if (n > 0) this.changed();145 return n;146 }147148 /** polled conditions (stuck transfers, expiring mandates, signer, connectors, audit chain, backup, disk, loops) */149 async conditions(): Promise<number> {150 const settings = await alertSettings();151 let n = 0;152 for (const d of await conditionDrafts(this.deps, settings)) if (await this.raise(d, { settings })) n += 1;153 if (n > 0) this.changed();154 return n;155 }156157 /** a refusal seen at the HTTP layer that is recorded nowhere else (pinned code changed, asset table mismatch, signer refusals) */158 async signal(code: string, message: string): Promise<void> {159 const ts = new Date(this.deps.now()), what = pinOf(code, message);160 let d: Draft | null = null;161 if (what) d = pinDraft(what, message, ts);162 else if (code === 'asset_table_mismatch') d = { kind: 'deriv.asset_table_mismatch', params: {}, dedupe: `asset_table:${Math.floor(ts.getTime() / 3_600_000)}` };163 else if (code.startsWith('signer_') && code !== 'signer_unreachable') d = { kind: 'security.signer_refused', params: { reason: message.slice(0, 160) }, dedupe: `signer_refused:${Math.floor(ts.getTime() / 900_000)}`, bump: true };164 if (d && (await this.raise(d, { ts }))) this.changed();165 }166167 /** pending alerts → push and webhook, with quiet hours and the hourly cap */168 async deliver(): Promise<{ sent: number; quiet: number; capped: number }> {169 const now = this.deps.now(), s = await alertSettings(), out = { sent: 0, quiet: 0, capped: 0 };170 const hour = Math.floor(now / 3_600_000);171 let cap = await getState<{ bucket: number; sent: number; suppressed: number }>('cap', { bucket: hour, sent: 0, suppressed: 0 });172 if (cap.bucket !== hour) {173 // the hour is over: ONE summary for everything the cap held back174 if (cap.suppressed > 0) await this.raise({ kind: 'system.alerts_capped', params: { count: cap.suppressed }, dedupe: `cap:${cap.bucket}` }, { settings: s });175 cap = { bucket: hour, sent: 0, suppressed: 0 };176 await putState('cap', cap);177 }178 const rows = await sql<AlertRow[]>`select * from alerts where delivery = 'pending' order by id limit 100`;179 if (!rows.length) return out;180 const subs = await sql<{ id: string; endpoint: string; p256dh: string; auth: string; locale: string }[]>`select id, endpoint, p256dh, auth, locale from push_subscriptions`;181 const hook = s.webhook.enabled && s.webhook.url ? webhookSecret() : null;182 for (const a of rows) {183 const critical = a.severity === 'critical';184 const decision = subs.length === 0 && !hook ? 'inapp'185 : Date.now() - a.created_at.getTime() > 3_600_000 ? 'stale'186 : !critical && inQuietHours(now, s.quiet) ? 'quiet'187 : !critical && a.kind !== 'system.alerts_capped' && cap.sent >= s.capPerHour ? 'capped' : 'sent';188 // claim first: a crash after this line never repeats a notification189 const claim = await sql`update alerts set delivery = ${decision} where id = ${a.id} and delivery = 'pending' returning id`;190 if (!claim.length) continue;191 if (decision === 'quiet') { out.quiet += 1; continue; }192 if (decision === 'capped') { out.capped += 1; cap.suppressed += 1; await putState('cap', cap); continue; }193 if (decision !== 'sent') continue;194 cap.sent += 1; await putState('cap', cap); out.sent += 1;195 const channels: Record<string, unknown> = {};196 if (subs.length) {197 const res = await Promise.all(subs.map((sub) => this.pushOne(sub, a, s.showDetails)));198 channels.push = { sent: res.filter((r) => r.ok).length, failed: res.filter((r) => !r.ok && !r.gone).length, removed: res.filter((r) => !r.ok && r.gone).length };199 }200 if (hook) {201 const r = await sendWebhook(s.webhook.url, hook, a, s.webhook.details, { now, fetch: this.deps.fetch, allowHttp: this.deps.allowHttpWebhook });202 channels.webhook = { ok: r.ok, status: r.status };203 }204 await sql`update alerts set channels = ${sql.json(channels as never)} where id = ${a.id}`;205 }206 return out;207 }208209 /** one encrypted push to one device; a subscription the push service no longer knows (404/410) is removed */210 async pushOne(sub: { id: string; endpoint: string; p256dh: string; auth: string; locale: string }, a: Pick<AlertRow, 'id' | 'kind' | 'severity' | 'params' | 'subject' | 'count' | 'ts'>, details: boolean): Promise<PushOutcome & { gone?: boolean }> {211 const p = { ...a.params, count: a.params.count ?? a.count };212 // minimal payload: a kind and parameters; without "show details" only names, networks and counts leave the server213 const payload = JSON.stringify({ v: 1, id: Number(a.id), kind: a.kind, sev: a.severity, ts: a.ts.getTime(), l: sub.locale === 'en' ? 'en' : 'fr', d: details, p: details ? p : safeParams(p), s: a.subject });214 const r = await sendPush(sub, Buffer.from(payload), vapidKeys(), { subject: vapidSubject(this.deps.origin), urgency: a.severity === 'critical' ? 'high' : 'normal', ttl: a.severity === 'critical' ? 86_400 : 14_400, fetch: this.deps.fetch });215 if (r.ok) await sql`update push_subscriptions set last_ok_at = now(), last_error = null, failures = 0 where id = ${sub.id}`;216 else if (r.gone) await sql`delete from push_subscriptions where id = ${sub.id}`;217 else await sql`update push_subscriptions set failures = failures + 1, last_error = ${r.error} where id = ${sub.id}`;218 return r;219 }220221 async step(): Promise<void> {222 await this.tail();223 const now = Date.now();224 if (now - this.lastConditions >= 30_000) { this.lastConditions = now; await this.conditions(); }225 await this.deliver();226 }227}228229let shared: AlertDispatcher | null = null;230/** the process-wide dispatcher on the real signer client, RPC health cache and worker heartbeats */231export async function alerts(): Promise<AlertDispatcher> {232 if (shared) return shared;233 const [{ config }, { signer }, worker] = await Promise.all([import('../config.ts'), import('../signer-client.ts'), import('../worker.ts')]);234 shared = new AlertDispatcher({235 now: Date.now, fetch, origin: config.origin, settleMs: 1500, allowHttpWebhook: false,236 signerStatus: () => signer.status().catch(() => null), rpcHealth: () => [...worker.healthCache.values()], loops: () => [...worker.loopStats.values()],237 backup: () => lastBackup(), disk: () => diskFree(),238 });239 return shared;240}241/** tests: replace the process-wide dispatcher */242export const setAlerts = (d: AlertDispatcher | null): void => { shared = d; };243244export const ALERTS_STEP_MS = 5_000;245/** Started with the other workers. Says once that the API (re)started. */246export async function startAlerts(): Promise<void> {247 await migrateAlerts();248 const d = await alerts(), { loop } = await import('../worker.ts');249 vapidKeys(); // generate the key pair at first start, not at the first subscription250 const boot = Date.now();251 await d.raise({ kind: 'system.api_restarted', params: {}, dedupe: `boot:${boot}` }).catch((e: Error) => console.error(`[alerts] ${e.message}`));252 loop('alerts', () => d.step(), ALERTS_STEP_MS);253}254