Real‑Time NgRx Statecraft in Angular 20+: WebSocket Streams, Optimistic Updates, and Typed Effects for Telemetry Dashboards

Real-world stakes

The failure mode is predictable: message storms overwhelm change detection, components render mid-sequence, and optimistic updates go missing when ACKs arrive late. at a major airline, even airport kiosk peripherals (scanners/printers) created similar burst patterns—just at the edge. The fix starts with typed state, not just faster sockets.

• Charter ads analytics: 100k+ events/min rollups

• an insurance technology company telematics: rolling 30s KPIs with out-of-order packets

• a broadcast media network VPS: schedule edits with optimistic confirmation under load

Typed events → typed actions → predictable reducers

If your protocol isn’t typed, your reducers are guessing. I model the wire format as a discriminated union and generate NgRx actions from it. Effects do the messy I/O, reducers remain boring, and selectors stay memoized. That’s how you keep WebSocket chaos out of your components.

• Contracts first: define a TelemetryEvent union

• Runtime guards: zod/typebox validate unknown messages

• Action groups mirror protocol; reducers remain pure

1) Typed protocol and actions

// telemetry.models.ts
export interface Telemetry { id: string; rpm: number; temp: number; ts: number; }
export interface TelemetryDelta { id: string; changes: Partial<Telemetry>; ts: number; }
export type TelemetryEvent =
  | { type: 'snapshot'; payload: Telemetry[]; ts: number }
  | { type: 'delta'; payload: TelemetryDelta[]; ts: number }
  | { type: 'ack'; correlationId: string; ts: number }
  | { type: 'error'; code: string; message: string; ts: number };

// Optional runtime guard with zod
import { z } from 'zod';
export const TelemetryEventSchema = z.discriminatedUnion('type', [
  z.object({ type: z.literal('snapshot'), payload: z.array(z.any()), ts: z.number() }),
  z.object({ type: z.literal('delta'), payload: z.array(z.any()), ts: z.number() }),
  z.object({ type: z.literal('ack'), correlationId: z.string(), ts: z.number() }),
  z.object({ type: z.literal('error'), code: z.string(), message: z.string(), ts: z.number() }),
]);

// actions.ts
import { createActionGroup, props, emptyProps } from '@ngrx/store';
export const TelemetryActions = createActionGroup({
  source: 'Telemetry',
  events: {
    'connect': emptyProps(),
    'connected': props<{ sessionId: string }>(),
    'disconnect': emptyProps(),
    'message received': props<{ event: TelemetryEvent }>(),
    'send command': props<{ command: any; correlationId: string }>(),
    'optimistic apply delta': props<{ delta: TelemetryDelta; correlationId: string }>(),
    'revert delta': props<{ correlationId: string }>(),
    'apply delta confirmed': props<{ correlationId: string }>(),
    'upsert telemetry': props<{ items: Telemetry[] }>(),
    'error': props<{ error: string }>(),
  }
});

2) Entity reducer with pending optimistic map

// reducer.ts
import { createReducer, on } from '@ngrx/store';
import { createEntityAdapter, EntityState } from '@ngrx/entity';
import { Telemetry, TelemetryDelta } from './telemetry.models';
import { TelemetryActions } from './actions';

export interface TelemetryState extends EntityState<Telemetry> {
  connected: boolean;
  pending: Record<string, TelemetryDelta>; // correlationId -> delta
  lastTs?: number;
}

const adapter = createEntityAdapter<Telemetry>({ selectId: t => t.id });
const initialState: TelemetryState = adapter.getInitialState({ connected: false, pending: {} });

function invertDelta(delta: TelemetryDelta): TelemetryDelta {
  // In production you'd compute changes from a snapshot; simplified here.
  return { id: delta.id, changes: {}, ts: Date.now() };
}

export const telemetryReducer = createReducer(
  initialState,
  on(TelemetryActions.connected, s => ({ ...s, connected: true })),
  on(TelemetryActions.upsertTelemetry, (s, { items }) => adapter.upsertMany(items, s)),
  on(TelemetryActions.optimisticApplyDelta, (s, { delta, correlationId }) => {
    const next = adapter.updateOne({ id: delta.id, changes: delta.changes }, s);
    next.pending[correlationId] = delta;
    return next;
  }),
  on(TelemetryActions.applyDeltaConfirmed, (s, { correlationId }) => {
    const next = { ...s };
    delete next.pending[correlationId];
    return next;
  }),
  on(TelemetryActions.revertDelta, (s, { correlationId }) => {
    const delta = s.pending[correlationId];
    if (!delta) return s;
    const revert = invertDelta(delta);
    const next = adapter.updateOne({ id: revert.id, changes: revert.changes }, s);
    delete next.pending[correlationId];
    return next;
  })
);

3) WebSocket effect with exponential backoff and heartbeats

// effects.ts
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { Injectable, inject } from '@angular/core';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { EMPTY, Observable, of, timer } from 'rxjs';
import { catchError, delay, map, mergeMap, scan, startWith, switchMap, takeUntil, tap } from 'rxjs/operators';
import { TelemetryActions } from './actions';
import { TelemetryEvent } from './telemetry.models';

@Injectable({ providedIn: 'root' })
export class TelemetryEffects {
  private actions$ = inject(Actions);
  private socket!: WebSocketSubject<unknown>;
  private url = (window as any).env?.WS_URL ?? 'wss://example/ws';

  connect$ = createEffect(() => this.actions$.pipe(
    ofType(TelemetryActions.connect),
    switchMap(() => {
      let attempts = 0;
      const stream$ = new Observable<TelemetryEvent>(subscriber => {
        const open = () => {
          this.socket = webSocket({ url: this.url, deserializer: e => JSON.parse(e.data) });
          const sub = this.socket.subscribe({
            next: msg => subscriber.next(msg as TelemetryEvent),
            error: err => subscriber.error(err),
            complete: () => subscriber.complete()
          });
          return () => sub.unsubscribe();
        };
        return open();
      });

      return stream$.pipe(
        startWith({ type: 'snapshot', payload: [], ts: Date.now() } as TelemetryEvent),
        // simple backoff on error
        catchError(err => of({ type: 'error', code: 'socket', message: String(err), ts: Date.now() } as TelemetryEvent)),
        scan((acc, event) => ({ attempts: event.type === 'error' ? acc.attempts + 1 : 0, event }), { attempts: 0, event: null as any }),
        mergeMap(({ attempts, event }) => attempts > 0
          ? timer(Math.min(1000 * 2 ** attempts, 15000)).pipe(map(() => event))
          : of(event)
        ),
        takeUntil(this.actions$.pipe(ofType(TelemetryActions.disconnect))),
        map(event => TelemetryActions.messageReceived({ event }))
      );
    })
  ));

  // ACK/ERR handling, delta batching
  messages$ = createEffect(() => this.actions$.pipe(
    ofType(TelemetryActions.messageReceived),
    mergeMap(({ event }) => {
      switch (event.type) {
        case 'snapshot':
          return of(TelemetryActions.upsertTelemetry({ items: event.payload }));
        case 'delta':
          return of(TelemetryActions.upsertTelemetry({ items: event.payload.map(d => ({ id: d.id, ...d.changes })) }));
        case 'ack':
          return of(TelemetryActions.applyDeltaConfirmed({ correlationId: event.correlationId }));
        case 'error':
          return of(TelemetryActions.error({ error: event.message }));
        default:
          return EMPTY;
      }
    })
  ));

  sendCommand$ = createEffect(() => this.actions$.pipe(
    ofType(TelemetryActions.sendCommand),
    mergeMap(({ command, correlationId }) => {
      // optimistic first, then send
      const optimistic = TelemetryActions.optimisticApplyDelta({ delta: command.delta, correlationId });
      try { this.socket?.next({ ...command, correlationId }); } catch {}
      return of(optimistic);
    })
  ));
}

• Gate connect/disconnect by feature flag (Firebase Remote Config) for safe rollouts.

• TakeUntil on disconnect to prevent orphan sockets.

• Backoff with scan + delay; instrument retries.

4) Selectors to Signals for ergonomic, zone-safe UI

// selectors.ts
import { createFeature, createSelector } from '@ngrx/store';
import { adapter } from './reducer';

export const telemetryFeature = createFeature({ name: 'telemetry', reducer: telemetryReducer });
const { selectAll } = adapter.getSelectors(telemetryFeature.selectTelemetryState);
export const selectSortedTop = (k: number) => createSelector(selectAll, rows => rows
  .slice().sort((a,b) => b.ts - a.ts).slice(0, k));

// component.ts
import { toSignal } from '@angular/core/rxjs-interop';
readonly rows = toSignal(this.store.select(selectSortedTop(500)), { initialValue: [] });

• Use toSignal() for component bindings with stable initialValue.

• Keep complex UI-only state in a SignalStore; server state in NgRx.

• Avoid accidental resubscribe loops; memoize selectors.

UI that won’t jitter under bursts

<!-- component.html -->
<p-table [value]="rows()" [virtualScroll]="true" [virtualScrollItemSize]="44" [style]="{height:'520px'}">
  <ng-template pTemplate="header">
    <tr><th>ID</th><th>RPM</th><th>TEMP</th><th>TS</th></tr>
  </ng-template>
  <ng-template pTemplate="body" let-r>
    <tr>
      <td>{{r.id}}</td>
      <td>{{r.rpm | number}}</td>
      <td>{{r.temp | number:'1.0-0'}}</td>
      <td>{{r.ts | date:'mediumTime'}}</td>
    </tr>
  </ng-template>
</p-table>

• Virtual scrolling + coalesced deltas

• 50–100ms buffer windows to smooth spikes

• Instrument re-render cost with Angular DevTools

Delta coalescing in effects

// Coalesce deltas to avoid thrash
import { bufferTime, filter, map, mergeMap, of } from 'rxjs';

coalesce$ = createEffect(() => this.actions$.pipe(
  ofType(TelemetryActions.messageReceived),
  map(({ event }) => event),
  filter(e => e.type === 'delta'),
  map(e => (e as any).payload),
  bufferTime(50),
  filter(b => b.length > 0),
  map(batches => batches.flat()),
  map(deltas => TelemetryActions.upsertTelemetry({ items: deltas.map(d => ({ id: d.id, ...d.changes })) }))
));

Optimistic edit with confirmation

// usage: user tweaks a threshold; UI updates immediately, ACK confirms or ERR reverts
const correlationId = crypto.randomUUID();
this.store.dispatch(TelemetryActions.sendCommand({
  command: { type: 'update-threshold', delta: { id: row.id, changes: { temp: newTemp }, ts: Date.now() } },
  correlationId,
}));

Bring in help if you see this

If your dashboard jitters or your engineers fear touching effects, it’s time to bring in a senior Angular engineer. I’ve stabilized these patterns at a global entertainment company, a broadcast media network, United, an insurance technology company, and Charter. We’ll keep delivery moving while we fix the pipeline.

• WebSocket reconnect storms and zombie sockets

• Optimistic updates that never reconcile (ghost rows)

• UI FPS drops under burst; auditTime/bufferTime missing

• Untyped actions/effects make tests flaky

• SSR hydration mismatches on initial render

My 6-step engagement

We start with a 60–90 minute architecture review, ship the typed protocol by end of week, and land a safe feature-flagged rollout the following week. gitPlumbers keeps uptime at 99.98% during these modernizations.

• Assess protocol and define TelemetryEvent union + runtime validation

• Introduce action groups and feature store; entity adapter for upserts

• Write socket effect with backoff, heartbeats, and disconnect guards

• Implement delta coalescing and optimistic reconciliation with correlation IDs

• Expose selectors as Signals; move UI-only state to a SignalStore

• Add CI guardrails (Nx, Jest/Karma, Cypress) and metrics (GA4/Firebase, OpenTelemetry)

Answers at a glance

See below, plus reach out to discuss your Angular roadmap. Typical discovery within 48 hours; assessment in 5 business days.