Skip to main content

WebSocket Real-Time Updates Pattern

Template for implementing real-time features in this project using RTK Query streaming subscriptions over Socket.IO.

Overview​

The frontend WebSocket layer used to be a custom useWebSocket React hook that each domain hook wrapped (one socket per consumer, manual reconnect, useState everywhere). That pattern has been retired.

The current pattern moves the socket lifecycle out of React and lets RTK Query own subscriber refcounting:

  • One persistent socket per namespace, owned by a module-scoped factory in @toast-guilder/web-shared/lib/socket.
  • Each real-time stream is an RTK Query endpoint with an empty queryFn and an onCacheEntryAdded body that opens/closes the subscription.
  • Components consume the generated useGet*Query hook — they never touch the socket directly.
  • On logout, closeAllSockets() is called from the auth layer so the next sign-in opens fresh sockets with the new session cookie.

This is the canonical pattern documented in the RTK Query docs as streaming updates.

Architecture​

flowchart TD
    subgraph Frontend
        comp["Component"]
        hook["Domain hook<br/>(useRosterSync, useSimSync, ...)"]
        rtkq["RTK Query endpoint<br/>(syncProgressApi)<br/>onCacheEntryAdded"]
        factory["socketFactory<br/>getSocket(namespace)"]
    end

    subgraph Backend["NestJS Backend"]
        gw["Gateway<br/>Room-based subscriptions"]
        proc["Bull Processor<br/>(emits progress)"]
    end

    comp --> hook
    hook --> rtkq
    rtkq -->|getSocket| factory
    factory <-->|Socket.IO| gw
    proc -->|emits| gw

Refcounting and dedup happens at two levels:

  1. Socket level — getSocket(namespace) returns the same Socket instance for the lifetime of the page. Multiple endpoints sharing a namespace share the connection.
  2. Subscription level — RTK Query refcounts at the cache-key level. Two components watching the same sessionId share one subscription; the first subscriber opens it, the last unsubscribes.

Backend Implementation​

Unchanged — the backend gateway pattern is still what it was. Summarized here for completeness; see existing gateways (character-sync.gateway.ts, roster-sync.gateway.ts, meta-build-sync.gateway.ts) for live examples.

1. Create WebSocket Gateway​

Location: apps/api/src/domains/[domain]/gateways/[feature].gateway.ts

@WebSocketGateway({
  cors: {
    origin: process.env.WEB_URL || 'http://localhost:5173',
    credentials: true,
  },
  namespace: '/[feature-name]',
})
export class FeatureGateway
  implements OnGatewayConnection, OnGatewayDisconnect
{
  @WebSocketServer()
  server: Server;

  @SubscribeMessage('subscribe-sync')
  async handleSubscribe(
    @MessageBody() data: { sessionId: string },
    @ConnectedSocket() client: Socket
  ) {
    client.join(data.sessionId);
    // Optionally send current state immediately so late joiners catch up
    const state = await this.prisma.featureSession.findUnique({
      where: { id: data.sessionId },
    });
    if (state) this.emitProgress(data.sessionId, state);
  }

  @SubscribeMessage('unsubscribe-sync')
  handleUnsubscribe(
    @MessageBody() data: { sessionId: string },
    @ConnectedSocket() client: Socket
  ) {
    client.leave(data.sessionId);
  }

  emitProgress(sessionId: string, progress: FeatureProgress) {
    this.server.to(sessionId).emit('sync-progress', progress);
  }

  emitComplete(sessionId: string, result: FeatureProgress) {
    this.server.to(sessionId).emit('sync-complete', result);
  }

  emitError(sessionId: string, error: string) {
    this.server.to(sessionId).emit('sync-error', { sessionId, error });
  }
}

2. Background Processor Emits Events​

Inject the gateway into your Bull processor and emit sync-progress / sync-complete / sync-error to the room as work moves along. (No change from the previous pattern.)

3. Authentication​

Gateways accept the same session cookie as the HTTP API (credentials: true + withCredentials: true on the client). No bespoke auth on the WS layer.

Frontend Implementation​

1. Use the Shared getSocket() Factory — Don't Open Your Own​

Lives in packages/web-shared/src/lib/socket/socketFactory.ts and re-exported as @toast-guilder/web-shared/lib.

import { getSocket, closeAllSockets } from '@toast-guilder/web-shared/lib';

const socket = getSocket('/feature-name');
  • One Socket instance per namespace, created lazily on first request.
  • withCredentials: true so the session cookie rides along — auth comes from the same cookie as REST.
  • closeAllSockets() is called from the auth logout() flow so a session change forces a clean reconnect.

Don't call io() yourself or hold a Socket in component state. That breaks the single-socket-per-namespace invariant.

2. Add a Streaming Endpoint to syncProgressApi​

All sync streams live in one place: apps/web/src/store/api/syncProgressApi.ts (which re-exports @toast-guilder/web-shared/api/syncProgressApi). Add a new endpoint there rather than spinning up a new file per feature.

// packages/web-shared/src/api/syncProgressApi.ts
export interface FeatureProgress {
  sessionId: string;
  status: 'pending' | 'processing' | 'completed' | 'failed';
  processed: number;
  total: number;
}

export interface FeatureState {
  progress: FeatureProgress | null;
  error: string | null;
}

export const syncProgressApi = baseApi.injectEndpoints({
  endpoints: (builder) => ({
    getFeatureSyncProgress: builder.query<FeatureState, string>({
      // queryFn never fetches HTTP — it just seeds the initial cache shape
      queryFn: () => ({ data: { progress: null, error: null } }),

      async onCacheEntryAdded(
        sessionId,
        { updateCachedData, cacheDataLoaded, cacheEntryRemoved, dispatch }
      ) {
        const socket = getSocket('/feature-name');

        try {
          await cacheDataLoaded;
        } catch {
          return; // cache entry removed before initial data resolved
        }

        socket.emit('subscribe-sync', { sessionId });

        const onProgress = (data: FeatureProgress) => {
          updateCachedData(() => ({ progress: data, error: null }));
        };

        const onComplete = (data: FeatureProgress) => {
          updateCachedData(() => ({ progress: data, error: null }));
          // Invalidate any HTTP queries that depend on this stream
          dispatch(
            baseApi.util.invalidateTags([{ type: 'Feature', id: 'LIST' }])
          );
        };

        const onError = (data: { sessionId: string; error: string }) => {
          updateCachedData((draft) => {
            draft.error = data.error;
          });
        };

        socket.on('sync-progress', onProgress);
        socket.on('sync-complete', onComplete);
        socket.on('sync-error', onError);

        await cacheEntryRemoved;

        socket.emit('unsubscribe-sync', { sessionId });
        socket.off('sync-progress', onProgress);
        socket.off('sync-complete', onComplete);
        socket.off('sync-error', onError);
      },
    }),
  }),
});

export const { useGetFeatureSyncProgressQuery } = syncProgressApi;

Key points:

  • queryFn returns the initial shape only. RTK Query is the cache, the socket is the producer.
  • onCacheEntryAdded is the entire lifecycle: subscribe on first interest, unsubscribe when the last consumer goes away (plus keepUnusedDataFor).
  • Use updateCachedData((draft) => {...}) (Immer) for partial mutations; use () => newState for full replacements.
  • Dispatch tag invalidation here when the stream completes, so dependent HTTP queries refetch.

3. Watching Multiple IDs at Once​

For admin/batch flows that subscribe to several sessions in one shot, use a list cache key with a custom serializeQueryArgs. The cached shape is a map keyed by id:

getRosterSyncProgress: builder.query<RosterSyncMap, string[]>({
  queryFn: () => ({ data: {} as RosterSyncMap }),
  serializeQueryArgs: ({ queryArgs }) => [...queryArgs].sort().join(','),
  async onCacheEntryAdded(sessionIds, { updateCachedData, cacheDataLoaded, cacheEntryRemoved }) {
    if (!sessionIds.length) return;
    const socket = getSocket('/roster-sync');
    await cacheDataLoaded;

    sessionIds.forEach((id) => socket.emit('subscribe-sync', { sessionId: id }));

    const onProgress = (data: RosterSyncProgress) => {
      updateCachedData((draft) => {
        draft[data.sessionId] = data;
      });
    };
    socket.on('sync-progress', onProgress);

    await cacheEntryRemoved;

    sessionIds.forEach((id) => socket.emit('unsubscribe-sync', { sessionId: id }));
    socket.off('sync-progress', onProgress);
  },
}),

This is the pattern used by getRosterSyncProgress (admin batch) and getBatchClassification. Single-session consumers pass [sessionId] and pluck their entry out.

4. Thin Domain Hook​

Domain hooks are now small wrappers around the generated query hook. No useState for progress, no useEffect for subscribe/unsubscribe, no manual socket plumbing.

// apps/web/src/hooks/sync/useFeatureSync.ts
import { useState } from 'react';
import { skipToken } from '@reduxjs/toolkit/query/react';
import { useGetFeatureSyncProgressQuery } from '@/store/api/syncProgressApi';

export function useFeatureSync() {
  const [sessionId, setSessionId] = useState<string | null>(null);

  const { data } = useGetFeatureSyncProgressQuery(sessionId ?? skipToken);

  return {
    progress: data?.progress ?? null,
    error: data?.error ?? null,
    isSyncing: sessionId !== null && data?.progress?.status !== 'completed',
    startSync: (id: string) => setSessionId(id),
  };
}

skipToken is critical — it tells RTK Query not to mount the subscription until you actually have a session id, so the socket isn't opened on every page render.

5. Component Usage​

import { useFeatureSync } from '@/hooks/sync/useFeatureSync';

function FeaturePage() {
  const { progress, isSyncing, startSync } = useFeatureSync();
  const [trigger] = useStartFeatureMutation();

  const handleStart = async () => {
    const { sessionId } = await trigger().unwrap();
    startSync(sessionId);
  };

  return (
    <>
      <Button onPress={handleStart} isDisabled={isSyncing}>Start</Button>
      {progress && <FeatureSyncProgress progress={progress} />}
    </>
  );
}

Why This Pattern​

  1. No socket-per-consumer leakage — components can't accidentally open duplicate sockets.
  2. No useState/useEffect plumbing in domain hooks — RTK Query owns the cache, the factory owns the connection.
  3. Automatic dedup — two components watching the same session share one subscription.
  4. StrictMode/Activity safe — keepUnusedDataFor keeps the cache entry alive across remounts, so the socket doesn't churn.
  5. Cross-app reuse — both apps/web and apps/admin consume the same @toast-guilder/web-shared/api/syncProgressApi.
  6. Clean logout — closeAllSockets() is called from auth/auth.ts on sign-out, ensuring the next user gets fresh connections.

Testing​

Streaming endpoints​

The pattern test files are in apps/web/src/store/api/syncProgressApi.realtime.test.tsx and notificationsApi.realtime.test.tsx. They mock socket.io-client with an event-emitting stub and assert that cached state updates as events fire.

Use those as the template — don't roll your own socket mock per test.

Manual gateway smoke test​

npm install -g wscat
wscat -c ws://localhost:3001/feature-name
{"event": "subscribe-sync", "data": {"sessionId": "uuid-here"}}

Migrating an Existing Hook​

If you find a hook that still does io(...) directly, holds a socket in useRef, or imports a now-deleted useWebSocket:

  1. Move its socket subscription into a new endpoint on syncProgressApi (onCacheEntryAdded).
  2. Replace the hook body with useGet*Query(arg ?? skipToken).
  3. Delete the manual useEffect subscribe/unsubscribe block.
  4. Confirm getSocket() is the only place io(...) is called.

File Structure​

packages/web-shared/src/
├── api/
│   └── syncProgressApi.ts          # All streaming RTK Query endpoints
└── lib/
    └── socket/
        └── socketFactory.ts        # getSocket() + closeAllSockets()

apps/web/src/
├── hooks/sync/
│   └── use[Feature]Sync.ts         # Thin wrapper around useGet*Query
├── lib/
│   ├── socket/socketFactory.ts     # Re-exports from web-shared
│   └── auth/auth.ts                # Calls closeAllSockets() on logout
└── store/api/
    ├── baseApi.ts                  # RTK Query base with tagTypes
    └── syncProgressApi.ts          # Re-exports from web-shared

Reference Implementations​

  • Roster sync (admin batch, multi-session): getRosterSyncProgress endpoint + useRosterSync / useAdminRosterSync hooks
  • Character sync (per-user Battle.net): getCharacterSyncProgress endpoint + useCharacterSync hook
  • GDL sync (game-data catalogs): getGdlSyncProgress endpoint
  • Meta-build sync: getMetaBuildSyncProgress endpoint
  • Notifications (non-sync streaming example): notificationsApi.onCacheEntryAdded for /notifications namespace

All of these consume getSocket() from the same factory; there is one socket per namespace shared across every consumer of every cache key inside that namespace.