Repository Public
1// Event rules: one function that turns an event the system ALREADY recorded (a row of audit_events, deriv_risk_events or2// research_log) into zero or more alert drafts. Nothing here writes to those tables or calls the money code; lookups are reads.3// Every draft is a kind + parameters (codes, names, formatted numbers) — the sentence is rendered where the alert is shown.4import { formatUnits, getNetwork, type AlertParams, type AlertSeverity, type AlertSubject } from '@spw/core';5import type { sql as Sql } from '../db.ts';6import type { AlertSettings } from './store.ts';78type Q = typeof Sql;9export interface AlertEvent {10 source: 'audit' | 'risk' | 'research'; id: number; ts: Date;11 /** audit: the event type; risk: the event kind; research: the log kind */12 type: string; actor?: string; code?: string | null; message?: string; title?: string; accountId?: string; agentId?: string | null;13 data: Record<string, unknown>;14}15export interface Draft {16 kind: string; params: AlertParams; subject?: AlertSubject; severity?: AlertSeverity;17 /** a second alert with the same key is not created; with `bump` its counter is incremented instead (digests) */18 dedupe?: string; bump?: boolean;19}20export interface RuleCtx { q: Q; settings: AlertSettings }2122const str = (v: unknown, max = 200): string => (v === null || v === undefined ? '' : String(v).slice(0, max));23const netName = (id: unknown): string => { try { return getNetwork(String(id)).name; } catch { return str(id, 40); } };24const trim = (s: string, max = 6): string => { const [i, f = ''] = s.split('.'); const ff = f.slice(0, max).replace(/0+$/, ''); return ff ? `${i}.${ff}` : i!; };25const units = (amount: unknown, decimals: number): string => { try { return trim(formatUnits(BigInt(String(amount)), decimals)); } catch { return ''; } };26const bucket = (ts: Date, minutes: number): number => Math.floor(ts.getTime() / (minutes * 60_000));27const day = (ts: Date): string => ts.toISOString().slice(0, 10);28export const INTERVAL_MS: Record<string, number> = { '1m': 60_000, '5m': 300_000, '15m': 900_000, '1h': 3_600_000, '4h': 14_400_000, '1d': 86_400_000 };2930interface Proposal { id: string; kind: string; network: string; symbol: string; decimals: number; amount: string; error: string | null; swap: { toSymbol?: string } | null; bridge: { toNetwork?: string; phase?: string } | null; out_amount: string | null }31const proposal = async (q: Q, id: unknown): Promise<Proposal | null> => {32 if (typeof id !== 'string' || !/^[0-9a-f-]{36}$/.test(id)) return null;33 const [p] = await q<Proposal[]>`select id, kind, network, symbol, decimals, amount::text as amount, error, swap, bridge, out_amount::text as out_amount from proposals where id = ${id}`;34 return p ?? null;35};36export const agentName = async (q: Q, id: unknown): Promise<string> => {37 if (typeof id !== 'string' || !/^[0-9a-f-]{36}$/.test(id)) return '';38 const [a] = await q<{ name: string }[]>`select name from agents where id = ${id}`;39 return a?.name ?? '';40};41/** label and nature of a derivatives account: paper accounts are simulated and always said to be */42export const derivAccount = async (q: Q, id: unknown): Promise<{ label: string; mode: '@paper' | '@live'; paper: boolean; baseline: Record<string, string> }> => {43 if (typeof id === 'string' && /^[0-9a-f-]{36}$/.test(id)) {44 const [p] = await q<{ label: string }[]>`select label from deriv_paper_accounts where id = ${id}`;45 if (p) return { label: p.label, mode: '@paper', paper: true, baseline: {} };46 const [v] = await q<{ label: string; network: string; meta: { baseline?: Record<string, string> } }[]>`select label, network, meta from deriv_accounts where id = ${id}`;47 if (v) return { label: `${v.label}${v.network === 'testnet' ? ' (testnet)' : ''}`, mode: '@live', paper: false, baseline: v.meta?.baseline ?? {} };48 }49 return { label: '', mode: '@live', paper: false, baseline: {} };50};5152// exact decimal arithmetic on the sizes of fills (strings with up to 18 decimals)53const SCALE = 18n;54const toFixed = (s: string): bigint => {55 const m = /^(-?)(\d+)(?:\.(\d+))?$/.exec(s.trim());56 if (!m) return 0n;57 const v = BigInt(m[2]!) * 10n ** SCALE + BigInt((m[3] ?? '').slice(0, Number(SCALE)).padEnd(Number(SCALE), '0'));58 return m[1] ? -v : v;59};60const fromFixed = (v: bigint): string => { const neg = v < 0n, a = neg ? -v : v; return `${neg ? '-' : ''}${trim(`${a / 10n ** SCALE}.${(a % 10n ** SCALE).toString().padStart(Number(SCALE), '0')}`, 8)}`; };6162/** position opened / closed, from the net size the alerts layer keeps per (account, instrument) — updated once per fill event */63async function positionChange(c: RuleCtx, accountId: string, acct: Awaited<ReturnType<typeof derivAccount>>, instrument: string, signed: bigint, before: Date, pnl: string | null, subject: AlertSubject): Promise<Draft[]> {64 const key = `pos:${accountId}:${instrument}`;65 const [st] = await c.q<{ value: { net: string } }[]>`select value from alert_state where key = ${key}`;66 let prev: bigint;67 if (st) prev = toFixed(st.value.net);68 else {69 // first fill seen for this pair: start from what was recorded strictly before it (plus the venue baseline the owner accepted)70 const [s] = await c.q<{ net: string }[]>`select coalesce(sum(case when side = 'buy' then size else -size end), 0)::text as net from deriv_fills where account_id = ${accountId} and instrument = ${instrument} and ts < ${before}`;71 prev = toFixed(s?.net ?? '0') + toFixed(acct.baseline[instrument] ?? '0');72 }73 const next = prev + signed;74 await c.q`insert into alert_state (key, value, updated_at) values (${key}, ${c.q.json({ net: fromFixed(next) })}, now()) on conflict (key) do update set value = excluded.value, updated_at = now()`;75 const base = { account: acct.label, mode: acct.mode, instrument };76 if (prev === 0n && next !== 0n) return [{ kind: 'deriv.position_opened', params: { ...base, size: fromFixed(next) }, subject }];77 if (prev !== 0n && next === 0n) return [{ kind: 'deriv.position_closed', params: { ...base, pnl: pnl ? trim(pnl, 2) : '' }, subject }];78 return [];79}8081const refusedDigest = (ts: Date, reason: string): Draft => ({ kind: 'security.signer_refused', params: { reason: reason.slice(0, 160) }, dedupe: `signer_refused:${bucket(ts, 15)}`, bump: true });82/** which pinned program a refusal text is about */83export function pinOf(code: string, text: string): '@router' | '@jupiter' | '@cctp' | null {84 if (code === 'contract_changed' || /CCTP transfers are refused/.test(text)) return '@cctp';85 if (code === 'router_changed' || /router bytecode is not the reviewed|answered with another router|not the reviewed deployment/.test(text)) return /Jupiter/.test(text) ? '@jupiter' : '@router';86 return null;87}88export const pinDraft = (what: string, reason: string, ts: Date): Draft => ({ kind: 'security.pin_mismatch', params: { what, reason: reason.slice(0, 200) }, dedupe: `pin:${what}:${bucket(ts, 60)}` });8990async function fromAudit(e: AlertEvent, c: RuleCtx): Promise<Draft[]> {91 const d = e.data, t = e.type;92 if (t.startsWith('transfer.')) {93 const state = t.slice('transfer.'.length);94 if (state !== 'finalized' && state !== 'failed') return [];95 const p = await proposal(c.q, d.id);96 if (!p) return [];97 const subject: AlertSubject = { type: 'proposal', id: p.id, network: p.network };98 const base = { network: netName(p.network), symbol: p.symbol, amount: units(p.amount, p.decimals) };99 if (state === 'finalized') {100 if (p.kind === 'bridge') return []; // the bridge is complete when the destination minted: bridge.done101 return p.kind === 'swap' ? [{ kind: 'swap.confirmed', params: { ...base, toSymbol: str(p.swap?.toSymbol, 20) }, subject }] : [{ kind: 'transfer.confirmed', params: base, subject }];102 }103 const reason = str(d.error ?? p.error, 200);104 if (p.kind === 'bridge') return [{ kind: 'bridge.failed', params: { ...base, to: netName(p.bridge?.toNetwork), reason }, subject }];105 return [{ kind: p.kind === 'swap' ? 'swap.failed' : 'transfer.failed', params: { ...base, reason }, subject }];106 }107 switch (t) {108 case 'bridge.done': {109 const p = await proposal(c.q, d.id);110 if (!p) return [];111 return [{ kind: 'bridge.confirmed', params: { network: netName(p.network), to: netName(d.to ?? p.bridge?.toNetwork), symbol: p.symbol, amount: units(d.minted ?? p.out_amount ?? p.amount, p.decimals) }, subject: { type: 'proposal', id: p.id, network: p.network } }];112 }113 case 'bridge.mint_reverted': {114 const p = await proposal(c.q, d.id);115 if (!p) return [];116 return [{ kind: 'bridge.failed', params: { network: netName(p.network), to: netName(p.bridge?.toNetwork), symbol: p.symbol, amount: units(p.amount, p.decimals), reason: 'the mint transaction reverted on the destination; the funds are not lost, finish the transfer with your passkey' }, subject: { type: 'proposal', id: p.id, network: p.network } }];117 }118 case 'deposit.seen': {119 let symbol = '', amount = '';120 try { const n = getNetwork(String(d.network)), a = d.asset === 'native' ? n.native : n.tokens.find((x) => x.id === d.asset); if (a) { symbol = a.symbol; amount = units(d.amount, a.decimals); } } catch { /* unknown network: no detail */ }121 return [{ kind: 'deposit.received', params: { network: netName(d.network), symbol, amount }, subject: { type: 'deposit', id: str(d.txid, 120), network: str(d.network, 40) } }];122 }123 case 'agent.order_filled': {124 const agent = str(d.agent, 40), n = c.settings.thresholds.agentDigestMinutes;125 const [o] = await c.q<{ from_symbol: string; to_symbol: string }[]>`select from_symbol, to_symbol from agent_orders where id = ${typeof d.id === 'string' ? d.id : '00000000-0000-0000-0000-000000000000'}`;126 return [{ kind: 'agent.order_filled', params: { agent: await agentName(c.q, agent), from: o?.from_symbol ?? '', to: o?.to_symbol ?? '', count: 1 }, subject: { type: 'agent', id: agent },127 ...(n > 0 ? { dedupe: `agent.fill:${agent}:${bucket(e.ts, n)}`, bump: true } : {}) }];128 }129 case 'agent.stopped': {130 const id = str(d.id, 40), reason = str(d.reason, 240), agent = await agentName(c.q, id), subject: AlertSubject = { type: 'agent', id };131 if (/^loss limit/.test(reason)) return [{ kind: 'agent.stopped_loss', params: { agent, reason }, subject }];132 if (/mandate expired/.test(reason)) return [{ kind: 'agent.stopped_mandate', params: { agent }, subject }];133 return [{ kind: 'agent.stopped', params: { agent, reason }, subject }];134 }135 case 'mandate.revoked': return [{ kind: 'agent.mandate_revoked', params: { agent: await agentName(c.q, d.agent) }, subject: { type: 'agent', id: str(d.agent, 40) } }];136 case 'agent.order_refused': return [refusedDigest(e.ts, str(d.reason))];137 // >>> v7 research runtime: a job handed in a strategy (the server report is attached) / a job stopped without one138 case 'research.submitted': return [{ kind: 'agent.research_submitted', params: { name: (await agentName(c.q, d.agent)) || str(d.name, 40), count: Number(d.trials ?? 0) }, subject: d.agent ? { type: 'agent', id: str(d.agent, 40) } : { type: 'none' }, dedupe: `research:${str(d.job, 40)}` }];139 case 'research.failed': return [{ kind: 'agent.research_failed', params: { name: (await agentName(c.q, d.agent)) || str(d.name, 40), state: `@${str(d.state, 20)}`, reason: str(d.reason, 200) }, subject: d.agent ? { type: 'agent', id: str(d.agent, 40) } : { type: 'none' }, dedupe: `research:${str(d.job, 40)}` }];140 // <<< v7 research runtime141 case 'deriv.paper.fill': {142 const id = str(d.account, 40), acct = await derivAccount(c.q, id), subject: AlertSubject = { type: 'account', id };143 const [f] = await c.q<{ closed_pnl: string | null; ts: Date }[]>`select closed_pnl::text as closed_pnl, ts from deriv_fills where account_id = ${id} and fill_id = ${str(d.fill, 80)}`;144 const signed = toFixed(str(d.size, 60)) * (d.side === 'sell' ? -1n : 1n);145 return [{ kind: 'deriv.paper.filled', params: { account: acct.label, instrument: str(d.instrument, 80), side: d.side === 'sell' ? '@sell' : '@buy', size: trim(str(d.size, 60), 8), price: trim(str(d.price, 60), 8) }, subject },146 ...(await positionChange(c, id, acct, str(d.instrument, 80), signed, f?.ts ?? e.ts, f?.closed_pnl ?? null, subject))];147 }148 case 'deriv.paper.liquidation': case 'deriv.paper.reset': {149 const id = str(d.account, 40);150 await c.q`delete from alert_state where key like ${`pos:${id}:%`}`; // every position of the account is gone151 if (t === 'deriv.paper.reset') return [];152 return [{ kind: 'deriv.paper.liquidation', params: { account: (await derivAccount(c.q, id)).label, positions: Number(d.positions ?? 0) }, subject: { type: 'account', id } }];153 }154 case 'deriv.paper.halt': {155 const id = str(d.account, 40);156 return [{ kind: 'deriv.loss_limit', severity: 'warning', params: { account: (await derivAccount(c.q, id)).label, mode: '@paper', scope: d.code === 'daily_loss_exceeded' ? '@daily' : '@total' }, subject: { type: 'account', id } }];157 }158 case 'deriv.live.halt': {159 const id = str(d.account, 40);160 return [{ kind: 'deriv.halt', params: { account: (await derivAccount(c.q, id)).label, reason: str(d.reason, 200) }, subject: { type: 'account', id } }];161 }162 case 'deriv.live.order_rejected': {163 const id = str(d.account, 40), acct = await derivAccount(c.q, id);164 return [{ kind: 'deriv.order_rejected', params: { account: acct.label, mode: acct.mode, reason: str(d.reason, 200), count: 1 }, subject: { type: 'account', id }, dedupe: `deriv.rejected:${id}:${bucket(e.ts, 10)}`, bump: true }];165 }166 case 'deriv.paper.policy_replaced': case 'deriv.live.policy_replaced': {167 const id = str(d.account, 40), acct = await derivAccount(c.q, id);168 return [{ kind: 'security.policy_changed', params: { what: '@derivatives', name: acct.label, detail: `${acct.label}${d.maxLeverage !== undefined ? `, max leverage ${str(d.maxLeverage, 10)}` : ''}` }, subject: { type: 'account', id } }];169 }170 case 'device.added': return [{ kind: 'security.passkey_added', params: { label: str(d.label, 40) }, subject: { type: 'none' } }];171 case 'login.failed': return [{ kind: 'security.login_failed', params: { ip: str(d.ip, 60), count: 1 }, dedupe: `login_failed:${bucket(e.ts, 15)}`, bump: true }];172 case 'login.ok': return newDevice(e, c);173 // v7 any-to-any plans: the plan saga records these itself (services/api/src/plan-exec.ts); every parameter comes from the event174 case 'plan.step_ready': return [{ kind: 'plan.step_ready', params: { what: str(d.what, 60), step: Number(d.step) || 0, steps: Number(d.steps) || 0, network: netName(d.network) }, subject: { type: 'plan', id: str(d.id, 40) }, dedupe: `plan_ready:${str(d.id, 40)}:${Number(d.step) || 0}:${e.id}` }];175 case 'plan.paused': return [{ kind: 'plan.paused', params: { what: str(d.what, 60), step: Number(d.step) || 0, steps: Number(d.steps) || 0, network: netName(d.network), reason: str(d.reason, 200) }, subject: { type: 'plan', id: str(d.id, 40) } }];176 case 'plan.done': {177 const [pl] = /^[0-9a-f-]{36}$/.test(str(d.id, 40)) ? await c.q<{ meta: { to?: { network: string; asset: string } } }[]>`select meta from plans where id = ${str(d.id, 40)}` : [];178 let symbol = '', amount = '';179 try { const to = pl?.meta.to; if (to) { const n = getNetwork(to.network), a = to.asset === 'native' ? n.native : n.tokens.find((x) => x.id.toLowerCase() === to.asset.toLowerCase()); if (a) { symbol = a.symbol; amount = units(d.received, a.decimals); } } } catch { /* no detail */ }180 return [{ kind: 'plan.done', params: { what: str(d.what, 60), steps: Number(d.steps) || 0, amount, symbol }, subject: { type: 'plan', id: str(d.id, 40) } }];181 }182 default:183 }184 if (t.startsWith('deriv.live.order_') && d.final === true) {185 // a venue order reached a final state: an alert when it filled, and the position it leaves behind186 const id = str(d.account, 40), filled = toFixed(str(d.filled, 60) || '0');187 const [o] = await c.q<{ instrument: string; side: string }[]>`select instrument, side from deriv_orders where account_id = ${id} and client_id = ${str(d.clientId, 80)}`;188 if (!o) return [];189 const acct = await derivAccount(c.q, id), subject: AlertSubject = { type: 'account', id }, out: Draft[] = [];190 if (t === 'deriv.live.order_filled') out.push({ kind: 'deriv.live.filled', params: { account: acct.label, instrument: o.instrument, side: o.side === 'sell' ? '@sell' : '@buy', size: trim(str(d.filled, 60), 8), price: trim(str(d.avgFillPrice, 60), 8) }, subject });191 if (filled !== 0n) out.push(...(await positionChange(c, id, acct, o.instrument, o.side === 'sell' ? -filled : filled, e.ts, null, subject)));192 return out;193 }194 if (t.endsWith('.denied')) return [refusedDigest(e.ts, str(d.reason ?? t))];195 return [];196}197198/** a sign-in whose browser was never seen: the set of known user agents lives in alert_state and starts from the sessions on record */199async function newDevice(e: AlertEvent, c: RuleCtx): Promise<Draft[]> {200 const [s] = await c.q<{ ua: string; ip: string }[]>`select ua, ip from sessions where ip = ${str(e.data.ip, 80)} order by created_at desc limit 1`;201 if (!s) return [];202 const key = 'known_devices';203 const [st] = await c.q<{ value: { ua: string[] } }[]>`select value from alert_state where key = ${key}`;204 let known: string[];205 if (st) known = st.value.ua;206 else {207 // first use: the browsers that opened a session before this sign-in are known; on a brand-new wallet the first device is not "new"208 known = (await c.q<{ ua: string }[]>`select distinct ua from sessions where created_at < ${new Date(e.ts.getTime() - 2000)}`).map((r) => r.ua);209 if (known.length === 0) known = [s.ua];210 }211 const fresh = !known.includes(s.ua);212 if (fresh || !st) await c.q`insert into alert_state (key, value, updated_at) values (${key}, ${c.q.json({ ua: [...new Set([...known, s.ua])].slice(-50) })}, now()) on conflict (key) do update set value = excluded.value, updated_at = now()`;213 return fresh ? [{ kind: 'security.new_device', params: { device: describeUa(s.ua), ip: s.ip }, subject: { type: 'none' } }] : [];214}215export function describeUa(ua: string): string {216 const os = /iPhone/.test(ua) ? 'iPhone' : /iPad/.test(ua) ? 'iPad' : /Android/.test(ua) ? 'Android' : /Mac OS X|Macintosh/.test(ua) ? 'Mac' : /Windows/.test(ua) ? 'Windows' : /Linux/.test(ua) ? 'Linux' : '';217 const br = /Edg\//.test(ua) ? 'Edge' : /Firefox\//.test(ua) ? 'Firefox' : /Chrome\//.test(ua) ? 'Chrome' : /Safari\//.test(ua) ? 'Safari' : '';218 return [br, os].filter(Boolean).join(' · ') || ua.slice(0, 40) || '?';219}220221async function fromRisk(e: AlertEvent, c: RuleCtx): Promise<Draft[]> {222 const id = e.accountId ?? '', acct = await derivAccount(c.q, id), subject: AlertSubject = { type: 'account', id }, base = { account: acct.label, mode: acct.mode };223 switch (e.type) {224 case 'margin_warning': return [{ kind: 'deriv.margin_warning', params: { ...base, reason: str(e.message, 240) }, subject, dedupe: `margin:${id}:${bucket(e.ts, 60)}` }];225 case 'order_refused':226 if (String(e.code ?? '').startsWith('signer')) return [refusedDigest(e.ts, str(e.message))];227 return [{ kind: 'deriv.order_rejected', params: { ...base, reason: str(e.message, 200), count: 1 }, subject, dedupe: `deriv.rejected:${id}:${bucket(e.ts, 10)}`, bump: true }];228 case 'divergence': return [{ kind: 'deriv.divergence', params: { account: acct.label, reason: str(e.message, 300) }, subject }];229 case 'divergence_cleared': return [{ kind: 'deriv.divergence_cleared', params: { account: acct.label }, subject }];230 default: return []; // liquidation, loss limit and halt come from the audit chain231 }232}233234async function fromResearch(e: AlertEvent, c: RuleCtx): Promise<Draft[]> {235 if (e.type !== 'hold' || !e.agentId) return [];236 const text = str(e.message, 300);237 if (e.title === 'Order not placed') {238 const what = pinOf('', text);239 return what ? [pinDraft(what, text, e.ts)] : [];240 }241 if (e.title !== 'No valid signal: holding') return [];242 // one such entry is written per bar at most: N of them inside N bars are N failures in a row243 const n = c.settings.thresholds.signalFailures;244 const [a] = await c.q<{ name: string; interval: string }[]>`select name, interval from agents where id = ${e.agentId}`;245 if (!a) return [];246 const span = (INTERVAL_MS[a.interval] ?? 3_600_000) * n + 120_000;247 const [r] = await c.q<{ n: number }[]>`select count(*)::int as n from research_log where agent_id = ${e.agentId} and kind = 'hold' and title = 'No valid signal: holding' and id <= ${e.id} and ts > ${new Date(e.ts.getTime() - span)}`;248 if ((r?.n ?? 0) < n) return [];249 return [{ kind: 'agent.signal_failures', params: { agent: a.name, count: r!.n }, subject: { type: 'agent', id: e.agentId }, dedupe: `sigfail:${e.agentId}:${day(e.ts)}` }];250}251252export async function deriveAlerts(e: AlertEvent, c: RuleCtx): Promise<Draft[]> {253 return e.source === 'audit' ? fromAudit(e, c) : e.source === 'risk' ? fromRisk(e, c) : fromResearch(e, c);254}255