Repository Public
1/**2 * Events: persisted rows in `events` (cursor = id) + NOTIFY doorbell on `aos_events`.3 * Persisted events are never shipped inside the NOTIFY payload: the doorbell only carries the last id4 * and listeners read by cursor, so the 8000-byte NOTIFY limit cannot affect them.5 * Transient `model.delta` events travel ONLY in the NOTIFY payload (chunked when too large).6 */7import type pg from 'pg';8import type { AosEvent, EventKind, Trust } from '@aos/shared';9import { EVENTS_CHANNEL, getKernel, iso, withTx, type Kernel, type NewEvent, type Row } from './db.js';1011/** Postgres refuses NOTIFY payloads of 8000 bytes or more. */12export const NOTIFY_MAX_BYTES = 7900;1314/** Append one persisted event outside any run (server notices, tests). Returns its id. */15export async function appendEvent(ev: NewEvent, k: Kernel = getKernel()): Promise<number> {16 return withTx(k, async (tx) => {17 tx.emit(ev);18 const ids = await tx.flush();19 return ids[0] as number;20 });21}2223function rowToEvent(r: Row): AosEvent {24 return {25 id: Number(r.id),26 missionId: String(r.mission_id),27 runId: r.run_id === null ? null : String(r.run_id),28 seq: r.seq === null ? null : Number(r.seq),29 ts: iso(r.ts),30 kind: r.kind as EventKind,31 trust: r.trust as Trust,32 payload: (r.payload ?? {}) as Record<string, unknown>,33 };34}3536/** Persisted events with id > cursor, in id order (for the server's WS replay). */37export async function readEvents(38 cursor: number,39 opts: { limit?: number; missionId?: string; kernel?: Kernel } = {},40): Promise<AosEvent[]> {41 const k = opts.kernel ?? getKernel();42 const params: unknown[] = [cursor, opts.limit ?? 500];43 let where = 'id > $1';44 if (opts.missionId) {45 params.push(opts.missionId);46 where += ' AND mission_id = $3';47 }48 const r = await k.pool.query(`SELECT id, mission_id, run_id, seq, ts, kind, trust, payload FROM events WHERE ${where} ORDER BY id LIMIT $2`, params);49 return (r.rows as Row[]).map(rowToEvent);50}5152export interface ListenOptions {53 kernel?: Kernel;54 /** Deliver persisted events with id > cursor. Default: only events created after the subscription starts. */55 cursor?: number;56 missionId?: string;57 pollMs?: number; // default 200058 signal?: AbortSignal;59}60export interface EventSubscription {61 /** Resolves (never rejects) once LISTEN is active and the initial cursor is known, or the first attempt failed. */62 ready: Promise<void>;63 cursor(): number;64 close(): Promise<void>;65}6667/**68 * LISTEN + poll fallback. Persisted events are delivered exactly once, in id order, by cursor.69 * Transient deltas are delivered best-effort with id -1. `cb` errors are logged and swallowed.70 */71export function listenEvents(cb: (ev: AosEvent) => void, opts: ListenOptions = {}): EventSubscription {72 const k = opts.kernel ?? getKernel();73 const pollMs = opts.pollMs ?? 2000;74 let cursor = opts.cursor ?? -1;75 let closed = false;76 let client: pg.PoolClient | null = null;77 let timer: NodeJS.Timeout | null = null;78 let draining: Promise<void> | null = null;79 let again = false;8081 const deliver = (ev: AosEvent): void => {82 if (opts.missionId && ev.missionId !== opts.missionId) return;83 try {84 cb(ev);85 } catch (err) {86 k.log.warn({ err: (err as Error).message }, 'listenEvents callback threw');87 }88 };89 const drainOnce = async (): Promise<void> => {90 if (cursor < 0) {91 // "From now on": start after the newest event. Retried by the poll if the database is down.92 const r = await k.pool.query('SELECT COALESCE(max(id), 0) AS id FROM events');93 if (cursor < 0) cursor = Number((r.rows[0] as { id: unknown }).id);94 return;95 }96 for (;;) {97 if (closed) return;98 const batch = await readEvents(cursor, { limit: 500, kernel: k });99 for (const ev of batch) {100 if (closed) return;101 cursor = ev.id;102 deliver(ev);103 }104 if (batch.length < 500) return;105 }106 };107 const drain = (): Promise<void> => {108 if (draining) {109 again = true;110 return draining;111 }112 draining = (async () => {113 try {114 do {115 again = false;116 await drainOnce();117 } while (again && !closed);118 } catch (err) {119 k.log.warn({ err: (err as Error).message }, 'listenEvents drain failed (will retry on next poll)');120 } finally {121 draining = null;122 }123 })();124 return draining;125 };126 const onNotification = (msg: pg.Notification): void => {127 if (msg.channel !== EVENTS_CHANNEL || !msg.payload) return;128 if (msg.payload.startsWith('{"t":')) {129 try {130 const d = JSON.parse(msg.payload) as { m: string; r: string | null; k: EventKind; p: Record<string, unknown>; ts: string };131 deliver({ id: -1, missionId: d.m, runId: d.r, seq: null, ts: d.ts, kind: d.k, trust: 'agent', payload: d.p });132 } catch {133 /* malformed transient payload: drop */134 }135 return;136 }137 void drain();138 };139 const dropClient = (err?: Error): void => {140 const c = client;141 client = null;142 if (c) {143 c.removeListener('notification', onNotification);144 c.release(err ?? true); // never return a LISTENing connection to the pool145 }146 };147 const connect = async (): Promise<void> => {148 if (closed || client) return;149 const c = await k.pool.connect();150 c.on('notification', onNotification);151 c.once('error', (err) => {152 if (client === c) dropClient(err);153 });154 client = c;155 await c.query(`LISTEN ${EVENTS_CHANNEL}`);156 if (closed) dropClient();157 };158 const tick = async (): Promise<void> => {159 try {160 if (!client) await connect(); // reconnect after a connection loss161 } catch (err) {162 dropClient(err as Error);163 }164 await drain();165 };166167 const ready = (async () => {168 try {169 await connect();170 } catch (err) {171 dropClient(err as Error);172 k.log.warn({ err: (err as Error).message }, 'listenEvents: LISTEN failed, falling back to polling');173 }174 await drain(); // initial cursor (or first catch-up read); never rejects175 if (!closed) timer = setInterval(() => void tick(), pollMs);176 })();177178 const close = async (): Promise<void> => {179 if (closed) return;180 closed = true;181 if (timer) clearInterval(timer);182 await ready.catch(() => undefined);183 if (timer) clearInterval(timer);184 if (draining) await draining;185 dropClient();186 };187 opts.signal?.addEventListener('abort', () => void close(), { once: true });188 return { ready, cursor: () => cursor, close };189}190191// ------------------------------------------------------------------------------------------192// Transient events (NOTIFY only).193194/** Splits an oversized delta on its `text` field so every NOTIFY payload fits; returns [] if impossible. */195export function chunkTransient(base: { m: string; r: string | null; k: string; ts: string }, payload: Record<string, unknown>): string[] {196 const whole = JSON.stringify({ t: 1, ...base, p: payload });197 if (Buffer.byteLength(whole) <= NOTIFY_MAX_BYTES) return [whole];198 const text = payload.text;199 if (typeof text !== 'string') return [];200 const overhead = Buffer.byteLength(JSON.stringify({ t: 1, ...base, p: { ...payload, text: '' } }));201 const room = NOTIFY_MAX_BYTES - overhead;202 if (room < 64) return [];203 const out: string[] = [];204 let piece = '';205 let size = 0;206 for (const ch of text) {207 const b = Buffer.byteLength(JSON.stringify(ch)) - 2;208 if (size + b > room && piece) {209 out.push(JSON.stringify({ t: 1, ...base, p: { ...payload, text: piece } }));210 piece = '';211 size = 0;212 }213 piece += ch;214 size += b;215 }216 if (piece) out.push(JSON.stringify({ t: 1, ...base, p: { ...payload, text: piece } }));217 return out;218}219220/** Ordered, bounded, fire-and-forget sender of transient deltas for one run. Never throws. */221export class TransientSender {222 private chain: Promise<void> = Promise.resolve();223 private backlog = 0;224 dropped = 0;225 constructor(226 private readonly k: Kernel,227 private readonly missionId: string,228 private readonly runId: string,229 private readonly maxBacklog = 512,230 ) {}231232 send(kind: string, payload: Record<string, unknown>): void {233 let msgs: string[];234 try {235 msgs = chunkTransient({ m: this.missionId, r: this.runId, k: kind, ts: new Date().toISOString() }, payload);236 } catch {237 msgs = [];238 }239 if (msgs.length === 0 || this.backlog + msgs.length > this.maxBacklog) {240 this.dropped++;241 return;242 }243 for (const m of msgs) {244 this.backlog++;245 this.chain = this.chain.then(async () => {246 try {247 await this.k.pool.query('SELECT pg_notify($1, $2)', [EVENTS_CHANNEL, m]);248 } catch {249 this.dropped++;250 } finally {251 this.backlog--;252 }253 });254 }255 }256 idle(): Promise<void> {257 return this.chain;258 }259}260