Repository Public
1// Alerts API: the in-app list, the live stream (SSE), the owner's switches, push subscriptions per device and the optional webhook.2// Every route is under /api/ and therefore behind the owner session and the CSRF check of server.ts — nothing here is public.3import { randomUUID } from 'node:crypto';4import type { FastifyInstance } from 'fastify';5import { z } from 'zod';6import { ALERT_DEFS, alertDef, type AlertGroup } from '@spw/core';7import { UserError } from '../config.ts';8import { audit, sql } from '../db.ts';9import { alerts, type AlertDispatcher } from './dispatcher.ts';10import { rotateWebhookSecret, vapidKeys, webhookSecret } from './secrets.ts';11import { AlertSettingsSchema, alertSettings, migrateAlerts, saveAlertSettings, type AlertRow, type AlertSettings } from './store.ts';12import { sendWebhook, webhookBody, webhookUrlOk } from './webhook.ts';13import { endpointAllowed } from './webpush.ts';1415const MAX_STREAMS = 8, MAX_SUBSCRIPTIONS = 12, STREAM_MAX_MS = 10 * 60_000;16/** error codes of other routes that are alerts in themselves and are recorded nowhere else */17const SIGNAL = new Set(['router_changed', 'contract_changed', 'asset_table_mismatch']);1819const view = (a: AlertRow) => ({ id: Number(a.id), ts: a.ts.getTime(), kind: a.kind, severity: a.severity, group: (alertDef(a.kind)?.group ?? 'system') as AlertGroup,20 params: { ...a.params, count: a.params.count ?? a.count }, subject: a.subject, count: a.count, readAt: a.read_at?.getTime() ?? null, delivery: a.delivery, channels: a.channels });21const unread = async () => {22 const [r] = await sql<{ unread: number; critical: number; last: string | null }[]>`select (count(*) filter (where read_at is null))::int as unread,23 (count(*) filter (where read_at is null and severity = 'critical'))::int as critical, max(id) as last from alerts`;24 return { unread: r?.unread ?? 0, critical: r?.critical ?? 0, lastId: Number(r?.last ?? 0) };25};26const subView = (s: { id: string; endpoint: string; label: string; locale: string; ua: string; created_at: Date; last_ok_at: Date | null; last_error: string | null; failures: number }) =>27 ({ id: s.id, label: s.label, locale: s.locale, service: new URL(s.endpoint).hostname, endpointTail: s.endpoint.slice(-12), ua: s.ua, createdAt: s.created_at.getTime(), lastOkAt: s.last_ok_at?.getTime() ?? null, lastError: s.last_error, failures: s.failures });2829const SettingsPatch = z.strictObject({30 rules: z.record(z.string(), z.boolean()).optional(), quiet: AlertSettingsSchema.shape.quiet.partial().optional(), showDetails: z.boolean().optional(), capPerHour: AlertSettingsSchema.shape.capPerHour.optional(),31 thresholds: AlertSettingsSchema.shape.thresholds.partial().optional(), webhook: AlertSettingsSchema.shape.webhook.partial().optional(),32});33const Subscription = z.strictObject({34 endpoint: z.string().url().max(1000), keys: z.strictObject({ p256dh: z.string().regex(/^[A-Za-z0-9_-]{80,100}=*$/), auth: z.string().regex(/^[A-Za-z0-9_-]{16,30}=*$/) }),35 label: z.string().trim().max(40).default(''), locale: z.enum(['fr', 'en']).default('fr'),36});3738export async function alertRoutes(app: FastifyInstance, o: { dispatcher?: AlertDispatcher } = {}): Promise<void> {39 await migrateAlerts();40 const d = o.dispatcher ?? await alerts();41 const P = '/api/v1/alerts', strict = { config: { rateLimit: { max: 6, timeWindow: '1 minute' } } };4243 // one hook, at the HTTP layer: a refusal such as "the router bytecode changed" is thrown by the money code and stored nowhere;44 // seeing it leave as an error answer is enough to alert, without adding a call inside that code45 app.addHook('onError', async (_req, _reply, err) => {46 if (!(err instanceof UserError)) return;47 if (SIGNAL.has(err.code) || (err.code.startsWith('signer_') && err.code !== 'signer_unreachable')) void d.signal(err.code, err.message).catch(() => {});48 });4950 app.get(P, async (req) => {51 const q = z.object({ limit: z.coerce.number().int().min(1).max(200).default(50), before: z.coerce.number().int().positive().optional(), severity: z.enum(['info', 'warning', 'critical']).optional(),52 group: z.enum(['transfers', 'agents', 'derivatives', 'security', 'system']).optional(), unread: z.enum(['1']).optional() }).parse(req.query);53 const kinds = q.group ? ALERT_DEFS.filter((x) => x.group === q.group).map((x) => x.kind) : null;54 const rows = await sql<AlertRow[]>`select * from alerts where true ${q.before ? sql`and id < ${q.before}` : sql``} ${q.severity ? sql`and severity = ${q.severity}` : sql``}55 ${kinds ? sql`and kind = any(${kinds})` : sql``} ${q.unread ? sql`and read_at is null` : sql``} order by id desc limit ${q.limit}`;56 return { items: rows.map(view), ...(await unread()) };57 });58 app.get(`${P}/unread`, unread);59 app.post(`${P}/read`, async (req) => {60 const b = z.union([z.strictObject({ all: z.literal(true) }), z.strictObject({ ids: z.array(z.number().int().positive()).min(1).max(500) })]).parse(req.body);61 if ('all' in b) await sql`update alerts set read_at = now() where read_at is null`;62 else await sql`update alerts set read_at = now() where read_at is null and id = any(${b.ids})`;63 d.changed();64 return unread();65 });6667 // Server-Sent Events: `unread` now, then at every change. The stream ends by itself after ten minutes; the browser reconnects68 // through the session check, so a stream never outlives the session by more than that.69 let streams = 0;70 app.get(`${P}/stream`, async (req, reply) => {71 if (streams >= MAX_STREAMS) throw new UserError('too_many_streams', `at most ${MAX_STREAMS} alert streams at once`, 429);72 const first = await unread();73 streams += 1;74 reply.hijack();75 reply.raw.writeHead(200, { 'content-type': 'text/event-stream; charset=utf-8', 'cache-control': 'no-store', connection: 'keep-alive', 'x-accel-buffering': 'no' });76 const write = (chunk: string): void => { if (!reply.raw.writableEnded && !reply.raw.destroyed) reply.raw.write(chunk); };77 write(`retry: 5000\nevent: unread\ndata: ${JSON.stringify(first)}\n\n`);78 let busy = false, again = false;79 const push = (): void => {80 if (busy) { again = true; return; }81 busy = true;82 void unread().then((u) => write(`event: unread\ndata: ${JSON.stringify(u)}\n\n`), () => {}).finally(() => { busy = false; if (again) { again = false; push(); } });83 };84 const off = d.subscribe(push);85 const beat = setInterval(() => write(': keep-alive\n\n'), 15_000), stop = setTimeout(() => end(), STREAM_MAX_MS);86 beat.unref(); stop.unref();87 let done = false;88 const end = (): void => { if (done) return; done = true; streams -= 1; clearInterval(beat); clearTimeout(stop); off(); if (!reply.raw.writableEnded) reply.raw.end(); };89 req.raw.on('close', end); reply.raw.on('close', end); reply.raw.on('error', end);90 });9192 // ---- owner settings93 const settingsView = async (s: AlertSettings) => ({94 settings: s,95 kinds: ALERT_DEFS.map((x) => ({ kind: x.kind, group: x.group, severity: x.severity, default: x.on, fixed: !!x.fixed })),96 webhook: { secretSet: webhookSecret() !== null },97 /** exactly what a webhook call contains with the present "details" setting (shown in the UI before it is enabled) */98 webhookSample: JSON.parse(webhookBody({ id: '1', ts: new Date(0), kind: 'transfer.confirmed', severity: 'info', count: 1, params: { network: 'Base', symbol: 'USDC', amount: '25' } }, s.webhook.details)) as unknown,99 });100 app.get(`${P}/settings`, async () => settingsView(await alertSettings(true)));101 app.put(`${P}/settings`, async (req) => {102 const b = SettingsPatch.parse(req.body), cur = await alertSettings(true);103 for (const k of Object.keys(b.rules ?? {})) { const def = alertDef(k); if (!def || def.fixed) throw new UserError('bad_request', `unknown alert kind ${k.slice(0, 60)}`); }104 const next: AlertSettings = { rules: { ...cur.rules, ...b.rules }, quiet: { ...cur.quiet, ...b.quiet }, showDetails: b.showDetails ?? cur.showDetails, capPerHour: b.capPerHour ?? cur.capPerHour,105 thresholds: { ...cur.thresholds, ...b.thresholds }, webhook: { ...cur.webhook, ...b.webhook } };106 if (next.webhook.enabled) {107 if (!webhookUrlOk(next.webhook.url)) throw new UserError('bad_request', 'the webhook URL must be an https URL');108 if (webhookSecret() === null) throw new UserError('bad_request', 'generate the webhook secret first');109 } else if (next.webhook.url && !webhookUrlOk(next.webhook.url)) throw new UserError('bad_request', 'the webhook URL must be an https URL');110 const saved = await saveAlertSettings(next);111 const off = Object.entries(saved.rules).filter(([k, v]) => !v && cur.rules[k] !== v).map(([k]) => k), on = Object.entries(saved.rules).filter(([k, v]) => v && cur.rules[k] !== v).map(([k]) => k);112 // a record of who silenced what: URL host only, never the secret113 await audit('alerts.settings', 'owner', { off, on, quiet: saved.quiet.enabled, showDetails: saved.showDetails, webhook: saved.webhook.enabled ? { host: new URL(saved.webhook.url).host, details: saved.webhook.details } : null });114 return settingsView(saved);115 });116 app.post(`${P}/webhook/secret`, strict, async () => {117 const secret = rotateWebhookSecret();118 await audit('alerts.webhook_secret', 'owner', {});119 return { secret }; // shown once; only its presence is reported afterwards120 });121 app.post(`${P}/webhook/test`, strict, async () => {122 const s = await alertSettings(true), secret = webhookSecret();123 if (!s.webhook.url || !secret) throw new UserError('bad_request', 'set the URL and generate the secret first');124 const r = await sendWebhook(s.webhook.url, secret, { id: '0', ts: new Date(d.deps.now()), kind: 'system.test', severity: 'info', params: {}, count: 1 }, s.webhook.details, { fetch: d.deps.fetch, allowHttp: d.deps.allowHttpWebhook });125 return { ok: r.ok, status: r.status, error: r.error ?? null };126 });127128 // ---- Web Push: one subscription per device129 const PP = '/api/v1/push';130 const subs = () => sql<Parameters<typeof subView>[0][]>`select * from push_subscriptions order by created_at`;131 app.get(`${PP}/key`, async () => ({ publicKey: vapidKeys().publicKey }));132 app.get(`${PP}/subscriptions`, async () => (await subs()).map(subView));133 app.post(`${PP}/subscriptions`, async (req) => {134 const b = Subscription.parse(req.body);135 if (!endpointAllowed(b.endpoint)) throw new UserError('bad_request', 'this push endpoint is not one of the browsers’ push services');136 const [n] = await sql<{ n: number }[]>`select count(*)::int as n from push_subscriptions where endpoint <> ${b.endpoint}`;137 if ((n?.n ?? 0) >= MAX_SUBSCRIPTIONS) throw new UserError('too_many_subscriptions', `at most ${MAX_SUBSCRIPTIONS} devices`, 422);138 const ua = String(req.headers['user-agent'] ?? '').slice(0, 200);139 const [row] = await sql<Parameters<typeof subView>[0][]>`insert into push_subscriptions (id, endpoint, p256dh, auth, label, locale, ua) values (${randomUUID()}, ${b.endpoint}, ${b.keys.p256dh}, ${b.keys.auth}, ${b.label}, ${b.locale}, ${ua})140 on conflict (endpoint) do update set p256dh = excluded.p256dh, auth = excluded.auth, locale = excluded.locale, ua = excluded.ua,141 label = case when excluded.label <> '' then excluded.label else push_subscriptions.label end, failures = 0, last_error = null returning *`;142 await audit('alerts.push_subscribed', 'owner', { device: row!.id, label: row!.label, service: new URL(b.endpoint).hostname });143 return subView(row!);144 });145 app.patch(`${PP}/subscriptions/:id`, async (req) => {146 const id = z.uuid().parse((req.params as { id: string }).id), b = z.strictObject({ label: z.string().trim().max(40).optional(), locale: z.enum(['fr', 'en']).optional() }).parse(req.body);147 const [row] = await sql<Parameters<typeof subView>[0][]>`update push_subscriptions set label = coalesce(${b.label ?? null}, label), locale = coalesce(${b.locale ?? null}, locale) where id = ${id} returning *`;148 if (!row) throw new UserError('not_found', 'unknown device', 404);149 return subView(row);150 });151 app.delete(`${PP}/subscriptions/:id`, async (req) => {152 const id = z.uuid().parse((req.params as { id: string }).id);153 const gone = await sql`delete from push_subscriptions where id = ${id} returning id`;154 if (gone.length) await audit('alerts.push_unsubscribed', 'owner', { device: id });155 return { ok: true };156 });157 /** a test notification to one device (or all): the real path — encryption, VAPID, the push service — with a harmless payload */158 app.post(`${PP}/test`, strict, async (req) => {159 const b = z.strictObject({ id: z.uuid().optional() }).parse(req.body ?? {});160 const list = (await sql<{ id: string; endpoint: string; p256dh: string; auth: string; locale: string }[]>`select id, endpoint, p256dh, auth, locale from push_subscriptions`).filter((s) => !b.id || s.id === b.id);161 if (!list.length) throw new UserError('not_found', 'no subscribed device', 404);162 const a = { id: '0', kind: 'system.test', severity: 'info' as const, params: {}, subject: null, count: 1, ts: new Date(d.deps.now()) };163 const res = await Promise.all(list.map(async (s) => { const r = await d.pushOne(s, a, false); return { id: s.id, ok: r.ok, status: r.status, removed: !r.ok && r.gone, error: r.ok ? null : r.error }; }));164 return { results: res };165 });166}167