From 0cffce16de4638ede888971b0cf77ea25b4bc9d5 Mon Sep 17 00:00:00 2001 From: Nimatstar Date: Fri, 21 Aug 2026 08:49:38 +0000 Subject: [PATCH] feat(realtime): unified connection supervisor for reconnection, heartbeat and backpressure (#1143) Introduce a transport-agnostic ConnectionSupervisor (src/lib/realtime/connectionSupervisor.ts) that all realtime transports delegate their lifecycle to: exponential backoff with full jitter, shared heartbeat with ping-timeout detection, a bounded outbound queue with documented drop-oldest/block backpressure, inbound sequence tracking with catch-up, a resubscribe registry for rooms/subscriptions, and reconnection metrics plus alerts with graceful degradation to offline mode via the service worker. Refactor socket.io (websocketManager), the notification WebSocket and GraphQL subscriptions to use the supervisor, expose a single useRealtimeConnection status object adopted by useWebSocket, useCollaboration and useRealTimeAnalytics, and wire the synchronization engine to backfill state after a reconnect gap. --- public/sw.js | 11 +- src/app/store/messagingStore.ts | 12 +- src/constants/app.constants.ts | 13 + src/hooks/useCollaboration.ts | 57 +- src/hooks/useRealTimeAnalytics.ts | 26 +- src/hooks/useRealtimeConnection.ts | 58 ++ src/hooks/useWebSocket.ts | 124 ++- src/lib/graphql/subscriptionQueries.ts | 17 + src/lib/graphql/subscriptions.test.ts | 16 +- src/lib/graphql/subscriptions.ts | 222 ++++- src/lib/monitoring/alerts.ts | 15 + src/lib/monitoring/metrics.ts | 14 + src/lib/notifications/socket.ts | 224 ++--- .../__tests__/connectionSupervisor.test.ts | 617 +++++++++++++ src/lib/realtime/connectionSupervisor.ts | 841 ++++++++++++++++++ src/lib/websocketManager.ts | 263 +++--- src/serviceWorker.ts | 9 + src/store/synchronizationEngine.ts | 12 + 18 files changed, 2139 insertions(+), 412 deletions(-) create mode 100644 src/hooks/useRealtimeConnection.ts create mode 100644 src/lib/realtime/__tests__/connectionSupervisor.test.ts create mode 100644 src/lib/realtime/connectionSupervisor.ts diff --git a/public/sw.js b/public/sw.js index 2386c5d3..60f33d54 100644 --- a/public/sw.js +++ b/public/sw.js @@ -135,4 +135,13 @@ self.addEventListener('fetch', (event) => { return cachedResponse || cache.match(OFFLINE_URL); }) ); -}); \ No newline at end of file +}); +// Realtime transports degraded to offline mode — broadcast to all open clients +// so the app can switch to offline mode. +self.addEventListener('message', (event) => { + if (event.data && event.data.type === 'REALTIME_OFFLINE') { + self.clients.matchAll({ type: 'window', includeUncontrolled: true }).then((clients) => { + clients.forEach((client) => client.postMessage({ type: 'REALTIME_OFFLINE' })); + }); + } +}); diff --git a/src/app/store/messagingStore.ts b/src/app/store/messagingStore.ts index e4b673ff..00fed7dd 100644 --- a/src/app/store/messagingStore.ts +++ b/src/app/store/messagingStore.ts @@ -127,9 +127,9 @@ export const useMessagingStore = create((set, get) => ({ get().addMessage(message); - if (state.socket) { - state.socket.emit('message', message); - } + // Route through the supervisor so messages sent while disconnected are queued + // (bounded, ordered) and flushed on reconnect instead of being silently dropped. + wsManager.send('messaging', 'message', message); get().setTyping(false); }, @@ -139,7 +139,7 @@ export const useMessagingStore = create((set, get) => ({ messages: state.messages.map((msg) => (msg.id === messageId ? { ...msg, read: true } : msg)), })); - get().socket?.emit('read', { messageId }); + wsManager.send('messaging', 'read', { messageId }); }, markConversationAsRead: (conversationId) => { @@ -156,8 +156,8 @@ export const useMessagingStore = create((set, get) => ({ const socket = get().socket; const conversation = get().currentConversation; - if (socket && conversation) { - socket.emit('typing', { + if (conversation) { + wsManager.send('messaging', 'typing', { conversationId: conversation.id, isTyping, }); diff --git a/src/constants/app.constants.ts b/src/constants/app.constants.ts index 2e75f05e..9a64f83f 100644 --- a/src/constants/app.constants.ts +++ b/src/constants/app.constants.ts @@ -20,6 +20,19 @@ export const API_CACHE_TTL_DEFAULT = 300000; // 5 minutes // API URLs & Endpoints export const DEFAULT_SOCKET_URL = 'http://localhost:3001'; +// Realtime connection supervisor (see src/lib/realtime/connectionSupervisor.ts) +export const REALTIME_RECONNECT_BASE_DELAY_MS = 1000; +export const REALTIME_RECONNECT_MAX_DELAY_MS = 30000; +export const REALTIME_RECONNECT_MAX_ATTEMPTS = 5; +/** Jitter factor for reconnect backoff. 1 = full jitter (random 0..2x), 0 = deterministic. */ +export const REALTIME_RECONNECT_JITTER = 1; +export const REALTIME_HEARTBEAT_INTERVAL_MS = 30000; +export const REALTIME_HEARTBEAT_TIMEOUT_MS = 10000; +export const REALTIME_OUTBOUND_QUEUE_LIMIT = 100; +export const REALTIME_QUEUE_POLICY = 'drop-oldest' as const; +/** Message type used to signal clients (via the service worker) that realtime gave up. */ +export const REALTIME_OFFLINE_EVENT = 'REALTIME_OFFLINE'; + // Web3 Config export const DEFAULT_STARKNET_NETWORK = 'goerli-alpha'; export const STARKNET_NETWORKS = { diff --git a/src/hooks/useCollaboration.ts b/src/hooks/useCollaboration.ts index ff79aaef..76441a9c 100644 --- a/src/hooks/useCollaboration.ts +++ b/src/hooks/useCollaboration.ts @@ -2,6 +2,42 @@ import { useEffect, useRef, useState } from 'react'; import * as Y from 'yjs'; import { WebsocketProvider } from 'y-websocket'; import { Awareness } from 'y-protocols/awareness'; +import { + BaseRealtimeTransport, + ConnectionSupervisor, + registerSupervisor, +} from '@/lib/realtime/connectionSupervisor'; +import { useRealtimeConnection } from './useRealtimeConnection'; + +/** + * Bridges the y-websocket provider's lifecycle into the shared connection + * supervisor. y-websocket owns the actual transport and reconnection, so the + * supervisor runs with `manageReconnect: false` and only mirrors the unified + * status for consumers. + */ +class CollaborationTransport extends BaseRealtimeTransport { + readonly name = 'collaboration'; + + constructor(private readonly provider: WebsocketProvider) { + super(); + } + + connect(): void { + // y-websocket owns the socket; nothing to do here. + } + + isOpen(): boolean { + return this.provider.ws?.readyState === WebSocket.OPEN; + } + + bridgeConnected(): void { + this.events.emitOpen(); + } + + bridgeDisconnected(): void { + this.events.emitClose(); + } +} type CursorPosition = { line: number; @@ -51,12 +87,16 @@ export function useCollaboration(roomId: string, user: CollaborationUser, websoc const [whiteboardStrokes, setWhiteboardStrokes] = useState([]); const [error, setError] = useState(null); + const connectionName = `collaboration:${roomId}`; + const connection = useRealtimeConnection(connectionName); + const docRef = useRef(null); const providerRef = useRef(null); const awarenessRef = useRef(null); const yTextRef = useRef(null); const strokesRef = useRef | null>(null); const chatRef = useRef | null>(null); + const supervisorRef = useRef(null); const websocketEndpoint = websocketUrl || @@ -78,6 +118,12 @@ export function useCollaboration(roomId: string, user: CollaborationUser, websoc providerRef.current = provider; awarenessRef.current = provider.awareness; + // Register the unified connection status for this collaboration room. + const transport = new CollaborationTransport(provider); + const supervisor = new ConnectionSupervisor(transport, { manageReconnect: false }); + supervisorRef.current = supervisor; + const unregisterSupervisor = registerSupervisor(connectionName, supervisor); + const updatePresence = () => { const states = Array.from(awarenessRef.current?.getStates().values() ?? []); const nextUsers: CollaborationUser[] = states @@ -121,6 +167,11 @@ export function useCollaboration(roomId: string, user: CollaborationUser, websoc provider.on('status', ({ status: providerStatus }) => { setConnected(providerStatus === 'connected'); setStatus(providerStatus === 'connected' ? 'connected' : 'disconnected'); + if (providerStatus === 'connected') { + transport.bridgeConnected(); + } else { + transport.bridgeDisconnected(); + } }); provider.on('sync', () => { @@ -131,6 +182,9 @@ export function useCollaboration(roomId: string, user: CollaborationUser, websoc return () => { awarenessRef.current?.off('change', updatePresence); + unregisterSupervisor(); + supervisor.disconnect(); + supervisorRef.current = null; provider.disconnect(); doc.destroy(); docRef.current = null; @@ -140,7 +194,7 @@ export function useCollaboration(roomId: string, user: CollaborationUser, websoc strokesRef.current = null; chatRef.current = null; }; - }, [roomId, user.id, websocketEndpoint]); + }, [connectionName, roomId, user.id, websocketEndpoint]); useEffect(() => { const provider = providerRef.current; @@ -214,6 +268,7 @@ export function useCollaboration(roomId: string, user: CollaborationUser, websoc return { connected, status, + connection, editorText, users, messages, diff --git a/src/hooks/useRealTimeAnalytics.ts b/src/hooks/useRealTimeAnalytics.ts index a2be9fe6..57c7b4d8 100644 --- a/src/hooks/useRealTimeAnalytics.ts +++ b/src/hooks/useRealTimeAnalytics.ts @@ -1,4 +1,10 @@ import { useState, useEffect, useCallback } from 'react'; +import { + ConnectionSupervisor, + LocalRealtimeTransport, + registerSupervisor, +} from '@/lib/realtime/connectionSupervisor'; +import { useRealtimeConnection } from './useRealtimeConnection'; export interface AnalyticsDataPoint { timestamp: string; @@ -6,15 +12,20 @@ export interface AnalyticsDataPoint { category?: string; } +const ANALYTICS_CONNECTION = 'real-time-analytics'; + export const useRealTimeAnalytics = (initialData: AnalyticsDataPoint[] = []) => { const [data, setData] = useState(initialData); - const [isConnected, setIsConnected] = useState(false); + const connection = useRealtimeConnection(ANALYTICS_CONNECTION); - // In a real application, this would connect to a real WebSocket endpoint - // For the sake of this frontend implementation, we simulate the WebSocket stream + // The analytics stream is simulated client-side; the supervisor is registered + // with a locally-open transport so consumers observe the same unified + // connection status shape as every other realtime hook. useEffect(() => { - // Simulate WebSocket connection - setIsConnected(true); + const transport = new LocalRealtimeTransport(); + const supervisor = new ConnectionSupervisor(transport, { manageReconnect: false }); + const unregister = registerSupervisor(ANALYTICS_CONNECTION, supervisor); + supervisor.connect(); const interval = setInterval(() => { setData((prevData) => { @@ -32,7 +43,8 @@ export const useRealTimeAnalytics = (initialData: AnalyticsDataPoint[] = []) => return () => { clearInterval(interval); - setIsConnected(false); + unregister(); + supervisor.disconnect(); }; }, []); @@ -40,5 +52,5 @@ export const useRealTimeAnalytics = (initialData: AnalyticsDataPoint[] = []) => setData((prev) => [...prev, point]); }, []); - return { data, isConnected, addDataPoint }; + return { data, isConnected: connection.isConnected, connection, addDataPoint }; }; diff --git a/src/hooks/useRealtimeConnection.ts b/src/hooks/useRealtimeConnection.ts new file mode 100644 index 00000000..781fafd1 --- /dev/null +++ b/src/hooks/useRealtimeConnection.ts @@ -0,0 +1,58 @@ +'use client'; + +import { useEffect, useState } from 'react'; +import { + ConnectionSupervisor, + type ConnectionStatus, + getSupervisor, + onSupervisorRegistered, +} from '@/lib/realtime/connectionSupervisor'; + +const DEFAULT_STATUS: ConnectionStatus = { + phase: 'idle', + isConnected: false, + isReconnecting: false, + reconnectAttempts: 0, + queuedCount: 0, +}; + +/** + * Subscribe to the unified connection status of a named realtime connection + * (e.g. `websocket:messaging`, `notifications`, `graphql-subscriptions`, + * `collaboration:`). Falls back to the idle status while the connection + * has not been registered yet, so consumers see one consistent status shape + * regardless of the underlying transport. + */ +export function useRealtimeConnection(name: string): ConnectionStatus { + const [status, setStatus] = useState( + () => getSupervisor(name)?.getStatus() ?? DEFAULT_STATUS, + ); + + useEffect(() => { + let unsubscribeStatus: (() => void) | undefined; + + const attach = (supervisor: ConnectionSupervisor) => { + setStatus(supervisor.getStatus()); + unsubscribeStatus = supervisor.onStatusChange(setStatus); + }; + + const existing = getSupervisor(name); + if (existing) { + attach(existing); + } + + const removeRegistrationListener = onSupervisorRegistered((registeredName, supervisor) => { + if (registeredName === name) { + unsubscribeStatus?.(); + attach(supervisor); + } + }); + + return () => { + unsubscribeStatus?.(); + removeRegistrationListener(); + }; + }, [name]); + + return status; +} diff --git a/src/hooks/useWebSocket.ts b/src/hooks/useWebSocket.ts index 80e0bada..ef8a010f 100644 --- a/src/hooks/useWebSocket.ts +++ b/src/hooks/useWebSocket.ts @@ -1,4 +1,11 @@ -import { useCallback, useEffect, useRef, useState } from 'react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { + ConnectionSupervisor, + RawWebSocketTransport, + type ConnectionStatus as RealtimeConnectionStatus, + registerSupervisor, +} from '@/lib/realtime/connectionSupervisor'; +import { useRealtimeConnection } from './useRealtimeConnection'; type ConnectionStatus = 'idle' | 'connecting' | 'connected' | 'disconnected' | 'error'; type ConnectionMode = 'websocket' | 'broadcast' | 'disabled'; @@ -19,6 +26,8 @@ interface UseWebSocketResult { mode: ConnectionMode; lastMessage: TMessage | null; sendMessage: (message: TMessage) => void; + /** Unified realtime connection status shared by all realtime hooks. */ + connection: RealtimeConnectionStatus; } export const useWebSocket = ({ @@ -30,13 +39,16 @@ export const useWebSocket = ({ parse, serialize, }: UseWebSocketOptions): UseWebSocketResult => { - const [status, setStatus] = useState(enabled ? 'connecting' : 'idle'); + const connectionName = useMemo(() => (url ? `websocket:${url}` : 'broadcast'), [url]); + const realtimeConnection = useRealtimeConnection(connectionName); + const [mode, setMode] = useState('disabled'); const [lastMessage, setLastMessage] = useState(null); + const [parseFailed, setParseFailed] = useState(false); - const socketRef = useRef(null); + const supervisorRef = useRef(null); + const transportRef = useRef(null); const channelRef = useRef(null); - const reconnectTimerRef = useRef | null>(null); const canUseWindow = typeof window !== 'undefined'; @@ -61,16 +73,6 @@ export const useWebSocket = ({ ); const cleanup = useCallback(() => { - if (reconnectTimerRef.current) { - clearTimeout(reconnectTimerRef.current); - reconnectTimerRef.current = null; - } - - if (socketRef.current) { - socketRef.current.close(); - socketRef.current = null; - } - if (channelRef.current) { channelRef.current.close(); channelRef.current = null; @@ -80,67 +82,45 @@ export const useWebSocket = ({ useEffect(() => { if (!enabled || !canUseWindow) { cleanup(); + supervisorRef.current?.disconnect(); + supervisorRef.current = null; + transportRef.current = null; setMode('disabled'); - setStatus(enabled ? 'connecting' : 'idle'); return; } if (url) { - let cancelled = false; - - const connect = () => { - if (cancelled) { - return; + const transport = new RawWebSocketTransport(url); + transportRef.current = transport; + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: reconnectDelayMs, + }); + supervisorRef.current = supervisor; + const unregister = registerSupervisor(connectionName, supervisor); + + const unsubscribeMessage = transport.onMessage((payload) => { + try { + setLastMessage(safeParse(String(payload))); + setParseFailed(false); + } catch { + setParseFailed(true); } + }); - setMode('websocket'); - setStatus('connecting'); - - const socket = new WebSocket(url); - socketRef.current = socket; - - socket.onopen = () => { - if (!cancelled) { - setStatus('connected'); - } - }; - - socket.onmessage = (event) => { - try { - const message = safeParse(String(event.data)); - setLastMessage(message); - } catch { - setStatus('error'); - } - }; - - socket.onerror = () => { - if (!cancelled) { - setStatus('error'); - } - }; - - socket.onclose = () => { - if (cancelled) { - return; - } - - setStatus('disconnected'); - reconnectTimerRef.current = setTimeout(connect, reconnectDelayMs); - }; - }; - - connect(); + setMode('websocket'); + supervisor.connect(); return () => { - cancelled = true; - cleanup(); + unsubscribeMessage(); + unregister(); + supervisor.disconnect(); + supervisorRef.current = null; + transportRef.current = null; }; } const channelName = localChannelKey ?? `collaboration-room:${roomId ?? 'default'}`; setMode('broadcast'); - setStatus('connected'); const channel = new BroadcastChannel(channelName); channelRef.current = channel; @@ -151,9 +131,20 @@ export const useWebSocket = ({ return () => { cleanup(); - setStatus('disconnected'); }; - }, [canUseWindow, cleanup, enabled, localChannelKey, reconnectDelayMs, roomId, safeParse, url]); + }, [canUseWindow, cleanup, connectionName, enabled, localChannelKey, reconnectDelayMs, roomId, safeParse, url]); + + const status: ConnectionStatus = parseFailed + ? 'error' + : mode === 'broadcast' + ? 'connected' + : realtimeConnection.isConnected + ? 'connected' + : realtimeConnection.phase === 'connecting' || realtimeConnection.phase === 'reconnecting' + ? 'connecting' + : realtimeConnection.phase === 'idle' && enabled + ? 'connecting' + : 'disconnected'; const sendMessage = useCallback( (message: TMessage) => { @@ -161,8 +152,10 @@ export const useWebSocket = ({ return; } - if (mode === 'websocket' && socketRef.current?.readyState === WebSocket.OPEN) { - socketRef.current.send(safeSerialize(message)); + if (mode === 'websocket' && supervisorRef.current) { + // Bounded, ordered outbound queue — messages sent while disconnected are + // buffered and flushed in order on reconnect (drop-oldest on overflow). + supervisorRef.current.send(safeSerialize(message)); return; } @@ -179,5 +172,6 @@ export const useWebSocket = ({ mode, lastMessage, sendMessage, + connection: realtimeConnection, }; }; diff --git a/src/lib/graphql/subscriptionQueries.ts b/src/lib/graphql/subscriptionQueries.ts index 3ed6394f..72012e1c 100644 --- a/src/lib/graphql/subscriptionQueries.ts +++ b/src/lib/graphql/subscriptionQueries.ts @@ -278,3 +278,20 @@ export const PRESENCE_SUBSCRIPTION = gql` } } `; + +/** + * Catch-up query used to backfill events missed while the realtime connection was + * down. Consumed by the connection supervisor's inbound sequence-gap handler after + * a reconnect (`since` = last seen sequence number). + */ +export const REALTIME_CATCHUP_QUERY = gql` + query RealtimeCatchUp($since: ID!) { + realtimeEvents(since: $since) { + id + sequence + type + payload + createdAt + } + } +`; diff --git a/src/lib/graphql/subscriptions.test.ts b/src/lib/graphql/subscriptions.test.ts index f397b56f..65476b91 100644 --- a/src/lib/graphql/subscriptions.test.ts +++ b/src/lib/graphql/subscriptions.test.ts @@ -2,7 +2,9 @@ import { vi, describe, it, expect, beforeEach } from 'vitest'; // ── Mock heavy Apollo/graphql-ws deps before importing the module under test ── vi.mock('@apollo/client/link/subscriptions', () => ({ - GraphQLWsLink: vi.fn().mockImplementation(() => ({ request: vi.fn() })), + GraphQLWsLink: vi.fn().mockImplementation(function () { + return { request: vi.fn() }; + }), })); vi.mock('graphql-ws', () => ({ @@ -10,11 +12,17 @@ vi.mock('graphql-ws', () => ({ })); vi.mock('@apollo/client', () => { - const HttpLink = vi.fn().mockImplementation(() => ({ request: vi.fn() })); + const HttpLink = vi.fn().mockImplementation(function () { + return { request: vi.fn() }; + }); const ApolloLink = { from: vi.fn((links) => links[0]) }; const split = vi.fn((test, ws, http) => ({ _ws: ws, _http: http, _split: true })); - const ApolloClient = vi.fn().mockImplementation((opts) => ({ link: opts.link })); - const InMemoryCache = vi.fn().mockImplementation(() => ({})); + const ApolloClient = vi.fn().mockImplementation(function (opts: { link: unknown }) { + return { link: opts.link }; + }); + const InMemoryCache = vi.fn().mockImplementation(function () { + return {}; + }); return { HttpLink, ApolloLink, split, ApolloClient, InMemoryCache }; }); diff --git a/src/lib/graphql/subscriptions.ts b/src/lib/graphql/subscriptions.ts index 2caf4875..ee5745db 100644 --- a/src/lib/graphql/subscriptions.ts +++ b/src/lib/graphql/subscriptions.ts @@ -1,18 +1,32 @@ /** * GraphQL Subscriptions Configuration * Provides WebSocket-based real-time data updates using Apollo Client and graphql-ws + * + * The graphql-ws socket lifecycle (reconnect, heartbeat, queueing) is delegated to + * the shared `ConnectionSupervisor` (src/lib/realtime/connectionSupervisor.ts): + * graphql-ws only opens the socket lazily, the supervisor schedules reconnects and + * the transport re-subscribes every registered subscription after a reconnect. */ import { GraphQLWsLink } from '@apollo/client/link/subscriptions'; -import { createClient as createWSClient } from 'graphql-ws'; +import { createClient as createWSClient, type Client } from 'graphql-ws'; import { ApolloClient, InMemoryCache, ApolloLink, split, HttpLink } from '@apollo/client'; import { getMainDefinition } from '@apollo/client/utilities'; -import { DocumentNode } from 'graphql'; +import { DocumentNode, print } from 'graphql'; import { flagStore, evaluateFlag } from '@/lib/feature-flags'; import { createLogger } from '@/lib/logging'; +import { + BaseRealtimeTransport, + ConnectionSupervisor, + getSupervisor, + registerSupervisor, +} from '@/lib/realtime/connectionSupervisor'; const logger = createLogger('graphql-subscriptions'); +/** Name under which the GraphQL subscription supervisor is registered. */ +export const GRAPHQL_SUBSCRIPTIONS_CONNECTION = 'graphql-subscriptions'; + /** * WebSocket subscription configuration options */ @@ -205,6 +219,132 @@ export function isFeatureEnabled(flagId: string, context: Record return evaluateFlag(flag, context); } +interface SubscriptionEntry { + query: DocumentNode; + variables?: Record; + handler: (payload: any) => void; +} + +/** + * graphql-ws transport adapter. The socket opens lazily (only when there is at + * least one active subscription); the ConnectionSupervisor drives reconnects and + * this adapter re-subscribes every registered entry after each reconnect. + */ +class GraphQLWsTransport extends BaseRealtimeTransport { + readonly name = 'graphql'; + private client: Client | null = null; + private connected = false; + private readonly subscriptions = new Map(); + private readonly unsubscribes = new Map void>(); + + constructor(private readonly config: SubscriptionConfig) { + super(); + } + + getClient(): Client | null { + return this.client; + } + + connect(): void { + if (!this.client) { + const { reconnect } = { ...DEFAULT_SUBSCRIPTION_CONFIG, ...this.config }; + this.client = createWSClient({ + url: this.config.subscriptionUrl, + connectionParams: () => ({ + authorization: this.config.headers?.authorization ?? '', + }), + // Reconnection is owned by the ConnectionSupervisor. + shouldRetry: () => false, + retryAttempts: 0, + lazy: true, + keepAlive: 10_000, + on: { + connected: () => { + this.connected = true; + this.events.emitOpen(); + }, + error: (error) => { + this.events.emitError(error); + }, + closed: () => { + this.connected = false; + this.events.emitClose(); + }, + connecting: () => { + // Status is driven by the supervisor's own 'connecting' phase. + }, + }, + connectionAckWaitTimeout: this.config.connectionTimeoutMs ?? 5000, + }); + } + // Restore every subscription — opening the socket lazily if needed. + this.resubscribeAll(); + } + + disconnect(): void { + this.close(); + } + + close(): void { + this.connected = false; + this.unsubscribes.forEach((unsubscribe) => unsubscribe()); + this.unsubscribes.clear(); + this.client?.terminate(); + this.client = null; + } + + isOpen(): boolean { + return this.connected; + } + + send(): void { + // Subscriptions are the only outbound channel; nothing to queue here. + } + + sendPing(): void { + // graphql-ws handles protocol-level keep-alive via `keepAlive`. + } + + /** + * Register a subscription. It is (re-)established immediately and again after + * every reconnect driven by the supervisor. + */ + subscribe(id: string, entry: SubscriptionEntry): () => void { + this.subscriptions.set(id, entry); + this.resubscribe(id); + return () => { + this.unsubscribes.get(id)?.(); + this.unsubscribes.delete(id); + this.subscriptions.delete(id); + }; + } + + private resubscribeAll(): void { + this.subscriptions.forEach((_, id) => this.resubscribe(id)); + } + + private resubscribe(id: string): void { + const entry = this.subscriptions.get(id); + const client = this.client; + if (!entry || !client) { + return; + } + this.unsubscribes.get(id)?.(); + const unsubscribe = client.subscribe( + { + query: print(entry.query), + variables: entry.variables ?? {}, + }, + { + next: (result) => entry.handler(result.data), + error: (error) => this.events.emitError(error), + complete: () => undefined, + }, + ); + this.unsubscribes.set(id, unsubscribe); + } +} + /** * Creates a GraphQL subscriptions-enabled Apollo Client */ @@ -226,37 +366,36 @@ export function createSubscriptionClient(config: SubscriptionConfig): ApolloClie // Only build the WebSocket link when the feature is enabled const link: ApolloLink = subscriptionsEnabled ? (() => { - const wsClient = createWSClient({ - url: config.subscriptionUrl, - connectionParams: () => ({ - authorization: config.headers?.authorization ?? '', - }), - shouldRetry: (code) => { - return code !== 1000 && code !== 1001 && code !== 4000; - }, - retryAttempts: config.reconnect?.maxRetries ?? 5, - on: { - connected: () => { - manager.setState(ConnectionState.CONNECTED); - manager.resetRetryCount(); - }, - error: (error) => { - const normalizedError = - error instanceof Error - ? error - : new Error(typeof error === 'string' ? error : 'Unknown error'); - manager.setState(ConnectionState.ERROR, normalizedError); - }, - closed: () => { - manager.setState(ConnectionState.DISCONNECTED); - }, - connecting: () => { - manager.setState(ConnectionState.CONNECTING); - }, - }, - connectionAckWaitTimeout: config.connectionTimeoutMs ?? 5000, + const transport = new GraphQLWsTransport(config); + const { reconnect } = { ...DEFAULT_SUBSCRIPTION_CONFIG, ...config }; + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: reconnect?.initialDelayMs ?? 1000, + maxReconnectDelayMs: reconnect?.maxDelayMs ?? 30000, + maxReconnectAttempts: reconnect?.maxRetries ?? 5, + }); + + // Mirror the supervisor status into the legacy connection manager so + // existing `getConnectionManager()` consumers keep working. + supervisor.onStatusChange((status) => { + if (status.isConnected) { + manager.setState(ConnectionState.CONNECTED); + manager.resetRetryCount(); + } else if (status.phase === 'connecting' || status.phase === 'reconnecting') { + manager.setState(ConnectionState.CONNECTING); + } else if (status.phase === 'offline') { + manager.setState( + ConnectionState.ERROR, + new Error(status.lastError ?? 'Realtime connection unavailable'), + ); + } else { + manager.setState(ConnectionState.DISCONNECTED); + } }); + registerSupervisor(GRAPHQL_SUBSCRIPTIONS_CONNECTION, supervisor); + supervisor.connect(); + + const wsClient = transport.getClient()!; const wsLink = new GraphQLWsLink(wsClient); return split( @@ -289,6 +428,27 @@ export function createSubscriptionClient(config: SubscriptionConfig): ApolloClie return client; } +/** + * Subscribe to a realtime event through the GraphQL supervisor. The subscription + * is automatically restored after every reconnect (resubscribe registry). + * + * @returns unsubscribe function + */ +export function subscribeRealtime( + id: string, + query: DocumentNode, + variables: Record, + handler: (payload: any) => void, +): () => void { + const supervisor = getSupervisor(GRAPHQL_SUBSCRIPTIONS_CONNECTION); + const transport = supervisor?.getTransport() as GraphQLWsTransport | undefined; + if (!supervisor || !transport) { + logger.warn('[GraphQLSubscriptions] No active subscription supervisor; subscription dropped'); + return () => undefined; + } + return transport.subscribe(id, { query, variables, handler }); +} + /** * Get the current connection manager singleton */ diff --git a/src/lib/monitoring/alerts.ts b/src/lib/monitoring/alerts.ts index 2d31dd24..0704020c 100644 --- a/src/lib/monitoring/alerts.ts +++ b/src/lib/monitoring/alerts.ts @@ -57,6 +57,21 @@ export function checkAlerts(metrics: Metric[]): Alert[] { severity: 'low', }); } + + // Realtime transport reliability alerts (see src/lib/realtime/connectionSupervisor.ts) + if (m.name === 'realtime_offline' && m.value > 0) { + alerts.push({ + message: 'Realtime connection failed repeatedly — degraded to offline mode', + severity: 'high', + }); + } + + if (m.name === 'heartbeat_timeout' && m.value > 0) { + alerts.push({ + message: 'Realtime heartbeat timeout detected', + severity: 'low', + }); + } }); return alerts; diff --git a/src/lib/monitoring/metrics.ts b/src/lib/monitoring/metrics.ts index 9e6898a6..08e0868f 100644 --- a/src/lib/monitoring/metrics.ts +++ b/src/lib/monitoring/metrics.ts @@ -1,8 +1,22 @@ import { useEffect, useState } from 'react'; import { LocalMonitoringProvider, Metric } from './provider'; +import { createCounterMetric } from '@/lib/logging/performance'; const provider = new LocalMonitoringProvider(); +/** + * Record a realtime reliability metric (e.g. `reconnect_attempt`, `reconnect_success`, + * `heartbeat_timeout`, `queue_dropped`, `realtime_offline`). Metrics are surfaced + * through `useMetrics()` and evaluated by `checkAlerts()`. + */ +export function recordRealtimeMetric( + name: string, + value: number = 1, + tags?: Record, +): void { + createCounterMetric(name, value, tags); +} + export function useMetrics() { const [metrics, setMetrics] = useState([]); diff --git a/src/lib/notifications/socket.ts b/src/lib/notifications/socket.ts index 4cf008d1..6342aee5 100644 --- a/src/lib/notifications/socket.ts +++ b/src/lib/notifications/socket.ts @@ -1,5 +1,11 @@ import type { BaseNotification, NotificationEvent } from './types'; import { createLogger } from '@/lib/logging'; +import { + ConnectionSupervisor, + RawWebSocketTransport, + type ConnectionStatus, + registerSupervisor, +} from '@/lib/realtime/connectionSupervisor'; const logger = createLogger('notification-socket'); @@ -39,27 +45,26 @@ const DEFAULT_MAX_RECONNECT_DELAY = 30_000; const DEFAULT_MAX_RECONNECT_ATTEMPTS = 0; const DEFAULT_RECONNECT_JITTER = 0.2; +/** + * Notification WebSocket service. The raw socket I/O lives in + * `RawWebSocketTransport`; the connect/reconnect/heartbeat/queue lifecycle is + * delegated to a `ConnectionSupervisor` so notifications share the same unified + * status shape, jittered backoff and bounded outbound queue as every other + * realtime transport in the app. + */ export class NotificationSocketService { - private ws: WebSocket | null = null; + private readonly transport: RawWebSocketTransport; + private readonly supervisor: ConnectionSupervisor; private readonly listeners = new Set(); private readonly connectionListeners = new Set(); - private reconnectTimer: ReturnType | null = null; - private reconnectDelay: number; - private readonly initialReconnectDelay: number; - private readonly maxReconnectDelay: number; - private readonly maxReconnectAttempts: number; - private readonly reconnectJitter: number; private intentionallyClosed = false; - private reconnectAttempts = 0; private connectionState: NotificationSocketConnectionState = { status: 'idle', reconnectAttempts: 0, }; - private readonly outboundQueue: OutboundMessage[] = []; private readonly handleOnline = () => { - if (!this.intentionallyClosed && !this.isOpen()) { - this.clearReconnectTimer(); - this.open(); + if (!this.intentionallyClosed) { + this.supervisor.reconnectNow(); } }; private readonly handleVisibilityChange = () => { @@ -68,32 +73,31 @@ export class NotificationSocketService { } }; - constructor(private readonly url: string, options: NotificationSocketOptions = {}) { - this.initialReconnectDelay = options.initialReconnectDelay ?? DEFAULT_INITIAL_RECONNECT_DELAY; - this.maxReconnectDelay = options.maxReconnectDelay ?? DEFAULT_MAX_RECONNECT_DELAY; - this.maxReconnectAttempts = options.maxReconnectAttempts ?? DEFAULT_MAX_RECONNECT_ATTEMPTS; - this.reconnectJitter = options.reconnectJitter ?? DEFAULT_RECONNECT_JITTER; - this.reconnectDelay = this.initialReconnectDelay; + constructor(url: string, options: NotificationSocketOptions = {}) { + this.transport = new RawWebSocketTransport(url); + this.supervisor = new ConnectionSupervisor(this.transport, { + initialReconnectDelayMs: options.initialReconnectDelay ?? DEFAULT_INITIAL_RECONNECT_DELAY, + maxReconnectDelayMs: options.maxReconnectDelay ?? DEFAULT_MAX_RECONNECT_DELAY, + maxReconnectAttempts: options.maxReconnectAttempts ?? DEFAULT_MAX_RECONNECT_ATTEMPTS, + reconnectJitter: options.reconnectJitter ?? DEFAULT_RECONNECT_JITTER, + }); + this.transport.onMessage((payload) => this.handleIncoming(payload)); + this.supervisor.onStatusChange((status) => { + this.updateConnectionState(this.mapSupervisorStatus(status)); + }); + registerSupervisor('notifications', this.supervisor); } connect(): void { this.intentionallyClosed = false; this.registerNetworkListeners(); - this.open(); + this.supervisor.connect(); } disconnect(): void { this.intentionallyClosed = true; this.unregisterNetworkListeners(); - this.clearReconnectTimer(); - this.ws?.close(); - this.ws = null; - this.outboundQueue.length = 0; - this.updateConnectionState({ - status: 'disconnected', - reconnectAttempts: 0, - lastError: undefined, - }); + this.supervisor.disconnect(); } subscribe(listener: NotificationListener): () => void { @@ -111,149 +115,53 @@ export class NotificationSocketService { return this.connectionState; } + /** Send an outbound message through the supervisor's bounded, ordered queue. */ send(event: NotificationEventType, payload: unknown): void { const message: OutboundMessage = { event, payload }; - - if (this.isOpen()) { - this.ws?.send(JSON.stringify({ event, payload })); - return; - } - - this.outboundQueue.push(message); + this.supervisor.send(message); } - private open(): void { - if (this.ws || this.intentionallyClosed) { - return; - } - - if (this.hasReachedMaxAttempts()) { - this.updateConnectionState({ - status: 'disconnected', - reconnectAttempts: this.reconnectAttempts, - lastError: `Max reconnection attempts (${this.maxReconnectAttempts}) reached`, - }); - return; - } - + private handleIncoming(payload: unknown): void { try { - this.updateConnectionState({ - status: this.reconnectAttempts > 0 ? 'reconnecting' : 'connecting', - reconnectAttempts: this.reconnectAttempts, - }); - - this.ws = new WebSocket(this.url); + const data = JSON.parse(payload as string) as NotificationEvent; + if (data.event === 'notification') { + const notification = data.payload as BaseNotification; + notification.timestamp = new Date(notification.timestamp); + this.listeners.forEach((listener) => listener(notification)); + } + } catch { + logger.warn('[NotificationSocket] Failed to parse message', { data: payload }); + } + } - this.ws.onopen = () => { - this.reconnectDelay = this.initialReconnectDelay; - this.reconnectAttempts = 0; - this.updateConnectionState({ + private mapSupervisorStatus(status: ConnectionStatus): NotificationSocketConnectionState { + switch (status.phase) { + case 'idle': + return { status: 'idle', reconnectAttempts: status.reconnectAttempts }; + case 'connecting': + return { + status: status.reconnectAttempts > 0 ? 'reconnecting' : 'connecting', + reconnectAttempts: status.reconnectAttempts, + }; + case 'reconnecting': + return { status: 'reconnecting', reconnectAttempts: status.reconnectAttempts }; + case 'connected': + return { status: 'connected', reconnectAttempts: 0, - lastConnectedAt: new Date(), + lastConnectedAt: status.lastConnectedAt, lastError: undefined, - }); - this.flushOutboundQueue(); - }; - - this.ws.onmessage = (event: MessageEvent) => { - try { - const data = JSON.parse(event.data as string) as NotificationEvent; - if (data.event === 'notification') { - const notification = data.payload as BaseNotification; - notification.timestamp = new Date(notification.timestamp); - this.listeners.forEach((listener) => listener(notification)); - } - } catch { - logger.warn('[NotificationSocket] Failed to parse message', { data: event.data }); - } - }; - - this.ws.onclose = () => { - this.ws = null; - if (!this.intentionallyClosed) { - this.updateConnectionState({ - status: 'disconnected', - reconnectAttempts: this.reconnectAttempts, - }); - this.scheduleReconnect(); - } - }; - - this.ws.onerror = () => { - this.updateConnectionState({ + }; + case 'disconnected': + case 'offline': + return { status: 'disconnected', - reconnectAttempts: this.reconnectAttempts, - lastError: 'WebSocket connection error', - }); - this.ws?.close(); - }; - } catch (error) { - const message = error instanceof Error ? error.message : 'Failed to open WebSocket'; - logger.error('[NotificationSocket] Failed to open', { error }); - this.updateConnectionState({ - status: 'disconnected', - reconnectAttempts: this.reconnectAttempts, - lastError: message, - }); - this.scheduleReconnect(); + reconnectAttempts: status.reconnectAttempts, + lastError: status.lastError, + }; } } - private scheduleReconnect(): void { - if (this.intentionallyClosed || this.reconnectTimer || this.hasReachedMaxAttempts()) { - return; - } - - this.reconnectAttempts += 1; - const delay = this.getReconnectDelay(); - - this.updateConnectionState({ - status: 'reconnecting', - reconnectAttempts: this.reconnectAttempts, - }); - - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - this.open(); - this.reconnectDelay = Math.min(this.reconnectDelay * 2, this.maxReconnectDelay); - }, delay); - } - - private getReconnectDelay(): number { - const jitterRange = this.reconnectDelay * this.reconnectJitter; - const jitter = (Math.random() * 2 - 1) * jitterRange; - return Math.max(0, Math.round(this.reconnectDelay + jitter)); - } - - private clearReconnectTimer(): void { - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - } - } - - private flushOutboundQueue(): void { - if (!this.isOpen() || this.outboundQueue.length === 0) { - return; - } - - const queuedMessages = [...this.outboundQueue]; - this.outboundQueue.length = 0; - - queuedMessages.forEach(({ event, payload }) => { - this.ws?.send(JSON.stringify({ event, payload })); - }); - } - - private isOpen(): boolean { - return this.ws?.readyState === WebSocket.OPEN; - } - - private hasReachedMaxAttempts(): boolean { - return this.maxReconnectAttempts > 0 && this.reconnectAttempts >= this.maxReconnectAttempts; - } - private updateConnectionState(updates: Partial): void { this.connectionState = { ...this.connectionState, diff --git a/src/lib/realtime/__tests__/connectionSupervisor.test.ts b/src/lib/realtime/__tests__/connectionSupervisor.test.ts new file mode 100644 index 00000000..535af956 --- /dev/null +++ b/src/lib/realtime/__tests__/connectionSupervisor.test.ts @@ -0,0 +1,617 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + ConnectionSupervisor, + type RealtimeTransport, + getSupervisor, + onAnyReconnect, + onSupervisorRegistered, + registerSupervisor, +} from '../connectionSupervisor'; +import { getRecordedMetrics } from '@/lib/logging/performance'; + +type Handler = (...args: any[]) => void; + +class FakeTransport implements RealtimeTransport { + readonly name = 'fake'; + sent: unknown[] = []; + pingCount = 0; + connectCount = 0; + private opened = false; + private openHandlers = new Set(); + private closeHandlers = new Set(); + private errorHandlers = new Set(); + private messageHandlers = new Set(); + private pongHandlers = new Set(); + + connect(): void { + this.connectCount += 1; + this.opened = false; + } + + disconnect(): void { + this.opened = false; + } + + close(): void { + this.opened = false; + this.closeHandlers.forEach((handler) => handler()); + } + + isOpen(): boolean { + return this.opened; + } + + send(payload: unknown): void { + this.sent.push(payload); + } + + sendPing(): void { + this.pingCount += 1; + } + + simulateOpen(): void { + this.opened = true; + this.openHandlers.forEach((handler) => handler()); + } + + simulateClose(): void { + this.opened = false; + this.closeHandlers.forEach((handler) => handler()); + } + + simulateMessage(payload: unknown): void { + this.messageHandlers.forEach((handler) => handler(payload)); + } + + simulatePong(): void { + this.pongHandlers.forEach((handler) => handler()); + } + + onOpen(handler: Handler): () => void { + this.openHandlers.add(handler); + return () => this.openHandlers.delete(handler); + } + + onClose(handler: Handler): () => void { + this.closeHandlers.add(handler); + return () => this.closeHandlers.delete(handler); + } + + onError(handler: Handler): () => void { + this.errorHandlers.add(handler); + return () => this.errorHandlers.delete(handler); + } + + onMessage(handler: Handler): () => void { + this.messageHandlers.add(handler); + return () => this.messageHandlers.delete(handler); + } + + onPong(handler: Handler): () => void { + this.pongHandlers.add(handler); + return () => this.pongHandlers.delete(handler); + } +} + +const hasMetric = (name: string): boolean => + getRecordedMetrics(200).some((metric) => metric.name === name); + +describe('ConnectionSupervisor', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); + // Reset the global metric store so assertions don't leak across tests. + delete (globalThis as { __TEACHLINK_METRICS__?: unknown }).__TEACHLINK_METRICS__; + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + describe('connectivity lifecycle', () => { + it('connects and reports a unified connected status', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + + supervisor.connect(); + expect(supervisor.getStatus().phase).toBe('connecting'); + + transport.simulateOpen(); + const status = supervisor.getStatus(); + expect(status.phase).toBe('connected'); + expect(status.isConnected).toBe(true); + expect(status.isReconnecting).toBe(false); + expect(status.reconnectAttempts).toBe(0); + expect(status.lastConnectedAt).toBeInstanceOf(Date); + }); + + it('notifies status listeners of every change', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + const phases: string[] = []; + supervisor.onStatusChange((status) => phases.push(status.phase)); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateClose(); + + expect(phases).toEqual(['connecting', 'connected', 'disconnected', 'reconnecting']); + }); + + it('is idempotent while connecting or connected', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + + supervisor.connect(); + supervisor.connect(); + expect(transport.connectCount).toBe(1); + + transport.simulateOpen(); + supervisor.connect(); + expect(transport.connectCount).toBe(1); + }); + + it('disconnect is intentional and never reconnects', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + initialReconnectDelayMs: 100, + }); + + supervisor.connect(); + transport.simulateOpen(); + supervisor.disconnect(); + + expect(supervisor.getStatus().phase).toBe('disconnected'); + expect(supervisor.getStatus().queuedCount).toBe(0); + vi.advanceTimersByTime(10_000); + expect(transport.connectCount).toBe(1); + }); + }); + + describe('exponential backoff with jitter', () => { + it('reconnects with deterministic exponential backoff when jitter is 0', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: 100, + reconnectJitter: 0, + maxReconnectDelayMs: 10_000, + maxReconnectAttempts: 0, + }); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateClose(); + + expect(supervisor.getStatus().phase).toBe('reconnecting'); + expect(supervisor.getStatus().reconnectAttempts).toBe(1); + + // First backoff: 100ms + vi.advanceTimersByTime(99); + expect(transport.connectCount).toBe(1); + vi.advanceTimersByTime(1); + expect(transport.connectCount).toBe(2); + + transport.simulateClose(); + // Second backoff: 200ms + vi.advanceTimersByTime(199); + expect(transport.connectCount).toBe(2); + vi.advanceTimersByTime(1); + expect(transport.connectCount).toBe(3); + expect(supervisor.getStatus().reconnectAttempts).toBe(2); + }); + + it('applies full jitter range to the backoff delay', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: 100, + reconnectJitter: 1, + maxReconnectDelayMs: 10_000, + maxReconnectAttempts: 0, + }); + + supervisor.connect(); + transport.simulateOpen(); + + // Full jitter → delay ∈ [0, 2*base]. With Math.random() = 0.5, delay = 100. + transport.simulateClose(); + vi.advanceTimersByTime(100); + expect(transport.connectCount).toBe(2); + + transport.simulateOpen(); + // Math.random() = 1 → delay = 2*100 = 200 (full jitter upper bound) + vi.mocked(Math.random).mockReturnValue(1); + transport.simulateClose(); + vi.advanceTimersByTime(199); + expect(transport.connectCount).toBe(2); + vi.advanceTimersByTime(1); + expect(transport.connectCount).toBe(3); + }); + + it('caps the backoff delay at maxReconnectDelayMs', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: 100, + reconnectJitter: 0, + maxReconnectDelayMs: 250, + maxReconnectAttempts: 0, + }); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateClose(); + vi.advanceTimersByTime(100); + transport.simulateClose(); + vi.advanceTimersByTime(200); + transport.simulateClose(); + // 4th attempt would be 800ms raw → capped at 250ms + vi.advanceTimersByTime(250); + expect(transport.connectCount).toBe(4); + }); + + it('emits reconnect_attempt and reconnect_success metrics', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: 100, + reconnectJitter: 0, + }); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateClose(); + expect(hasMetric('reconnect_attempt')).toBe(true); + + vi.advanceTimersByTime(100); + transport.simulateOpen(); + expect(hasMetric('reconnect_success')).toBe(true); + }); + }); + + describe('heartbeat with ping-timeout detection', () => { + it('sends pings on the heartbeat interval and treats a pong as alive', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + heartbeatIntervalMs: 100, + heartbeatTimeoutMs: 50, + }); + + supervisor.connect(); + transport.simulateOpen(); + + vi.advanceTimersByTime(100); + expect(transport.pingCount).toBe(1); + + transport.simulatePong(); + // Pong received for the first ping → still alive at the next check. + vi.advanceTimersByTime(49); + expect(transport.connectCount).toBe(1); + expect(supervisor.getStatus().phase).toBe('connected'); + + // Answer the second ping as well → still no timeout. + vi.advanceTimersByTime(51); + expect(transport.pingCount).toBe(2); + transport.simulatePong(); + vi.advanceTimersByTime(49); + expect(supervisor.getStatus().phase).toBe('connected'); + }); + + it('forces a reconnect when the peer stops answering pings', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + initialReconnectDelayMs: 100, + heartbeatIntervalMs: 100, + heartbeatTimeoutMs: 50, + }); + + supervisor.connect(); + transport.simulateOpen(); + + vi.advanceTimersByTime(100); // ping sent + expect(transport.pingCount).toBe(1); + + vi.advanceTimersByTime(50); // no pong within timeout → close → reconnect scheduled + expect(supervisor.getStatus().phase).toBe('reconnecting'); + + vi.advanceTimersByTime(100); + expect(transport.connectCount).toBe(2); + expect(hasMetric('heartbeat_timeout')).toBe(true); + }); + }); + + describe('bounded outbound queue with backpressure', () => { + it('queues messages while disconnected and flushes them in order on connect', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + queueLimit: 100, + }); + + supervisor.send('first'); + supervisor.send('second'); + expect(supervisor.getStatus().queuedCount).toBe(2); + + supervisor.connect(); + transport.simulateOpen(); + + expect(transport.sent).toEqual(['first', 'second']); + expect(supervisor.getStatus().queuedCount).toBe(0); + }); + + it('drops the oldest message once the queue limit is reached (drop-oldest)', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + queueLimit: 3, + queuePolicy: 'drop-oldest', + }); + + supervisor.send('a'); + supervisor.send('b'); + supervisor.send('c'); + supervisor.send('d'); // 'a' dropped + + expect(supervisor.getStatus().queuedCount).toBe(3); + supervisor.connect(); + transport.simulateOpen(); + expect(transport.sent).toEqual(['b', 'c', 'd']); + expect(hasMetric('queue_dropped')).toBe(true); + }); + + it('blocks (drops the new message) under the block policy', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + queueLimit: 2, + queuePolicy: 'block', + }); + + supervisor.send('a'); + supervisor.send('b'); + supervisor.send('c'); // blocked — queue unchanged + + expect(supervisor.getStatus().queuedCount).toBe(2); + supervisor.connect(); + transport.simulateOpen(); + expect(transport.sent).toEqual(['a', 'b']); + }); + + it('sends immediately when connected', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + + supervisor.connect(); + transport.simulateOpen(); + supervisor.send('live'); + expect(transport.sent).toEqual(['live']); + expect(supervisor.getStatus().queuedCount).toBe(0); + }); + }); + + describe('inbound sequence tracking and catch-up', () => { + it('requests catch-up when an inbound sequence gap is detected', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + const catchUp = vi.fn(); + supervisor.setCatchUpHandler(catchUp); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateMessage({ sequence: 1 }); + expect(catchUp).not.toHaveBeenCalled(); + + transport.simulateMessage({ sequence: 3 }); + expect(catchUp).toHaveBeenCalledTimes(1); + expect(supervisor.getStatus().lastSequence).toBe(3); + expect(supervisor.getLastSequence()).toBe(3); + }); + + it('tracks sequences with no gap', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + const catchUp = vi.fn(); + supervisor.setCatchUpHandler(catchUp); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateMessage({ sequence: 5 }); + transport.simulateMessage({ sequence: 6 }); + expect(catchUp).not.toHaveBeenCalled(); + expect(supervisor.getStatus().lastSequence).toBe(6); + }); + }); + + describe('resubscribe registry', () => { + it('restores registered subscriptions after every reconnect', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + initialReconnectDelayMs: 100, + }); + const restored: string[] = []; + + supervisor.registerResubscribe('room:general', () => restored.push('room:general')); + supervisor.registerResubscribe('sub:notifications', () => restored.push('sub:notifications')); + + supervisor.connect(); + transport.simulateOpen(); + expect(restored).toEqual(['room:general', 'sub:notifications']); + + transport.simulateClose(); + vi.advanceTimersByTime(100); + transport.simulateOpen(); + expect(restored).toEqual(['room:general', 'sub:notifications', 'room:general', 'sub:notifications']); + }); + + it('honours unsubscribe and stops restoring that key', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + const restored: string[] = []; + + const unsubscribe = supervisor.registerResubscribe('room:general', () => restored.push('room:general')); + unsubscribe(); + + supervisor.connect(); + transport.simulateOpen(); + expect(restored).toEqual([]); + }); + + it('invokes onReconnect callbacks after reconnects', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + initialReconnectDelayMs: 100, + }); + const onReconnect = vi.fn(); + supervisor.onReconnect(onReconnect); + + supervisor.connect(); + transport.simulateOpen(); + expect(onReconnect).toHaveBeenCalledTimes(1); + + transport.simulateClose(); + vi.advanceTimersByTime(100); + transport.simulateOpen(); + expect(onReconnect).toHaveBeenCalledTimes(2); + }); + }); + + describe('offline degradation', () => { + it('gives up after max attempts, emits realtime_offline and signals the service worker', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: 100, + reconnectJitter: 0, + maxReconnectAttempts: 2, + }); + const postMessage = vi.fn(); + (navigator.serviceWorker as { controller: unknown }).controller = { postMessage }; + + supervisor.connect(); + transport.simulateClose(); // attempt 1 + vi.advanceTimersByTime(100); // attempt 2 (connect) + transport.simulateClose(); // attempt 2 fails → attempt 2 retry scheduled + vi.advanceTimersByTime(200); // retry timer fires → give up + + const status = supervisor.getStatus(); + expect(status.phase).toBe('offline'); + expect(status.isConnected).toBe(false); + expect(status.lastError).toContain('Max reconnection attempts (2) reached'); + expect(hasMetric('realtime_offline')).toBe(true); + expect(postMessage).toHaveBeenCalledWith({ type: 'REALTIME_OFFLINE' }); + }); + + it('does not emit a reconnect_success metric for a first connect', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + + supervisor.connect(); + transport.simulateOpen(); + const metrics = getRecordedMetrics(200).filter((m) => m.name === 'reconnect_success'); + expect(metrics).toHaveLength(0); + }); + }); + + describe('reconnectNow (browser online)', () => { + it('reconnects immediately, skipping the pending backoff timer', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: 5_000, + reconnectJitter: 0, + }); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateClose(); + expect(transport.connectCount).toBe(1); + + supervisor.reconnectNow(); + expect(transport.connectCount).toBe(2); + expect(supervisor.getStatus().phase).toBe('connecting'); + }); + + it('is a no-op when connected or intentionally closed', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + + supervisor.connect(); + transport.simulateOpen(); + supervisor.reconnectNow(); + expect(transport.connectCount).toBe(1); + + supervisor.disconnect(); + supervisor.reconnectNow(); + expect(transport.connectCount).toBe(1); + }); + }); + + describe('manageReconnect = false', () => { + it('mirrors status but never schedules reconnects itself', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + initialReconnectDelayMs: 100, + manageReconnect: false, + }); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateClose(); + + expect(supervisor.getStatus().phase).toBe('disconnected'); + vi.advanceTimersByTime(10_000); + expect(transport.connectCount).toBe(1); + }); + }); +}); + +describe('supervisor registry', () => { + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it('registers, looks up and unregisters supervisors by name', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + const unregister = registerSupervisor('test:registry', supervisor); + + expect(getSupervisor('test:registry')).toBe(supervisor); + unregister(); + expect(getSupervisor('test:registry')).toBeUndefined(); + }); + + it('replays registered supervisors to late subscribers', () => { + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { reconnectJitter: 0 }); + registerSupervisor('test:replay', supervisor); + + const seen: string[] = []; + onSupervisorRegistered((name) => seen.push(name)); + expect(seen).toContain('test:replay'); + }); + + it('fires onAnyReconnect for supervisors registered after subscribing', () => { + vi.useFakeTimers(); + const callback = vi.fn(); + const unsubscribe = onAnyReconnect(callback); + + const transport = new FakeTransport(); + const supervisor = new ConnectionSupervisor(transport, { + reconnectJitter: 0, + initialReconnectDelayMs: 100, + }); + registerSupervisor('test:any-reconnect', supervisor); + + supervisor.connect(); + transport.simulateOpen(); + transport.simulateClose(); + vi.advanceTimersByTime(100); + transport.simulateOpen(); + + expect(callback).toHaveBeenCalledTimes(2); + unsubscribe(); + }); +}); diff --git a/src/lib/realtime/connectionSupervisor.ts b/src/lib/realtime/connectionSupervisor.ts new file mode 100644 index 00000000..803661be --- /dev/null +++ b/src/lib/realtime/connectionSupervisor.ts @@ -0,0 +1,841 @@ +/** + * ConnectionSupervisor — transport-agnostic realtime connection lifecycle. + * + * All realtime transports in the app (socket.io via `src/lib/websocketManager.ts`, + * the hand-rolled notification WebSocket in `src/lib/notifications/socket.ts` and + * GraphQL subscriptions in `src/lib/graphql/subscriptions.ts`) delegate their + * connect/reconnect/heartbeat lifecycle to this supervisor so every consumer sees + * one unified `ConnectionStatus` shape and identical reliability behaviour. + * + * Delivery guarantees (outbound): + * - `send()` while connected delivers immediately through the transport. + * - `send()` while disconnected is queued (bounded by `queueLimit`) and flushed + * in FIFO order on the next successful connect. + * - Overflow is bounded by the queue policy: + * - `'drop-oldest'` (default): the oldest queued message is dropped to make room, + * a `queue_dropped` metric is emitted. + * - `'block'`: the new message is dropped, a `queue_dropped` metric is emitted. + * + * Delivery guarantees (inbound): + * - Messages carrying a numeric `sequence` envelope are tracked; a gap between the + * last seen sequence and the incoming sequence triggers the registered catch-up + * handler (used to backfill events missed while the transport was down). + * + * Reconnection: + * - Exponential backoff with configurable jitter (full jitter by default). + * - A shared heartbeat (ping / pong with ping-timeout detection) triggers an + * immediate reconnect when the peer stops answering. + * - Rooms/subscriptions registered via `registerResubscribe()` are automatically + * restored after every (re)connect, and `onReconnect()` callbacks run afterwards + * so the synchronization engine can backfill the gap window. + * - Once `maxReconnectAttempts` is exceeded the supervisor degrades to `offline` + * mode: it emits a `realtime_offline` metric, raises a reconnect-failure alert + * (see `src/lib/monitoring/alerts.ts`) and signals the service worker so the app + * can switch to offline mode. + */ + +import { createLogger } from '@/lib/logging'; +import { createCounterMetric } from '@/lib/logging/performance'; +import { + REALTIME_HEARTBEAT_INTERVAL_MS, + REALTIME_HEARTBEAT_TIMEOUT_MS, + REALTIME_OFFLINE_EVENT, + REALTIME_OUTBOUND_QUEUE_LIMIT, + REALTIME_QUEUE_POLICY, + REALTIME_RECONNECT_BASE_DELAY_MS, + REALTIME_RECONNECT_JITTER, + REALTIME_RECONNECT_MAX_ATTEMPTS, + REALTIME_RECONNECT_MAX_DELAY_MS, +} from '@/constants/app.constants'; + +const logger = createLogger('connection-supervisor'); + +export type ConnectionPhase = + | 'idle' + | 'connecting' + | 'connected' + | 'reconnecting' + | 'disconnected' + | 'offline'; + +/** Unified connection status shared by every realtime consumer hook. */ +export interface ConnectionStatus { + phase: ConnectionPhase; + isConnected: boolean; + isReconnecting: boolean; + reconnectAttempts: number; + lastConnectedAt?: Date; + /** @deprecated Alias of `lastConnectedAt`, kept for backwards compatibility. */ + lastConnected?: Date; + lastError?: string; + /** Number of messages currently buffered by the outbound queue. */ + queuedCount: number; + /** Highest inbound sequence observed (set when the transport sends sequenced envelopes). */ + lastSequence?: number; +} + +export type QueuePolicy = 'drop-oldest' | 'block'; + +export interface ConnectionSupervisorOptions { + /** Base delay for the first reconnect attempt (ms). Default 1000. */ + initialReconnectDelayMs?: number; + /** Ceiling for the exponential backoff delay (ms). Default 30000. */ + maxReconnectDelayMs?: number; + /** Max reconnect attempts before degrading to offline. 0 = retry forever. Default 5. */ + maxReconnectAttempts?: number; + /** Jitter factor 0..1 applied to the backoff delay. 1 = full jitter. Default 1. */ + reconnectJitter?: number; + /** How often to ping the peer (ms). Default 30000. */ + heartbeatIntervalMs?: number; + /** Max time without a pong before forcing a reconnect (ms). Default 10000. */ + heartbeatTimeoutMs?: number; + /** Bound of the outbound queue. Default 100. */ + queueLimit?: number; + /** Overflow policy for a full outbound queue. Default 'drop-oldest'. */ + queuePolicy?: QueuePolicy; + /** + * When false the transport owns reconnection (e.g. y-websocket); the supervisor + * only mirrors status, heartbeat and queueing. Default true. + */ + manageReconnect?: boolean; +} + +/** + * A transport-agnostic socket abstraction. Transports implement this interface and + * feed lifecycle events back to the supervisor through the `on*` hooks. + */ +export interface RealtimeTransport { + readonly name: string; + connect(): void; + disconnect(): void; + close(): void; + isOpen(): boolean; + /** Deliver an outbound payload (already serialized by the caller if needed). */ + send(payload: unknown): void; + /** Send a protocol-level heartbeat ping. */ + sendPing(): void; + onOpen(handler: () => void): () => void; + onClose(handler: (reason?: string) => void): () => void; + onError(handler: (error: unknown) => void): () => void; + onMessage(handler: (payload: unknown) => void): () => void; + onPong(handler: () => void): () => void; +} + +/** Minimal event bus used by transport implementations to fan out lifecycle events. */ +export class TransportEventBus { + private readonly openHandlers = new Set<() => void>(); + private readonly closeHandlers = new Set<(reason?: string) => void>(); + private readonly errorHandlers = new Set<(error: unknown) => void>(); + private readonly messageHandlers = new Set<(payload: unknown) => void>(); + private readonly pongHandlers = new Set<() => void>(); + + onOpen(handler: () => void): () => void { + this.openHandlers.add(handler); + return () => this.openHandlers.delete(handler); + } + + onClose(handler: (reason?: string) => void): () => void { + this.closeHandlers.add(handler); + return () => this.closeHandlers.delete(handler); + } + + onError(handler: (error: unknown) => void): () => void { + this.errorHandlers.add(handler); + return () => this.errorHandlers.delete(handler); + } + + onMessage(handler: (payload: unknown) => void): () => void { + this.messageHandlers.add(handler); + return () => this.messageHandlers.delete(handler); + } + + onPong(handler: () => void): () => void { + this.pongHandlers.add(handler); + return () => this.pongHandlers.delete(handler); + } + + emitOpen(): void { + this.openHandlers.forEach((handler) => { + try { + handler(); + } catch (error) { + logger.warn('[TransportEventBus] open handler failed', { error }); + } + }); + } + + emitClose(reason?: string): void { + this.closeHandlers.forEach((handler) => { + try { + handler(reason); + } catch (error) { + logger.warn('[TransportEventBus] close handler failed', { error }); + } + }); + } + + emitError(error: unknown): void { + this.errorHandlers.forEach((handler) => { + try { + handler(error); + } catch (handlerError) { + logger.warn('[TransportEventBus] error handler failed', { error: handlerError }); + } + }); + } + + emitMessage(payload: unknown): void { + this.messageHandlers.forEach((handler) => { + try { + handler(payload); + } catch (error) { + logger.warn('[TransportEventBus] message handler failed', { error }); + } + }); + } + + emitPong(): void { + this.pongHandlers.forEach((handler) => { + try { + handler(); + } catch (error) { + logger.warn('[TransportEventBus] pong handler failed', { error }); + } + }); + } +} + +/** Base class providing the event plumbing shared by transport implementations. */ +export abstract class BaseRealtimeTransport implements RealtimeTransport { + protected readonly events = new TransportEventBus(); + + abstract readonly name: string; + abstract connect(): void; + + disconnect(): void { + this.close(); + } + + close(): void { + // no-op by default + } + + isOpen(): boolean { + return false; + } + + send(_payload: unknown): void { + // no-op by default + } + + sendPing(): void { + // no-op by default + } + + onOpen(handler: () => void): () => void { + return this.events.onOpen(handler); + } + + onClose(handler: (reason?: string) => void): () => void { + return this.events.onClose(handler); + } + + onError(handler: (error: unknown) => void): () => void { + return this.events.onError(handler); + } + + onMessage(handler: (payload: unknown) => void): () => void { + return this.events.onMessage(handler); + } + + onPong(handler: () => void): () => void { + return this.events.onPong(handler); + } +} + +/** Raw browser `WebSocket` transport (used by notifications + generic websocket hook). */ +export class RawWebSocketTransport extends BaseRealtimeTransport { + readonly name = 'websocket'; + private ws: WebSocket | null = null; + + constructor(private readonly url: string) { + super(); + } + + connect(): void { + if (this.ws && (this.ws.readyState === WebSocket.OPEN || this.ws.readyState === WebSocket.CONNECTING)) { + return; + } + + const ws = new WebSocket(this.url); + this.ws = ws; + + ws.onopen = () => this.events.emitOpen(); + ws.onmessage = (event) => { + this.events.emitMessage(event.data); + // Detect protocol-level pongs so the supervisor can reset its heartbeat. + if (typeof event.data === 'string') { + try { + const parsed = JSON.parse(event.data) as { type?: string }; + if (parsed && parsed.type === 'pong') { + this.events.emitPong(); + } + } catch { + // not JSON — ignore + } + } + }; + ws.onerror = () => this.events.emitError('WebSocket connection error'); + ws.onclose = () => { + if (this.ws === ws) { + this.ws = null; + } + this.events.emitClose(); + }; + } + + close(): void { + this.ws?.close(); + } + + disconnect(): void { + this.close(); + this.ws = null; + } + + isOpen(): boolean { + return this.ws?.readyState === WebSocket.OPEN; + } + + send(payload: unknown): void { + if (this.isOpen()) { + const data = typeof payload === 'string' ? payload : JSON.stringify(payload); + this.ws?.send(data); + } + } + + sendPing(): void { + if (this.isOpen()) { + this.ws?.send(JSON.stringify({ type: 'ping' })); + } + } +} + +/** Locally-open transport used to surface a unified status for simulated streams. */ +export class LocalRealtimeTransport extends BaseRealtimeTransport { + readonly name = 'local'; + private opened = false; + + connect(): void { + if (this.opened) return; + this.opened = true; + this.events.emitOpen(); + } + + close(): void { + this.opened = false; + } + + isOpen(): boolean { + return this.opened; + } +} + +export class ConnectionSupervisor { + private status: ConnectionStatus; + private reconnectTimer: ReturnType | null = null; + private heartbeatTimer: ReturnType | null = null; + private heartbeatCheckTimer: ReturnType | null = null; + private reconnectDelayMs: number; + private lastPingSentAt = 0; + private lastPongAt = 0; + private intentionallyClosed = false; + private outboundQueue: unknown[] = []; + private readonly statusListeners = new Set<(status: ConnectionStatus) => void>(); + private readonly resubscribeRegistry = new Map void>(); + private readonly reconnectCallbacks = new Set<() => void>(); + private catchUpHandler: (() => void) | null = null; + private readonly transportUnsubscribers: Array<() => void> = []; + + constructor( + private readonly transport: RealtimeTransport, + private readonly options: ConnectionSupervisorOptions = {}, + ) { + this.reconnectDelayMs = options.initialReconnectDelayMs ?? REALTIME_RECONNECT_BASE_DELAY_MS; + this.status = this.createInitialStatus(); + this.transportUnsubscribers.push( + transport.onOpen(() => this.handleOpen()), + transport.onClose((reason) => this.handleClose(reason)), + transport.onError((error) => this.handleError(error)), + transport.onMessage((payload) => this.handleMessage(payload)), + transport.onPong(() => this.handlePong()), + ); + } + + connect(): void { + this.intentionallyClosed = false; + if (this.status.phase === 'connecting' || this.status.phase === 'connected') { + return; + } + this.attemptConnect(); + } + + disconnect(): void { + this.intentionallyClosed = true; + this.clearReconnectTimer(); + this.stopHeartbeat(); + this.transport.disconnect(); + this.outboundQueue.length = 0; + this.setStatus({ + phase: 'disconnected', + isConnected: false, + isReconnecting: false, + reconnectAttempts: 0, + lastError: undefined, + queuedCount: 0, + }); + } + + /** + * Publish a message with bounded backpressure. See module docs for the exact + * delivery guarantees (immediate when connected, bounded FIFO queue otherwise, + * `drop-oldest` or `block` overflow policy). + */ + send(payload: unknown): void { + if (this.status.isConnected && this.transport.isOpen()) { + this.transport.send(payload); + return; + } + + const limit = this.options.queueLimit ?? REALTIME_OUTBOUND_QUEUE_LIMIT; + const policy = this.options.queuePolicy ?? REALTIME_QUEUE_POLICY; + + if (this.outboundQueue.length >= limit) { + if (policy === 'drop-oldest') { + this.outboundQueue.shift(); + } + createCounterMetric('queue_dropped', 1, { transport: this.transport.name, policy }); + if (policy === 'block') { + return; + } + } + + this.outboundQueue.push(payload); + this.setStatus({ queuedCount: this.outboundQueue.length }); + } + + getStatus(): ConnectionStatus { + return this.status; + } + + getTransport(): RealtimeTransport { + return this.transport; + } + + onStatusChange(listener: (status: ConnectionStatus) => void): () => void { + this.statusListeners.add(listener); + return () => this.statusListeners.delete(listener); + } + + /** + * Register a callback that is re-run after every successful (re)connect so that + * socket rooms, GraphQL subscriptions etc. are automatically restored. + */ + registerResubscribe(key: string, resubscribe: () => void): () => void { + this.resubscribeRegistry.set(key, resubscribe); + return () => { + this.resubscribeRegistry.delete(key); + }; + } + + /** + * Register a callback invoked after each successful (re)connect. Used by the + * synchronization engine to backfill state missed during the reconnect gap. + */ + onReconnect(callback: () => void): () => void { + this.reconnectCallbacks.add(callback); + return () => this.reconnectCallbacks.delete(callback); + } + + /** Register (or clear) the handler invoked when an inbound sequence gap is detected. */ + setCatchUpHandler(handler: (() => void) | null): void { + this.catchUpHandler = handler; + } + + getLastSequence(): number | undefined { + return this.status.lastSequence; + } + + /** Immediately attempt a reconnect, skipping any pending backoff timer. */ + reconnectNow(): void { + if (this.intentionallyClosed || this.status.isConnected) { + return; + } + this.clearReconnectTimer(); + this.attemptConnect(); + } + + private createInitialStatus(): ConnectionStatus { + return { + phase: 'idle', + isConnected: false, + isReconnecting: false, + reconnectAttempts: 0, + queuedCount: 0, + }; + } + + private attemptConnect(): void { + if (this.intentionallyClosed) { + return; + } + this.setStatus({ + phase: 'connecting', + isConnected: false, + isReconnecting: this.status.reconnectAttempts > 0, + }); + try { + this.transport.connect(); + } catch (error) { + this.handleError(error); + } + } + + private handleOpen(): void { + const afterAttempts = this.status.reconnectAttempts; + this.reconnectDelayMs = this.options.initialReconnectDelayMs ?? REALTIME_RECONNECT_BASE_DELAY_MS; + this.lastPingSentAt = 0; + this.lastPongAt = Date.now(); + + this.setStatus({ + phase: 'connected', + isConnected: true, + isReconnecting: false, + reconnectAttempts: 0, + lastConnectedAt: new Date(), + lastConnected: new Date(), + lastError: undefined, + queuedCount: this.outboundQueue.length, + }); + + if (afterAttempts > 0) { + createCounterMetric('reconnect_success', 1, { transport: this.transport.name, afterAttempts }); + } + + this.flushQueue(); + this.startHeartbeat(); + this.runResubscribeRegistry(); + this.runReconnectCallbacks(); + } + + private handleClose(reason?: string): void { + if (this.intentionallyClosed) { + return; + } + this.stopHeartbeat(); + this.setStatus({ + phase: 'disconnected', + isConnected: false, + lastError: reason ? `Disconnected: ${reason}` : undefined, + }); + this.scheduleReconnect(reason); + } + + private handleError(error: unknown): void { + if (this.intentionallyClosed) { + return; + } + const message = + error instanceof Error + ? error.message + : typeof error === 'string' + ? error + : 'Unknown connection error'; + this.setStatus({ + phase: 'disconnected', + isConnected: false, + lastError: message, + }); + this.scheduleReconnect(message); + } + + private handleMessage(payload: unknown): void { + const message = this.normalizeMessage(payload); + if (!message || typeof message !== 'object') { + return; + } + + const envelope = message as { sequence?: unknown; type?: unknown }; + if (typeof envelope.sequence === 'number') { + const last = this.status.lastSequence; + if (last !== undefined && envelope.sequence > last + 1) { + logger.debug('[ConnectionSupervisor] Inbound sequence gap detected', { + transport: this.transport.name, + from: last, + to: envelope.sequence, + }); + try { + this.catchUpHandler?.(); + } catch (error) { + logger.warn('[ConnectionSupervisor] Catch-up handler failed', { error }); + } + } + this.setStatus({ lastSequence: envelope.sequence }); + } + } + + private handlePong(): void { + this.lastPongAt = Date.now(); + } + + private scheduleReconnect(reason?: string): void { + if (this.intentionallyClosed || this.reconnectTimer) { + return; + } + + const manageReconnect = this.options.manageReconnect ?? true; + if (!manageReconnect) { + // Transport owns reconnection (e.g. y-websocket) — supervisor only mirrors status. + return; + } + + const maxAttempts = this.options.maxReconnectAttempts ?? REALTIME_RECONNECT_MAX_ATTEMPTS; + if (maxAttempts > 0 && this.status.reconnectAttempts >= maxAttempts) { + this.giveUp(); + return; + } + + const attempt = this.status.reconnectAttempts + 1; + this.setStatus({ + phase: 'reconnecting', + isReconnecting: true, + reconnectAttempts: attempt, + }); + createCounterMetric('reconnect_attempt', 1, { transport: this.transport.name, attempt }); + + const delay = this.computeBackoffDelay(attempt); + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + const attempts = this.status.reconnectAttempts; + if (maxAttempts > 0 && attempts >= maxAttempts) { + this.giveUp(); + return; + } + this.attemptConnect(); + }, delay); + } + + private computeBackoffDelay(attempt: number): number { + const base = this.options.initialReconnectDelayMs ?? REALTIME_RECONNECT_BASE_DELAY_MS; + const max = this.options.maxReconnectDelayMs ?? REALTIME_RECONNECT_MAX_DELAY_MS; + const jitter = this.options.reconnectJitter ?? REALTIME_RECONNECT_JITTER; + const exponential = Math.min(base * Math.pow(2, attempt - 1), max); + if (jitter <= 0) { + return exponential; + } + const low = exponential * (1 - jitter); + const high = exponential * (1 + jitter); + const jittered = low + Math.random() * (high - low); + return Math.min(Math.max(Math.round(jittered), 0), max); + } + + private giveUp(): void { + this.stopHeartbeat(); + const maxAttempts = this.options.maxReconnectAttempts ?? REALTIME_RECONNECT_MAX_ATTEMPTS; + this.setStatus({ + phase: 'offline', + isConnected: false, + isReconnecting: false, + lastError: `Max reconnection attempts (${maxAttempts}) reached`, + }); + createCounterMetric('realtime_offline', 1, { transport: this.transport.name }); + this.signalOfflineMode(); + } + + private signalOfflineMode(): void { + if (typeof navigator === 'undefined' || !('serviceWorker' in navigator)) { + return; + } + try { + navigator.serviceWorker.controller?.postMessage({ type: REALTIME_OFFLINE_EVENT }); + } catch (error) { + logger.warn('[ConnectionSupervisor] Failed to signal offline mode to service worker', { error }); + } + } + + private flushQueue(): void { + if (this.outboundQueue.length === 0 || !this.transport.isOpen()) { + return; + } + const queued = [...this.outboundQueue]; + this.outboundQueue.length = 0; + this.setStatus({ queuedCount: 0 }); + queued.forEach((payload) => { + try { + this.transport.send(payload); + } catch (error) { + logger.warn('[ConnectionSupervisor] Failed to flush queued message', { error }); + } + }); + } + + private startHeartbeat(): void { + this.stopHeartbeat(); + const intervalMs = this.options.heartbeatIntervalMs ?? REALTIME_HEARTBEAT_INTERVAL_MS; + const timeoutMs = this.options.heartbeatTimeoutMs ?? REALTIME_HEARTBEAT_TIMEOUT_MS; + + this.lastPingSentAt = 0; + this.lastPongAt = Date.now(); + + this.heartbeatTimer = setInterval(() => { + if (!this.status.isConnected) { + return; + } + this.lastPingSentAt = Date.now(); + this.transport.sendPing(); + }, intervalMs); + + this.heartbeatCheckTimer = setInterval(() => { + if (!this.status.isConnected || this.lastPingSentAt === 0) { + return; + } + if (this.lastPongAt < this.lastPingSentAt && Date.now() - this.lastPingSentAt >= timeoutMs) { + logger.warn('[ConnectionSupervisor] Heartbeat timeout detected', { + transport: this.transport.name, + }); + createCounterMetric('heartbeat_timeout', 1, { transport: this.transport.name }); + this.transport.close(); + } + }, timeoutMs); + } + + private stopHeartbeat(): void { + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + if (this.heartbeatCheckTimer) { + clearInterval(this.heartbeatCheckTimer); + this.heartbeatCheckTimer = null; + } + this.lastPingSentAt = 0; + } + + private runResubscribeRegistry(): void { + this.resubscribeRegistry.forEach((resubscribe, key) => { + try { + resubscribe(); + } catch (error) { + logger.warn('[ConnectionSupervisor] Resubscribe failed', { key, error }); + } + }); + } + + private runReconnectCallbacks(): void { + this.reconnectCallbacks.forEach((callback) => { + try { + callback(); + } catch (error) { + logger.warn('[ConnectionSupervisor] onReconnect callback failed', { error }); + } + }); + } + + private clearReconnectTimer(): void { + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + } + + private normalizeMessage(payload: unknown): unknown { + if (typeof payload === 'string') { + try { + return JSON.parse(payload); + } catch { + return payload; + } + } + return payload; + } + + private setStatus(updates: Partial): void { + this.status = { ...this.status, ...updates }; + this.statusListeners.forEach((listener) => { + try { + listener(this.status); + } catch (error) { + logger.warn('[ConnectionSupervisor] Status listener failed', { error }); + } + }); + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// Global registry of named supervisors (one per transport connection). +// Consumers (hooks, synchronization engine) look supervisors up by name so they +// all observe the same unified status without knowing the underlying transport. +// ───────────────────────────────────────────────────────────────────────────── + +const supervisors = new Map(); +const registrationListeners = new Set<(name: string, supervisor: ConnectionSupervisor) => void>(); +const reconnectListeners = new Set<() => void>(); + +export function registerSupervisor(name: string, supervisor: ConnectionSupervisor): () => void { + supervisors.set(name, supervisor); + registrationListeners.forEach((listener) => { + try { + listener(name, supervisor); + } catch (error) { + logger.warn('[ConnectionSupervisor] Registration listener failed', { error }); + } + }); + return () => { + if (supervisors.get(name) === supervisor) { + supervisors.delete(name); + } + }; +} + +export function getSupervisor(name: string): ConnectionSupervisor | undefined { + return supervisors.get(name); +} + +/** + * Subscribe to supervisor registrations. Replays currently registered supervisors + * immediately so late subscribers don't miss already-active connections. + */ +export function onSupervisorRegistered( + listener: (name: string, supervisor: ConnectionSupervisor) => void, +): () => void { + registrationListeners.add(listener); + supervisors.forEach((supervisor, name) => { + try { + listener(name, supervisor); + } catch (error) { + logger.warn('[ConnectionSupervisor] Registration replay failed', { error }); + } + }); + return () => registrationListeners.delete(listener); +} + +/** + * Subscribe to reconnects across every registered supervisor (including ones + * registered after this call). Used by the synchronization engine to backfill + * state after a reconnect gap. + */ +export function onAnyReconnect(callback: () => void): () => void { + reconnectListeners.add(callback); + const subscriptions: Array<() => void> = []; + const attach = (supervisor: ConnectionSupervisor) => { + subscriptions.push(supervisor.onReconnect(callback)); + }; + supervisors.forEach(attach); + const removeRegistrationListener = onSupervisorRegistered((_name, supervisor) => attach(supervisor)); + + return () => { + reconnectListeners.delete(callback); + removeRegistrationListener(); + subscriptions.forEach((unsubscribe) => unsubscribe()); + }; +} diff --git a/src/lib/websocketManager.ts b/src/lib/websocketManager.ts index b96f32ac..fe92bcf7 100644 --- a/src/lib/websocketManager.ts +++ b/src/lib/websocketManager.ts @@ -1,6 +1,12 @@ 'use client'; import { io, Socket } from 'socket.io-client'; +import { + BaseRealtimeTransport, + ConnectionSupervisor, + type ConnectionStatus, + registerSupervisor, +} from '@/lib/realtime/connectionSupervisor'; export interface WebSocketConfig { url: string; @@ -11,193 +17,172 @@ export interface WebSocketConfig { timeout?: number; } -export interface ConnectionStatus { - isConnected: boolean; - isReconnecting: boolean; - reconnectAttempts: number; - lastConnected?: Date; - lastError?: string; -} +export type { ConnectionStatus }; -export class WebSocketManager { - private static instance: WebSocketManager; - private connections: Map = new Map(); - private configs: Map = new Map(); - private statuses: Map = new Map(); - private heartbeatIntervals: Map = new Map(); - private reconnectTimeouts: Map = new Map(); +/** + * socket.io transport adapter. Reconnection is disabled at the socket.io level + * (`reconnection: false`) so the ConnectionSupervisor owns the reconnect loop, + * heartbeat and outbound queue; this adapter only maps socket.io events onto + * the transport lifecycle hooks. + */ +class SocketIoTransport extends BaseRealtimeTransport { + readonly name = 'socket.io'; + private socket: Socket | null = null; - private constructor() {} + constructor(private readonly config: WebSocketConfig) { + super(); + } - static getInstance(): WebSocketManager { - if (!WebSocketManager.instance) { - WebSocketManager.instance = new WebSocketManager(); - } - return WebSocketManager.instance; + getSocket(): Socket | null { + return this.socket; } - connect(key: string, config: WebSocketConfig): Socket { - if (this.connections.has(key)) { - return this.connections.get(key)!; + connect(): void { + if (this.socket?.connected) { + return; } + this.socket?.removeAllListeners(); + this.socket?.disconnect(); - const socket = io(config.url + (config.namespace || ''), { + const socket = io(this.config.url + (this.config.namespace || ''), { reconnection: false, - timeout: config.timeout || 20000, + timeout: this.config.timeout || 20000, forceNew: true, }); + this.socket = socket; - this.connections.set(key, socket); - this.configs.set(key, config); - this.setupSocketListeners(key, socket, config); - this.startHeartbeat(key, config); + socket.on('connect', () => this.events.emitOpen()); + socket.on('disconnect', (reason: string) => this.events.emitClose(reason)); + socket.on('connect_error', (error: Error) => this.events.emitError(error)); + socket.on('pong', () => this.events.emitPong()); socket.connect(); - return socket; } - disconnect(key: string): void { - const socket = this.connections.get(key); - if (socket) { - socket.disconnect(); - this.connections.delete(key); - } - this.cleanup(key); + disconnect(): void { + this.socket?.removeAllListeners(); + this.socket?.disconnect(); + this.socket = null; } - getStatus(key: string): ConnectionStatus { - return ( - this.statuses.get(key) || { - isConnected: false, - isReconnecting: false, - reconnectAttempts: 0, - } - ); + close(): void { + this.socket?.disconnect(); } - getSocket(key: string): Socket | null { - return this.connections.get(key) || null; + isOpen(): boolean { + return this.socket?.connected ?? false; } - getAllStatuses(): Record { - const result: Record = {}; - this.statuses.forEach((status, key) => { - result[key] = status; - }); - return result; + send(payload: unknown): void { + const { event, payload: data } = payload as { event: string; payload?: unknown }; + this.socket?.emit(event, data); } - private setupSocketListeners(key: string, socket: Socket, config: WebSocketConfig): void { - socket.on('connect', () => { - this.updateStatus(key, { - isConnected: true, - isReconnecting: false, - reconnectAttempts: 0, - lastConnected: new Date(), - lastError: undefined, - }); - }); + sendPing(): void { + this.socket?.emit('ping'); + } +} - socket.on('disconnect', (reason) => { - this.updateStatus(key, { - isConnected: false, - isReconnecting: false, - reconnectAttempts: this.getStatus(key).reconnectAttempts, - lastError: `Disconnected: ${reason}`, - }); +export class WebSocketManager { + private static instance: WebSocketManager; + private supervisors: Map = new Map(); + private transports: Map = new Map(); + private configs: Map = new Map(); - if (reason !== 'io client disconnect') { - this.scheduleReconnect(key, config); - } - }); + private constructor() {} - socket.on('connect_error', (error) => { - this.updateStatus(key, { - isConnected: false, - isReconnecting: false, - reconnectAttempts: this.getStatus(key).reconnectAttempts, - lastError: error.message, - }); + static getInstance(): WebSocketManager { + if (!WebSocketManager.instance) { + WebSocketManager.instance = new WebSocketManager(); + } + return WebSocketManager.instance; + } - this.scheduleReconnect(key, config); - }); + connect(key: string, config: WebSocketConfig): Socket { + const existing = this.supervisors.get(key); + if (existing) { + return this.transports.get(key)!.getSocket()!; + } - socket.on('pong', () => { - const currentStatus = this.getStatus(key); - if (currentStatus.lastError) { - this.updateStatus(key, { ...currentStatus, lastError: undefined }); - } + const transport = new SocketIoTransport(config); + const supervisor = new ConnectionSupervisor(transport, { + initialReconnectDelayMs: config.reconnectionDelay ?? 1000, + maxReconnectAttempts: config.reconnectionAttempts ?? 5, + maxReconnectDelayMs: (config.reconnectionDelay ?? 1000) * 32, + heartbeatIntervalMs: config.heartbeatInterval ?? 30000, }); - } - private updateStatus(key: string, updates: Partial): void { - const currentStatus = this.getStatus(key); - this.statuses.set(key, { ...currentStatus, ...updates }); - } + this.supervisors.set(key, supervisor); + this.transports.set(key, transport); + this.configs.set(key, config); + registerSupervisor(`websocket:${key}`, supervisor); - private scheduleReconnect(key: string, config: WebSocketConfig): void { - const status = this.getStatus(key); - const maxAttempts = config.reconnectionAttempts || 5; + supervisor.connect(); + return transport.getSocket()!; + } - if (status.reconnectAttempts >= maxAttempts) { - this.updateStatus(key, { - isReconnecting: false, - lastError: `Max reconnection attempts (${maxAttempts}) reached`, - }); + /** + * Send an event through the supervisor's bounded outbound queue. Messages sent + * while disconnected are buffered (FIFO) and flushed on reconnect; overflow + * follows the supervisor's drop-oldest policy. + */ + send(key: string, event: string, payload?: unknown): void { + const supervisor = this.supervisors.get(key); + if (!supervisor) { return; } + supervisor.send({ event, payload }); + } - this.updateStatus(key, { - isReconnecting: true, - reconnectAttempts: status.reconnectAttempts + 1, + /** Join a socket.io room and automatically re-join it after every reconnect. */ + joinRoom(key: string, room: string): () => void { + const supervisor = this.supervisors.get(key); + const socket = this.getSocket(key); + if (!supervisor || !socket) { + return () => undefined; + } + socket.emit('join', { room }); + return supervisor.registerResubscribe(`room:${room}`, () => { + this.getSocket(key)?.emit('join', { room }); }); - - const delay = (config.reconnectionDelay || 1000) * Math.pow(2, status.reconnectAttempts); - const timeout = setTimeout(() => { - this.attemptReconnect(key); - }, delay); - - this.reconnectTimeouts.set(key, timeout); } - private attemptReconnect(key: string): void { - const socket = this.connections.get(key); - if (socket && !socket.connected) { - socket.connect(); + disconnect(key: string): void { + const supervisor = this.supervisors.get(key); + if (supervisor) { + supervisor.disconnect(); + this.supervisors.delete(key); } + this.transports.delete(key); + this.configs.delete(key); } - private startHeartbeat(key: string, config: WebSocketConfig): void { - const interval = config.heartbeatInterval || 30000; - const heartbeat = setInterval(() => { - const socket = this.connections.get(key); - if (socket && socket.connected) { - socket.emit('ping'); + getStatus(key: string): ConnectionStatus { + return ( + this.supervisors.get(key)?.getStatus() || { + phase: 'idle', + isConnected: false, + isReconnecting: false, + reconnectAttempts: 0, + queuedCount: 0, } - }, interval); - - this.heartbeatIntervals.set(key, heartbeat); + ); } - private cleanup(key: string): void { - const heartbeat = this.heartbeatIntervals.get(key); - if (heartbeat) { - clearInterval(heartbeat); - this.heartbeatIntervals.delete(key); - } - - const timeout = this.reconnectTimeouts.get(key); - if (timeout) { - clearTimeout(timeout); - this.reconnectTimeouts.delete(key); - } + getSocket(key: string): Socket | null { + return this.transports.get(key)?.getSocket() || null; + } - this.configs.delete(key); - this.statuses.delete(key); + getAllStatuses(): Record { + const result: Record = {}; + this.supervisors.forEach((supervisor, key) => { + result[key] = supervisor.getStatus(); + }); + return result; } disconnectAll(): void { - this.connections.forEach((_, key) => { + this.supervisors.forEach((_, key) => { this.disconnect(key); }); } diff --git a/src/serviceWorker.ts b/src/serviceWorker.ts index 0519670c..4a65d41f 100644 --- a/src/serviceWorker.ts +++ b/src/serviceWorker.ts @@ -1,5 +1,6 @@ /// import { clientsClaim } from 'workbox-core'; +import { REALTIME_OFFLINE_EVENT } from '@/constants/app.constants'; import { ExpirationPlugin } from 'workbox-expiration'; import { precacheAndRoute, createHandlerBoundToURL } from 'workbox-precaching'; import { registerRoute } from 'workbox-routing'; @@ -222,6 +223,14 @@ self.addEventListener('message', (event) => { if (event.data && event.data.type === 'SKIP_WAITING') { self.skipWaiting(); } + + // Realtime transports degraded to offline mode (max reconnect attempts exceeded) — + // broadcast to every open client so the app can switch to offline mode. + if (event.data && event.data.type === REALTIME_OFFLINE_EVENT) { + self.clients.matchAll({ type: 'window', includeUncontrolled: true }).then((clients) => { + clients.forEach((client) => client.postMessage({ type: REALTIME_OFFLINE_EVENT })); + }); + } }); self.addEventListener('activate', (event) => { diff --git a/src/store/synchronizationEngine.ts b/src/store/synchronizationEngine.ts index 728a0af9..facfea3f 100644 --- a/src/store/synchronizationEngine.ts +++ b/src/store/synchronizationEngine.ts @@ -1,5 +1,6 @@ import { useStore } from './stateManager'; import { createLogger } from '@/lib/logging'; +import { onAnyReconnect } from '@/lib/realtime/connectionSupervisor'; const logger = createLogger('synchronization-engine'); @@ -23,11 +24,20 @@ const SYNC_KEYS = ['user', 'preferences', 'offlineMode', 'lastSynced'] as const; export class SynchronizationEngine { private channel: BroadcastChannel | null = null; private isProcessingSync = false; + private unsubscribeReconnect: (() => void) | null = null; constructor() { if (typeof window !== 'undefined' && 'BroadcastChannel' in window) { this.channel = new BroadcastChannel(CHANNEL_NAME); this.setupListeners(); + + // Catch-up: after any realtime transport reconnects, re-broadcast the + // current state so other tabs converge on updates that may have been + // missed while the transport was down (reconnect gap recovery). + this.unsubscribeReconnect = onAnyReconnect(() => { + logger.debug('[SyncEngine] Realtime reconnected — broadcasting state for catch-up'); + this.broadcastState(useStore.getState()); + }); } } @@ -74,6 +84,8 @@ export class SynchronizationEngine { } public disconnect() { + this.unsubscribeReconnect?.(); + this.unsubscribeReconnect = null; if (this.channel) { this.channel.close(); this.channel = null;