From 57694a48a62eba6a6ae411d9ea8ba66ce0ef7f74 Mon Sep 17 00:00:00 2001 From: Vaibhaav Date: Fri, 26 Jun 2026 04:01:41 +0530 Subject: [PATCH] feat: add shared SSE event transport for real-time sync --- packages/shared/src/index.ts | 2 + .../shared/src/lib/event-transport.test.ts | 31 +++++++ packages/shared/src/lib/event-transport.ts | 89 +++++++++++++++++++ packages/shared/src/lib/events-connection.ts | 22 +++++ 4 files changed, 144 insertions(+) create mode 100644 packages/shared/src/lib/event-transport.test.ts create mode 100644 packages/shared/src/lib/event-transport.ts create mode 100644 packages/shared/src/lib/events-connection.ts diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index cf2f4b91e..6fd690a89 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -3,5 +3,7 @@ export * from './api/client'; export * from './api/paths'; export * from './api/types'; export * from './lib/bridge'; +export * from './lib/event-transport'; +export * from './lib/events-connection'; export * from './hooks/useWorkspaceQuery'; export * from './hooks/useDaemonStatus'; diff --git a/packages/shared/src/lib/event-transport.test.ts b/packages/shared/src/lib/event-transport.test.ts new file mode 100644 index 000000000..aa579d0bf --- /dev/null +++ b/packages/shared/src/lib/event-transport.test.ts @@ -0,0 +1,31 @@ +// packages/shared/src/lib/event-transport.test.ts +import { describe, it, expect, beforeEach } from '@jest/globals'; +import { createEventTransport } from './event-transport'; +import { QueryClient } from '@tanstack/react-query'; +import { setPlatformBridge } from './bridge'; + +describe('EventTransport', () => { + let queryClient: QueryClient; + + beforeEach(() => { + queryClient = new QueryClient(); + setPlatformBridge({ + getDaemonStatus: async () => ({ running: true, port: 3001 }), + startDaemon: async () => {}, + stopDaemon: async () => {}, + getApiBaseUrl: () => 'http://test:3001', + subscribeApiBaseUrl: () => () => {}, + }); + }); + + it('should create transport with connect method', () => { + const transport = createEventTransport(queryClient); + expect(typeof transport.connect).toBe('function'); + }); + + it('should return cleanup function from connect', () => { + const transport = createEventTransport(queryClient); + const cleanup = transport.connect(); + expect(typeof cleanup).toBe('function'); + }); +}); diff --git a/packages/shared/src/lib/event-transport.ts b/packages/shared/src/lib/event-transport.ts new file mode 100644 index 000000000..1d5c58d2d --- /dev/null +++ b/packages/shared/src/lib/event-transport.ts @@ -0,0 +1,89 @@ +// packages/shared/src/lib/event-transport.ts +import type { QueryClient } from '@tanstack/react-query'; +import { getPlatformBridge } from './bridge'; +import { setEventsConnectionState } from './events-connection'; +import { WORKSPACE_QUERY_KEY } from '../hooks/useWorkspaceQuery'; + +export type EventTransport = { + connect: () => () => void; +}; + +const INVALIDATE_DEBOUNCE_MS = 150; +const SSE_RETRY_MS = 5_000; +const EVENTSOURCE_CLOSED = 2; + +const CDC_EVENT_TYPES = [ + 'session_created', + 'session_updated', + 'pr_created', + 'pr_updated', + 'pr_check_recorded', + 'pr_session_changed', + 'pr_review_thread_added', + 'pr_review_thread_resolved', +] as const; + +export function createEventTransport(queryClient: QueryClient): EventTransport { + return { + connect() { + let debounce: ReturnType | undefined; + let retryTimer: ReturnType | undefined; + let source: EventSource | undefined; + let sourceBaseUrl: string | undefined; + + const refreshWorkspaces = () => { + if (debounce) clearTimeout(debounce); + debounce = setTimeout(() => { + void queryClient.invalidateQueries({ queryKey: WORKSPACE_QUERY_KEY }); + }, INVALIDATE_DEBOUNCE_MS); + }; + + const scheduleRetry = () => { + if (retryTimer) return; + retryTimer = setTimeout(() => { + retryTimer = undefined; + connectSource(); + }, SSE_RETRY_MS); + }; + + const bridge = getPlatformBridge(); + + const connectSource = () => { + if (typeof EventSource === 'undefined') return; + const baseUrl = bridge.getApiBaseUrl(); + if (source && sourceBaseUrl === baseUrl && source.readyState !== EVENTSOURCE_CLOSED) return; + source?.close(); + source = undefined; + sourceBaseUrl = baseUrl; + try { + source = new EventSource(`${baseUrl.replace(/\/+$/, '')}/api/v1/events`); + source.onopen = () => { + setEventsConnectionState('connected'); + refreshWorkspaces(); + }; + source.onerror = () => { + setEventsConnectionState('disconnected'); + if (source?.readyState === EVENTSOURCE_CLOSED) scheduleRetry(); + }; + source.onmessage = refreshWorkspaces; + for (const type of CDC_EVENT_TYPES) { + source.addEventListener(type, refreshWorkspaces); + } + } catch { + source = undefined; + } + }; + + const removeBaseUrlListener = bridge.subscribeApiBaseUrl(connectSource); + connectSource(); + + return () => { + if (debounce) clearTimeout(debounce); + if (retryTimer) clearTimeout(retryTimer); + removeBaseUrlListener(); + source?.close(); + setEventsConnectionState('idle'); + }; + }, + }; +} diff --git a/packages/shared/src/lib/events-connection.ts b/packages/shared/src/lib/events-connection.ts new file mode 100644 index 000000000..1ba73c0ad --- /dev/null +++ b/packages/shared/src/lib/events-connection.ts @@ -0,0 +1,22 @@ +// packages/shared/src/lib/events-connection.ts +type ConnectionState = 'idle' | 'connecting' | 'connected' | 'disconnected'; + +let connectionState: ConnectionState = 'idle'; +const listeners: Set<(state: ConnectionState) => void> = new Set(); + +export function setEventsConnectionState(state: ConnectionState): void { + connectionState = state; + listeners.forEach((listener) => listener(state)); +} + +export function getEventsConnectionState(): ConnectionState { + return connectionState; +} + +export function subscribeEventsConnectionState( + callback: (state: ConnectionState) => void +): () => void { + listeners.add(callback); + callback(connectionState); + return () => listeners.delete(callback); +}