Repository Public
1// Alerts storage: tables, owner settings, small state store. The tables are created here (not in db.ts) so the alerts layer stays2// in its own files; `migrateAlerts()` is idempotent and is called when the routes are registered and when the dispatcher starts.3import { z } from 'zod';4import { ALERT_DEFS, type AlertParams, type AlertSeverity, type AlertSubject } from '@spw/core';5import { getSetting, setSetting, sql } from '../db.ts';67const SCHEMA = `8create table if not exists alerts (9 id bigserial primary key, ts timestamptz not null, kind text not null, severity text not null,10 params jsonb not null default '{}', subject jsonb, dedupe_key text, source text not null default 'rule', source_id text,11 count int not null default 1, read_at timestamptz, delivery text not null default 'pending', channels jsonb not null default '{}',12 created_at timestamptz not null default now());13create unique index if not exists alerts_dedupe on alerts (dedupe_key) where dedupe_key is not null;14create unique index if not exists alerts_source on alerts (source, source_id, kind) where source_id is not null;15create index if not exists alerts_unread on alerts (id) where read_at is null;16create index if not exists alerts_pending on alerts (id) where delivery = 'pending';17-- one row per tailed table: the id up to which events were turned into alerts18-- (floor = the id at which tailing began: nothing at or before it is ever turned into an alert)19create table if not exists alert_cursors (source text primary key, cursor bigint not null, floor bigint not null default 0);20-- ids already handled inside the overlap window of a tailed table whose ids can commit out of order (bigserial)21create table if not exists alert_seen (source text not null, id bigint not null, primary key (source, id));22-- rule state that must survive a restart (debounce clocks, known devices, counters)23create table if not exists alert_state (key text primary key, value jsonb not null, updated_at timestamptz not null default now());24create table if not exists push_subscriptions (25 id uuid primary key, endpoint text unique not null, p256dh text not null, auth text not null, label text not null default '',26 locale text not null default 'fr', ua text not null default '', created_at timestamptz not null default now(),27 last_ok_at timestamptz, last_error text, failures int not null default 0);28`;29let migrated: Promise<void> | null = null;30export function migrateAlerts(force = false): Promise<void> {31 if (force || !migrated) migrated = sql.unsafe(SCHEMA).then(() => {});32 return migrated;33}3435export interface AlertRow {36 id: string; ts: Date; kind: string; severity: AlertSeverity; params: AlertParams; subject: AlertSubject | null; dedupe_key: string | null;37 source: string; source_id: string | null; count: number; read_at: Date | null; delivery: string; channels: Record<string, unknown>; created_at: Date;38}3940const HHMM = z.string().regex(/^([01]\d|2[0-3]):[0-5]\d$/);41const validTz = (tz: string): boolean => { try { new Intl.DateTimeFormat('en-CA', { timeZone: tz }); return true; } catch { return false; } };42export const ThresholdsSchema = z.strictObject({43 /** a transfer or swap still not final after this many minutes */44 stuckMinutes: z.number().int().min(5).max(1440),45 bridgeStuckMinutes: z.number().int().min(10).max(2880),46 /** at most one "order filled" notification per agent per this many minutes (0 = one per order) */47 agentDigestMinutes: z.number().int().min(0).max(1440),48 staleBars: z.number().int().min(1).max(100),49 signalFailures: z.number().int().min(1).max(50),50 /** absolute funding payment, in the account's collateral currency (USD) */51 fundingUsd: z.number().min(0).max(1_000_000),52 diskFreePct: z.number().min(1).max(50),53 backupHours: z.number().int().min(1).max(720),54 /** a connector must be down this long before an alert, so a flapping feed says nothing */55 feedDebounceSec: z.number().int().min(30).max(3600),56 unknownOrderSec: z.number().int().min(15).max(3600),57});58export const AlertSettingsSchema = z.strictObject({59 rules: z.record(z.string(), z.boolean()),60 quiet: z.strictObject({ enabled: z.boolean(), start: HHMM, end: HHMM, tz: z.string().min(1).max(64).refine(validTz, 'unknown time zone') }),61 /** amounts, prices and reasons in push notifications (lock screen). Off by default. */62 showDetails: z.boolean(),63 /** push notifications and webhook calls per hour, criticals excepted */64 capPerHour: z.number().int().min(1).max(500),65 thresholds: ThresholdsSchema,66 webhook: z.strictObject({ enabled: z.boolean(), url: z.string().max(500), details: z.boolean() }),67});68export type AlertSettings = z.infer<typeof AlertSettingsSchema>;6970export const DEFAULT_SETTINGS: AlertSettings = {71 rules: Object.fromEntries(ALERT_DEFS.filter((d) => !d.fixed).map((d) => [d.kind, d.on])),72 quiet: { enabled: false, start: '22:00', end: '07:00', tz: 'America/Toronto' },73 showDetails: false, capPerHour: 20,74 thresholds: { stuckMinutes: 30, bridgeStuckMinutes: 90, agentDigestMinutes: 30, staleBars: 3, signalFailures: 3, fundingUsd: 5, diskFreePct: 10, backupHours: 36, feedDebounceSec: 180, unknownOrderSec: 60 },75 webhook: { enabled: false, url: '', details: false },76};77const KEY = 'alerts';78let cachedSettings: { at: number; v: AlertSettings } | null = null;79/** stored settings merged over the defaults: a kind added by a release gets its default switch, an unknown key is dropped */80export async function alertSettings(fresh = false): Promise<AlertSettings> {81 if (!fresh && cachedSettings && Date.now() - cachedSettings.at < 3000) return cachedSettings.v;82 const s = await getSetting<Partial<AlertSettings>>(KEY, {});83 const rules = { ...DEFAULT_SETTINGS.rules };84 for (const k of Object.keys(rules)) if (typeof s.rules?.[k] === 'boolean') rules[k] = s.rules[k]!;85 const merged = { rules, quiet: { ...DEFAULT_SETTINGS.quiet, ...s.quiet }, showDetails: s.showDetails ?? DEFAULT_SETTINGS.showDetails, capPerHour: s.capPerHour ?? DEFAULT_SETTINGS.capPerHour,86 thresholds: { ...DEFAULT_SETTINGS.thresholds, ...s.thresholds }, webhook: { ...DEFAULT_SETTINGS.webhook, ...s.webhook } };87 const p = AlertSettingsSchema.safeParse(merged);88 const v = p.success ? p.data : DEFAULT_SETTINGS;89 cachedSettings = { at: Date.now(), v };90 return v;91}92export async function saveAlertSettings(v: AlertSettings): Promise<AlertSettings> {93 await setSetting(KEY, AlertSettingsSchema.parse(v));94 cachedSettings = null;95 return alertSettings(true);96}9798type Q = typeof sql;99export async function getState<T>(key: string, dflt: T, q: Q = sql): Promise<T> {100 const [r] = await q<{ value: T }[]>`select value from alert_state where key = ${key}`;101 return r ? r.value : dflt;102}103export async function putState(key: string, value: unknown, q: Q = sql): Promise<void> {104 await q`insert into alert_state (key, value, updated_at) values (${key}, ${q.json(value as never)}, now()) on conflict (key) do update set value = excluded.value, updated_at = now()`;105}106export async function delState(key: string, q: Q = sql): Promise<void> { await q`delete from alert_state where key = ${key}`; }107