feat(sync): recover missed events after realtime reconnect (#1175)

src/lib/graphql/subscriptions.test.ts

@@ -18,12 +18,30 @@ vi.mock('@apollo/client', () => {
   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(function (opts: { link: unknown }) {
-    return { link: opts.link };
+    return {
+      link: opts.link,
+      query: vi.fn().mockResolvedValue({
+        data: {
+          realtimeEvents: [
+            {
+              id: 'evt-1',
+              sequence: 1,
+              type: 'state.updated',
+              payload: { user: { id: 'u-1' } },
+              createdAt: '2026-01-01T00:00:00Z',
+            },
+          ],
+        },
+      }),
+    };
   });
   const InMemoryCache = vi.fn().mockImplementation(function () {
     return {};
   });
-  return { HttpLink, ApolloLink, split, ApolloClient, InMemoryCache };
+  const gql = vi.fn((strings: TemplateStringsArray, ..._values: unknown[]) =>
+    (Array.from(strings) as Array<string | unknown>).join(' '),
+  );
+  return { HttpLink, ApolloLink, split, ApolloClient, InMemoryCache, gql };
 });
 
 vi.mock('@apollo/client/utilities', () => ({
@@ -42,6 +60,8 @@ import {
   GRAPHQL_SUBSCRIPTIONS_CONNECTION,
   getActiveSubscriptions,
   getActiveSubscriptionCount,
+  onSubscriptionCatchUp,
+  requestRealtimeCatchUp,
 } from './subscriptions';
 
 const BASE_CONFIG: SubscriptionConfig = {
@@ -412,3 +432,45 @@ describe('createSubscriptionClient', () => {
     expect(createWSClient).not.toHaveBeenCalled();
   });
 });
+
+// ── Realtime catch-up (issue #1175) ─────────────────────────────────────────
+
+describe('realtime catch-up', () => {
+  it('notifies catch-up listeners when an inbound sequence gap is detected', () => {
+    createSubscriptionClient(BASE_CONFIG);
+    const supervisor = getSupervisor(GRAPHQL_SUBSCRIPTIONS_CONNECTION) as any;
+    const listener = vi.fn();
+    const unsubscribe = onSubscriptionCatchUp(listener);
+
+    supervisor.handleMessage({ sequence: 2 });
+    // Gap: next observed sequence jumps past the last seen one.
+    supervisor.handleMessage({ sequence: 5 });
+
+    expect(listener).toHaveBeenCalledTimes(1);
+    // `since` reflects the last applied sequence before the gap.
+    expect(listener).toHaveBeenCalledWith(2);
+
+    unsubscribe();
+    supervisor.handleMessage({ sequence: 7 });
+    expect(listener).toHaveBeenCalledTimes(1);
+  });
+
+  it('notifies catch-up listeners after a supervisor reconnect', () => {
+    createSubscriptionClient(BASE_CONFIG);
+    const supervisor = getSupervisor(GRAPHQL_SUBSCRIPTIONS_CONNECTION) as any;
+    const listener = vi.fn();
+    onSubscriptionCatchUp(listener);
+
+    supervisor.handleOpen();
+
+    expect(listener).toHaveBeenCalledTimes(1);
+  });
+
+  it('resolves missed events through the active client catch-up query', async () => {
+    createSubscriptionClient(BASE_CONFIG);
+    const events = await requestRealtimeCatchUp(2);
+    expect(Array.isArray(events)).toBe(true);
+    expect(events?.length).toBe(1);
+    expect(events?.[0]).toMatchObject({ id: 'evt-1', sequence: 1 });
+  });
+});

src/lib/graphql/subscriptions.ts

@@ -15,6 +15,7 @@ import { getMainDefinition } from '@apollo/client/utilities';
 import { DocumentNode, print } from 'graphql';
 import { flagStore, evaluateFlag } from '@/lib/feature-flags';
 import { createLogger } from '@/lib/logging';
+import { REALTIME_CATCHUP_QUERY } from './subscriptionQueries';
 import {
   BaseRealtimeTransport,
   ConnectionSupervisor,
@@ -27,6 +28,77 @@ const logger = createLogger('graphql-subscriptions');
 /** Name under which the GraphQL subscription supervisor is registered. */
 export const GRAPHQL_SUBSCRIPTIONS_CONNECTION = 'graphql-subscriptions';
 
+/** The most recently created GraphQL client (for catch-up queries). */
+let activeClient: ApolloClient<any> | null = null;
+
+/** Listeners notified when the realtime connection may have missed events. */
+const catchUpListeners = new Set<(since: number | undefined) => void>();
+
+/**
+ * Register a listener invoked when events may have been missed while the
+ * realtime transport was down or when an inbound sequence gap is detected.
+ * The `since` argument is the last sequence number observed by the supervisor
+ * (`undefined` when none was ever seen).
+ */
+export function onSubscriptionCatchUp(
+  listener: (since: number | undefined) => void,
+): () => void {
+  catchUpListeners.add(listener);
+  return () => catchUpListeners.delete(listener);
+}
+
+function emitSubscriptionCatchUp(): void {
+  const since = getLastRealtimeSequence();
+  catchUpListeners.forEach((listener) => {
+    try {
+      listener(since);
+    } catch (error) {
+      logger.warn('[GraphQLSubscriptions] Catch-up listener failed', { error });
+    }
+  });
+}
+
+/** Highest inbound sequence observed by the GraphQL supervisor, if any. */
+export function getLastRealtimeSequence(): number | undefined {
+  return getSupervisor(GRAPHQL_SUBSCRIPTIONS_CONNECTION)?.getLastSequence();
+}
+
+/** A single event returned by the realtime catch-up query. */
+export interface RealtimeEvent {
+  id: string;
+  sequence: number;
+  type: string;
+  payload: Record<string, unknown> | null;
+  createdAt: string;
+}
+
+/**
+ * Fetches events that occurred after `since` through the active GraphQL
+ * client (HTTP). Returns `null` when no client is available or the query
+ * fails, so consumers can fall back to the live subscription stream.
+ */
+export async function requestRealtimeCatchUp(
+  since: string | number,
+): Promise<RealtimeEvent[] | null> {
+  const client = activeClient;
+  if (!client || typeof client.query !== 'function') {
+    logger.warn('[GraphQLSubscriptions] No active client for catch-up query');
+    return null;
+  }
+  try {
+    const result = await client.query({
+      query: REALTIME_CATCHUP_QUERY,
+      variables: { since: String(since) },
+      fetchPolicy: 'network-only',
+    });
+    const events = result.data?.realtimeEvents;
+    return Array.isArray(events) ? events : null;
+  } catch (error) {
+    logger.error('[GraphQLSubscriptions] Catch-up query failed', { error });
+    return null;
+  }
+}
+
 /**
  * WebSocket subscription configuration options
  */
@@ -496,6 +568,12 @@ export function createSubscriptionClient(config: SubscriptionConfig): ApolloClie
         registerSupervisor(GRAPHQL_SUBSCRIPTIONS_CONNECTION, supervisor);
         supervisor.connect();
 
+        // Notify catch-up listeners when a inbound sequence gap is detected
+        // (events missed while the socket stayed open) and after every
+        // successful (re)connect so consumers can backfill the gap window.
+        supervisor.setCatchUpHandler(emitSubscriptionCatchUp);
+        supervisor.onReconnect(emitSubscriptionCatchUp);
+
         const wsClient = transport.getClient()!;
         const wsLink = new GraphQLWsLink(wsClient);
 
@@ -526,6 +604,8 @@ export function createSubscriptionClient(config: SubscriptionConfig): ApolloClie
     }),
   });
 
+  activeClient = client;
+
   return client;
 }
 

src/store/__tests__/synchronizationEngine.test.ts

@@ -0,0 +1,180 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+const h = vi.hoisted(() => {
+  const storage = new Map<string, string>();
+  return {
+    storage,
+    reconnectCb: { current: null as null | (() => void) },
+    catchUpCb: { current: null as null | ((since?: number) => void) },
+    requestCatchUp: vi.fn(),
+    lastSequence: { current: undefined as number | undefined },
+  };
+});
+
+vi.mock('@/lib/realtime/connectionSupervisor', () => ({
+  onAnyReconnect: (cb: () => void) => {
+    h.reconnectCb.current = cb;
+    return () => {
+      h.reconnectCb.current = null;
+    };
+  },
+}));
+
+vi.mock('@/lib/graphql/subscriptions', () => ({
+  getLastRealtimeSequence: () => h.lastSequence.current,
+  onSubscriptionCatchUp: (cb: (since?: number) => void) => {
+    h.catchUpCb.current = cb;
+    return () => {
+      h.catchUpCb.current = null;
+    };
+  },
+  requestRealtimeCatchUp: h.requestCatchUp,
+}));
+
+vi.mock('@/store/persistenceLayer', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('@/store/persistenceLayer')>();
+  return {
+    ...actual,
+    persistenceLayer: {
+      getItem: async (name: string) => h.storage.get(name) ?? null,
+      setItem: async (name: string, value: string) => {
+        h.storage.set(name, value);
+      },
+      removeItem: async (name: string) => {
+        h.storage.delete(name);
+      },
+      getJSON: async <T>(name: string): Promise<T | null> => {
+        const raw = h.storage.get(name);
+        return raw ? (JSON.parse(raw) as T) : null;
+      },
+      setJSON: async (name: string, value: unknown) => {
+        h.storage.set(name, JSON.stringify(value));
+      },
+    },
+  };
+});
+
+let syncEngine: any;
+let useStore: any;
+
+async function loadEngine() {
+  class MockBroadcastChannel {
+    onmessage: ((event: unknown) => void) | null = null;
+    postMessage = vi.fn();
+    close = vi.fn();
+    constructor(public name: string) {}
+  }
+  (window as any).BroadcastChannel = MockBroadcastChannel;
+  const engine = await import('@/store/synchronizationEngine');
+  const state = await import('@/store/stateManager');
+  syncEngine = engine.syncEngine;
+  useStore = state.useStore;
+}
+
+function persistedCursor(): number {
+  const raw = h.storage.get('realtime_event_cursor');
+  return raw ? JSON.parse(raw).sequence : 0;
+}
+
+function makeEvent(sequence: number, payload: Record<string, unknown>) {
+  return {
+    id: `evt-${sequence}`,
+    sequence,
+    type: 'state.updated',
+    payload,
+    createdAt: '2026-01-01T00:00:00Z',
+  };
+}
+
+beforeEach(async () => {
+  vi.resetModules();
+  h.storage.clear();
+  h.reconnectCb.current = null;
+  h.catchUpCb.current = null;
+  h.lastSequence.current = undefined;
+  h.requestCatchUp.mockReset();
+  await loadEngine();
+});
+
+describe('synchronization engine — reconnect catch-up', () => {
+  it('backfills missed events and advances the persisted cursor after a reconnect', async () => {
+    h.storage.set('realtime_event_cursor', JSON.stringify({ sequence: 1 }));
+    h.requestCatchUp.mockResolvedValue([
+      makeEvent(2, { user: { id: 'u-9' } }),
+      makeEvent(3, { app: { offlineMode: false } }),
+    ]);
+
+    h.reconnectCb.current!();
+
+    await vi.waitFor(() => {
+      expect(h.requestCatchUp).toHaveBeenCalledWith(1);
+    });
+    await vi.waitFor(() => {
+      expect(useStore.getState().user.id).toBe('u-9');
+    });
+    await vi.waitFor(() => {
+      expect(persistedCursor()).toBe(3);
+    });
+    expect(useStore.getState().app.lastSynced).not.toBeNull();
+  });
+
+  it('prefers the supervisor live sequence over the persisted cursor', async () => {
+    h.storage.set('realtime_event_cursor', JSON.stringify({ sequence: 1 }));
+    h.lastSequence.current = 5;
+    h.requestCatchUp.mockResolvedValue([
+      makeEvent(6, { user: { id: 'u-6' } }),
+    ]);
+
+    h.reconnectCb.current!();
+
+    await vi.waitFor(() => {
+      expect(h.requestCatchUp).toHaveBeenCalledWith(5);
+    });
+    await vi.waitFor(() => {
+      expect(persistedCursor()).toBe(6);
+    });
+  });
+
+  it('recovers missed events when the subscriptions layer signals a gap', async () => {
+    h.requestCatchUp.mockResolvedValue([
+      makeEvent(4, { user: { id: 'u-gap' } }),
+    ]);
+
+    h.catchUpCb.current?.(3);
+
+    await vi.waitFor(() => {
+      expect(h.requestCatchUp).toHaveBeenCalledWith(3);
+    });
+    await vi.waitFor(() => {
+      expect(useStore.getState().user.id).toBe('u-gap');
+    });
+  });
+
+  it('skips applying events that carry no owned state slices but still advances the cursor', async () => {
+    h.requestCatchUp.mockResolvedValue([
+      makeEvent(2, { unrelated: { anything: true } }),
+    ]);
+
+    h.reconnectCb.current!();
+
+    await vi.waitFor(() => {
+      expect(h.requestCatchUp).toHaveBeenCalledTimes(1);
+    });
+    expect(useStore.getState().user.id).toBeNull();
+    await vi.waitFor(() => {
+      expect(persistedCursor()).toBe(2);
+    });
+  });
+
+  it('survives a catch-up fetch failure without touching the cursor', async () => {
+    h.storage.set('realtime_event_cursor', JSON.stringify({ sequence: 2 }));
+    h.requestCatchUp.mockResolvedValue(null);
+
+    h.reconnectCb.current!();
+
+    await vi.waitFor(() => {
+      expect(h.requestCatchUp).toHaveBeenCalledTimes(1);
+    });
+    expect(persistedCursor()).toBe(2);
+  });
+});