Repository Public
1'use client';2/**3 * Single shared connection to WS /api/v1/events. Pushes server events into the react-query4 * cache; on every reconnect the affected lists are refetched (events may have been missed).5 */6import '@/lib/zod-config';7import type { QueryClient } from '@tanstack/react-query';8import { ServerEvent, type Notification, type UserSettings } from '@cr/shared';9import { redirectToLogin, wsUrl } from './api';10import { DEFAULT_SETTINGS, qk, upsertSession, upsertTask } from './queries';11import { useConn } from '@/stores/conn';12import { useToasts, type ToastTone } from '@/stores/toasts';13import type { TerminalSession, AgentTask } from '@cr/shared';1415let ws: WebSocket | null = null;16let qc: QueryClient | null = null;17let attempt = 0;18let timer: ReturnType<typeof setTimeout> | null = null;19let everOpened = false;20let stopped = true;2122const TONE: Record<Notification['kind'], ToastTone> = {23 'session.waiting': 'warning', 'session.exited': 'info', 'task.succeeded': 'success', 'task.failed': 'error',24 'node.offline': 'warning', 'storage.low': 'warning', system: 'info',25};2627function handle(ev: ServerEvent): void {28 if (!qc) return;29 switch (ev.type) {30 case 'hello':31 break;32 case 'session.updated':33 upsertSession(qc, ev.session);34 break;35 case 'session.removed':36 for (const a of [false, true]) qc.setQueryData<TerminalSession[]>(qk.sessions(a), (l) => l?.filter((s) => s.id !== ev.id));37 break;38 case 'task.updated':39 upsertTask(qc, ev.task);40 break;41 case 'notification': {42 const n = ev.notification;43 qc.setQueryData<Notification[]>(qk.notifications, (l) => (l ? [n, ...l.filter((x) => x.id !== n.id)] : [n]));44 const settings = qc.getQueryData<UserSettings>(qk.settings) ?? DEFAULT_SETTINGS;45 if (settings.notifications.inApp) useToasts.getState().push({ title: n.title, body: n.body, tone: TONE[n.kind], href: n.link ?? undefined });46 break;47 }48 case 'auth.expired':49 redirectToLogin();50 break;51 }52}5354function connect(): void {55 if (stopped || ws) return;56 useConn.getState().set({ events: 'connecting' });57 const sock = new WebSocket(wsUrl('/events'));58 ws = sock;59 sock.onopen = () => {60 attempt = 0;61 useConn.getState().set({ events: 'open' });62 if (everOpened && qc) {63 void qc.invalidateQueries({ queryKey: ['sessions'] });64 void qc.invalidateQueries({ queryKey: qk.tasks });65 void qc.invalidateQueries({ queryKey: qk.notifications });66 }67 everOpened = true;68 };69 sock.onmessage = (m) => {70 if (typeof m.data !== 'string') return;71 try {72 const parsed = ServerEvent.safeParse(JSON.parse(m.data));73 if (parsed.success) handle(parsed.data);74 } catch {75 /* ignore malformed frames */76 }77 };78 sock.onclose = (e) => {79 if (ws === sock) ws = null;80 useConn.getState().set({ events: 'closed' });81 if (e.code === 4401) return redirectToLogin();82 schedule();83 };84}85function schedule(): void {86 if (stopped || timer) return;87 const delay = Math.min(10_000, 500 * 2 ** attempt++) * (0.7 + Math.random() * 0.6);88 timer = setTimeout(() => {89 timer = null;90 connect();91 }, delay);92}93function wake(): void {94 if (stopped || ws || document.visibilityState !== 'visible') return;95 if (timer) clearTimeout(timer);96 timer = null;97 attempt = 0;98 connect();99}100101export function startEvents(client: QueryClient): void {102 qc = client;103 if (!stopped) return;104 stopped = false;105 connect();106 document.addEventListener('visibilitychange', wake);107 window.addEventListener('online', wake);108 window.addEventListener('pageshow', wake);109}110export function stopEvents(): void {111 stopped = true;112 if (timer) clearTimeout(timer);113 timer = null;114 ws?.close();115 ws = null;116 document.removeEventListener('visibilitychange', wake);117 window.removeEventListener('online', wake);118 window.removeEventListener('pageshow', wake);119}120export type { AgentTask };121