Repository Public
1'use client';2/**3 * Single shared connection to WS /api/v1/events?cursor=<lastEventId>.4 *5 * Resume: the id of the last persisted event is kept in memory; every reconnect passes it as6 * `cursor`, so the server replays exactly what was missed. Persisted events at or below the7 * cursor are ignored (a replay overlapping what we already saw changes nothing). Without a cursor8 * (nothing received yet) or after a `system.notice` gap, everything on screen is refetched.9 * Events are signals: they invalidate queries / trigger a journal sync; nothing is rebuilt from10 * event payloads alone, except transient deltas.11 */12import '@/lib/zod-config';13import type { QueryClient } from '@tanstack/react-query';14import { AosEvent, type AosSettings } from '@aos/shared';15import { Notification } from './dto-auth';16import { qs, redirectToLogin, wsUrl } from './api';17import { DEFAULT_SETTINGS, qk } from './queries';18import { useConn } from '@/stores/conn';19import { toast, useToasts } from '@/stores/toasts';20import { timeline, useTimeline } from '@/stores/timeline';2122let ws: WebSocket | null = null;23let qc: QueryClient | null = null;24let attempt = 0;25let timer: ReturnType<typeof setTimeout> | null = null;26let stopped = true;27/** Highest persisted event id seen (the resume cursor). Memory only. */28let cursor: number | null = null;29let connectedWithCursor = false;30let everOpened = false;3132const pendingKeys = new Map<string, readonly unknown[]>();33let flushTimer: ReturnType<typeof setTimeout> | null = null;34/** Coalesce invalidations: a burst of events refetches each list once. */35function invalidate(...keys: (readonly unknown[])[]): void {36 for (const k of keys) pendingKeys.set(JSON.stringify(k), k);37 flushTimer ??= setTimeout(() => {38 flushTimer = null;39 const list = [...pendingKeys.values()];40 pendingKeys.clear();41 for (const k of list) void qc?.invalidateQueries({ queryKey: k });42 }, 120);43}44function refetchEverything(): void {45 void qc?.invalidateQueries();46 timeline.resyncAll();47}48const fresh = (ts: string): boolean => Math.abs(Date.now() - new Date(ts).getTime()) < 20_000;4950function handle(ev: AosEvent): void {51 if (ev.id > 0) {52 if (cursor !== null && ev.id <= cursor) return; // already seen (overlapping replay)53 cursor = ev.id;54 }55 timeline.onEvent(ev);56 const m = ev.missionId;57 switch (ev.kind) {58 case 'model.delta':59 return;60 case 'mission.created':61 case 'mission.status':62 // a terminal status also creates an in-app notification on the server63 invalidate(qk.missions, qk.mission(m), qk.runs(m), qk.projects, qk.notifications);64 break;65 case 'run.created':66 case 'run.status':67 invalidate(qk.missions, qk.mission(m), qk.runs(m), qk.waits);68 break;69 case 'run.usage':70 invalidate(qk.missions, qk.mission(m), qk.runs(m), qk.spend);71 break;72 case 'step.completed':73 invalidate(qk.mission(m)); // done criteria may have been flipped by update_criteria74 break;75 case 'wait.created': {76 invalidate(qk.waits, qk.missions, qk.mission(m), qk.runs(m), qk.notifications);77 const settings = qc?.getQueryData<AosSettings>(qk.settings) ?? DEFAULT_SETTINGS;78 if (settings.notifications.inApp && fresh(ev.ts) && !location.pathname.startsWith('/approvals') && !location.pathname.startsWith('/home')) {79 // wait.created = {waitId, kind, payload, resolveBy}; the title / question text lives in the wait's own payload80 const question = ev.payload.kind === 'question';81 const wp = (ev.payload.payload ?? {}) as Record<string, unknown>;82 const label = question ? wp.text : wp.title;83 const detail = settings.notifications.detail === 'titles' && typeof label === 'string' ? label.slice(0, 160) : undefined;84 useToasts.getState().push({ title: question ? 'The agent has a question' : 'Approval needed', body: detail, tone: 'warning', href: '/approvals/' });85 }86 break;87 }88 case 'wait.resolved':89 invalidate(qk.waits, qk.missions, qk.mission(m), qk.runs(m), qk.policy);90 break;91 case 'artifact.created':92 if (ev.runId) invalidate(qk.artifacts(ev.runId));93 break;94 case 'system.notice':95 // server notices: {code:'gap', message, latestId} and {code:'auth.expired'} (id -1, missionId '');96 // engine notice: {code:'criteria', doneCriteria} when criteria are confirmed or updated.97 if (ev.payload.code === 'gap') {98 if (typeof ev.payload.latestId === 'number' && (cursor === null || ev.payload.latestId > cursor)) cursor = ev.payload.latestId;99 refetchEverything();100 } else if (ev.payload.code === 'auth.expired') redirectToLogin();101 else if (ev.payload.code === 'criteria' && m) invalidate(qk.mission(m), qk.missions);102 break;103 default:104 break;105 }106}107/** Frames of the previous protocol the server may still send next to AosEvent frames. */108function handleLegacy(raw: unknown): void {109 const o = raw as { type?: unknown; notification?: unknown };110 if (o.type === 'auth.expired') return redirectToLogin();111 if (o.type === 'notification') {112 const n = Notification.safeParse(o.notification);113 if (!n.success || !qc) return;114 qc.setQueryData<Notification[]>(qk.notifications, (l) => (l ? [n.data, ...l.filter((x) => x.id !== n.data.id)] : [n.data]));115 const settings = qc.getQueryData<AosSettings>(qk.settings) ?? DEFAULT_SETTINGS;116 if (settings.notifications.inApp) toast.info(n.data.title, n.data.body);117 }118}119120function connect(): void {121 if (stopped || ws) return;122 useConn.getState().set({ events: 'connecting' });123 connectedWithCursor = cursor !== null;124 const sock = new WebSocket(wsUrl(`/events${qs({ cursor })}`));125 ws = sock;126 sock.onopen = () => {127 attempt = 0;128 useConn.getState().set({ events: 'open' });129 // Reconnected without a cursor: nothing tells us what was missed.130 if (everOpened && !connectedWithCursor) refetchEverything();131 everOpened = true;132 };133 sock.onmessage = (m) => {134 if (typeof m.data !== 'string') return;135 try {136 const json: unknown = JSON.parse(m.data);137 const parsed = AosEvent.safeParse(json);138 if (parsed.success) handle(parsed.data);139 else handleLegacy(json);140 } catch {141 /* ignore malformed frames */142 }143 };144 sock.onclose = (e) => {145 if (ws === sock) ws = null;146 useConn.getState().set({ events: 'closed' });147 if (e.code === 4401) return redirectToLogin();148 schedule();149 };150}151function schedule(): void {152 if (stopped || timer) return;153 const delay = Math.min(10_000, 500 * 2 ** attempt++) * (0.7 + Math.random() * 0.6);154 timer = setTimeout(() => {155 timer = null;156 connect();157 }, delay);158}159function wake(): void {160 if (stopped || ws || document.visibilityState !== 'visible') return;161 if (timer) clearTimeout(timer);162 timer = null;163 attempt = 0;164 connect();165}166167export function startEvents(client: QueryClient): void {168 qc = client;169 if (!stopped) return;170 stopped = false;171 connect();172 document.addEventListener('visibilitychange', wake);173 window.addEventListener('online', wake);174 window.addEventListener('pageshow', wake);175 if (process.env.NEXT_PUBLIC_E2E) {176 // Test hook, compiled out of production builds (NEXT_PUBLIC_E2E is inlined as '').177 (window as unknown as { __aos: unknown }).__aos = {178 dropEvents: () => ws?.close(),179 cursor: () => cursor,180 seqs: (runId: string) => useTimeline.getState().runs[runId]?.steps.map((s) => s.seq) ?? null,181 };182 }183}184export function stopEvents(): void {185 stopped = true;186 if (timer) clearTimeout(timer);187 timer = null;188 ws?.close();189 ws = null;190 document.removeEventListener('visibilitychange', wake);191 window.removeEventListener('online', wake);192 window.removeEventListener('pageshow', wake);193}194