import { useQueryClient } from "@tanstack/react-query"; import { useCallback, useEffect, useRef } from "react"; import { serverEventNames, type ServerEventPayloadMap } from "~/schemas/server-events"; type LifecycleEventPayloadMap = { connected: { type: "connected"; timestamp: number }; heartbeat: { timestamp: number }; }; type ServerEventsPayloadMap = LifecycleEventPayloadMap & ServerEventPayloadMap; type ServerEventType = keyof ServerEventsPayloadMap; type EventHandler = (data: ServerEventsPayloadMap[T]) => void; type EventHandlerSet = Set>; type EventHandlerMap = { [K in ServerEventType]?: EventHandlerSet; }; const invalidatingEvents = new Set([ "backup:completed", "volume:updated", "volume:status_changed", "mirror:completed", "doctor:started", "doctor:completed", "doctor:cancelled", ]); export type RestoreEvent = ServerEventsPayloadMap["restore:started"] | ServerEventsPayloadMap["restore:completed"]; export type RestoreProgressEvent = ServerEventsPayloadMap["restore:progress"]; export type RestoreCompletedEvent = ServerEventsPayloadMap["restore:completed"]; export type BackupProgressEvent = ServerEventsPayloadMap["backup:progress"]; const parseEventData = (event: Event): ServerEventsPayloadMap[T] => JSON.parse((event as MessageEvent).data) as ServerEventsPayloadMap[T]; /** * Hook to listen to Server-Sent Events (SSE) from the backend * Automatically handles cache invalidation for backup and volume events */ export function useServerEvents() { const queryClient = useQueryClient(); const eventSourceRef = useRef(null); const handlersRef = useRef({}); const emit = useCallback((eventName: T, data: ServerEventsPayloadMap[T]) => { const handlers = handlersRef.current[eventName] as EventHandlerSet | undefined; handlers?.forEach((handler) => { handler(data); }); }, []); useEffect(() => { const eventSource = new EventSource("/api/v1/events"); eventSourceRef.current = eventSource; eventSource.addEventListener("connected", (event) => { const data = parseEventData<"connected">(event); console.info("[SSE] Connected to server events"); emit("connected", data); }); eventSource.addEventListener("heartbeat", (event) => { emit("heartbeat", parseEventData<"heartbeat">(event)); }); for (const eventName of serverEventNames) { eventSource.addEventListener(eventName, (event) => { const data = parseEventData(event); console.info(`[SSE] ${eventName}:`, data); if (invalidatingEvents.has(eventName)) { void queryClient.invalidateQueries(); } if (eventName === "backup:completed") { void queryClient.refetchQueries(); } if (eventName === "volume:status_changed") { const statusData = data as ServerEventsPayloadMap["volume:status_changed"]; emit("volume:status_changed", statusData); emit("volume:updated", statusData); return; } emit(eventName, data); }); } eventSource.onerror = (error) => { console.error("[SSE] Connection error:", error); }; return () => { console.info("[SSE] Disconnecting from server events"); eventSource.close(); eventSourceRef.current = null; }; }, [emit, queryClient]); const addEventListener = useCallback( (eventName: T, handler: EventHandler, options?: { signal?: AbortSignal }) => { if (options?.signal?.aborted) { return () => {}; } const existingHandlers = handlersRef.current[eventName] as EventHandlerSet | undefined; const eventHandlers = existingHandlers ?? new Set>(); eventHandlers.add(handler); handlersRef.current[eventName] = eventHandlers as EventHandlerMap[T]; const unsubscribe = () => { const handlers = handlersRef.current[eventName] as EventHandlerSet | undefined; handlers?.delete(handler); if (handlers && handlers.size === 0) { delete handlersRef.current[eventName]; } if (options?.signal) { options.signal.removeEventListener("abort", unsubscribe); } }; if (options?.signal) { options.signal.addEventListener("abort", unsubscribe, { once: true }); } return unsubscribe; }, [], ); return { addEventListener }; }