feat(realtime): resilient realtime transport — unified reconnection, heartbeat and backpressure

public/sw.js

@@ -135,4 +135,13 @@ self.addEventListener('fetch', (event) => {
         return cachedResponse || cache.match(OFFLINE_URL);
       })
   );
-});
+});
+// 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' }));
+    });
+  }
+});

src/app/store/messagingStore.ts

@@ -127,9 +127,9 @@ export const useMessagingStore = create<MessagingState>((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<MessagingState>((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<MessagingState>((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,
       });

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 = {

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<WhiteboardStroke[]>([]);
   const [error, setError] = useState<string | null>(null);
 
+  const connectionName = `collaboration:${roomId}`;
+  const connection = useRealtimeConnection(connectionName);
+
   const docRef = useRef<Y.Doc | null>(null);
   const providerRef = useRef<WebsocketProvider | null>(null);
   const awarenessRef = useRef<Awareness | null>(null);
   const yTextRef = useRef<Y.Text | null>(null);
   const strokesRef = useRef<Y.Array<WhiteboardStroke> | null>(null);
   const chatRef = useRef<Y.Array<CollaborationMessage> | null>(null);
+  const supervisorRef = useRef<ConnectionSupervisor | null>(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,

src/hooks/useRealTimeAnalytics.ts

@@ -1,20 +1,31 @@
 import { useState, useEffect, useCallback } from 'react';
+import {
+  ConnectionSupervisor,
+  LocalRealtimeTransport,
+  registerSupervisor,
+} from '@/lib/realtime/connectionSupervisor';
+import { useRealtimeConnection } from './useRealtimeConnection';
 
 export interface AnalyticsDataPoint {
   timestamp: string;
   value: number;
   category?: string;
 }
 
+const ANALYTICS_CONNECTION = 'real-time-analytics';
+
 export const useRealTimeAnalytics = (initialData: AnalyticsDataPoint[] = []) => {
   const [data, setData] = useState<AnalyticsDataPoint[]>(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,13 +43,14 @@ export const useRealTimeAnalytics = (initialData: AnalyticsDataPoint[] = []) =>
 
     return () => {
       clearInterval(interval);
-      setIsConnected(false);
+      unregister();
+      supervisor.disconnect();
     };
   }, []);
 
   const addDataPoint = useCallback((point: AnalyticsDataPoint) => {
     setData((prev) => [...prev, point]);
   }, []);
 
-  return { data, isConnected, addDataPoint };
+  return { data, isConnected: connection.isConnected, connection, addDataPoint };
 };

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:<roomId>`). 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<ConnectionStatus>(
+    () => 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;
+}