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

When Your Dashboard Jitters: NgRx Patterns for Real‑Time Telemetry in Angular 20+

Real‑time means bursts, partial failures, and evolving schemas. The stack here: Angular 20, NgRx Store/Effects/Entity, Signals/SignalStore, RxJS 7, PrimeNG tables/charts, and Firebase/GA4 for telemetry. Nx keeps the contracts shared across libs and services.

Why Angular Teams Break on WebSockets and Optimistic Updates

Architecture: NgRx + Signals + WebSocketSubject + EntityAdapter

// libs/telemetry/src/lib/contracts/telemetry-event.ts
export type TelemetryEvent =
  | { type: 'metric'; id: string; ts: number; kpi: string; value: number; tenant: string }
  | { type: 'status'; id: string; ts: number; deviceId: string; state: 'online'|'offline'|'degraded'; tenant: string }
  | { type: 'ack'; correlationId: string; ok: boolean; reason?: string };

// libs/telemetry/src/lib/state/telemetry.actions.ts
import { createActionGroup, props } from '@ngrx/store';

export const TelemetryActions = createActionGroup({
  source: 'Telemetry',
  events: {
    'Connect': props<{ url: string; token: string }>(),
    'Connected': props<void>(),
    'Disconnected': props<{ reason?: string }>(),
    'Incoming Event': props<{ event: TelemetryEvent }>(),
    'Send Command': props<{ payload: unknown; correlationId: string }>(),
    'Command Ack': props<{ correlationId: string; ok: boolean; reason?: string }>(),
    'Error': props<{ error: unknown }>(),
  }
});

// libs/telemetry/src/lib/state/telemetry.effects.ts
import { Injectable, inject } from '@angular/core';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { TelemetryActions } from './telemetry.actions';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { EMPTY, interval, merge, of } from 'rxjs';
import { catchError, distinctUntilChanged, filter, map, mergeMap, retry, shareReplay, switchMap, takeUntil, tap } from 'rxjs/operators';

@Injectable({ providedIn: 'root' })
export class TelemetryEffects {
  private actions$ = inject(Actions);
  private socket?: WebSocketSubject<unknown>;

  connect$ = createEffect(() => this.actions$.pipe(
    ofType(TelemetryActions.connect),
    switchMap(({ url, token }) => {
      const ws = webSocket({ url: `${url}?token=${token}`, deserializer: e => JSON.parse(e.data) });
      this.socket = ws;

      const heartbeats$ = interval(15000).pipe(
        tap(() => ws.next({ type: 'ping', ts: Date.now() })),
      );

      const incoming$ = ws.pipe(
        map((event: any) => TelemetryActions.incomingEvent({ event })),
        catchError(error => of(TelemetryActions.error({ error })))
      );

      const stream$ = merge(incoming$, heartbeats$).pipe(
        retry({ count: Infinity, delay: (_e, i) => Math.min(1000 * 2 ** i, 30000) }),
        takeUntil(this.actions$.pipe(ofType(TelemetryActions.disconnected)))
      );

      return merge(of(TelemetryActions.connected()), stream$);
    })
  ));

  sendCommand$ = createEffect(() => this.actions$.pipe(
    ofType(TelemetryActions.sendCommand),
    tap(({ payload, correlationId }) => this.socket?.next({ ...payload, correlationId })),
    // noop effect: command echoes come via incomingEvent
    switchMap(() => EMPTY)
  ), { dispatch: false });
}

Reducers, Selectors, and Signal Bridges

// libs/telemetry/src/lib/state/telemetry.reducer.ts
import { createReducer, on, createFeature } from '@ngrx/store';
import { createEntityAdapter, EntityState } from '@ngrx/entity';
import { TelemetryActions } from './telemetry.actions';
import { TelemetryEvent } from '../contracts/telemetry-event';

interface Metric { id: string; ts: number; kpi: string; value: number; tenant: string; }
interface Status { id: string; ts: number; deviceId: string; state: 'online'|'offline'|'degraded'; tenant: string; }

const metricAdapter = createEntityAdapter<Metric>({ selectId: (m) => `${m.tenant}:${m.id}:${m.kpi}` });
const statusAdapter = createEntityAdapter<Status>({ selectId: (s) => `${s.tenant}:${s.deviceId}` });

interface State {
  metrics: EntityState<Metric>;
  statuses: EntityState<Status>;
  connected: boolean;
}

const initialState: State = {
  metrics: metricAdapter.getInitialState(),
  statuses: statusAdapter.getInitialState(),
  connected: false,
};

export const telemetryReducer = createReducer(
  initialState,
  on(TelemetryActions.connected, (s) => ({ ...s, connected: true })),
  on(TelemetryActions.disconnected, (s) => ({ ...s, connected: false })),
  on(TelemetryActions.incomingEvent, (s, { event }) => {
    switch (event.type) {
      case 'metric': return { ...s, metrics: metricAdapter.upsertOne(event, s.metrics) };
      case 'status': return { ...s, statuses: statusAdapter.upsertOne(event, s.statuses) };
      case 'ack': return s; // handled in optimistic effect
      default: return s;
    }
  })
);

export const telemetryFeature = createFeature({ name: 'telemetry', reducer: telemetryReducer });

// libs/telemetry/src/lib/state/telemetry.selectors.ts
import { createSelector } from '@ngrx/store';
const { selectMetricsState, selectStatusesState } = telemetryFeature;
const metricSelectors = metricAdapter.getSelectors(selectMetricsState);
const statusSelectors = statusAdapter.getSelectors(selectStatusesState);

export const selectTenantMetrics = (tenant: string) => createSelector(
  metricSelectors.selectAll,
  (all) => all.filter(m => m.tenant === tenant)
);

export const selectDeviceStatus = (tenant: string, deviceId: string) => createSelector(
  statusSelectors.selectEntities,
  (entities) => entities[`${tenant}:${deviceId}`]
);

// libs/telemetry/src/lib/dashboard/dashboard.component.ts
import { toSignal } from '@angular/core/rxjs-interop';

const tenant = 'acme';
metricsSig = toSignal(this.store.select(selectTenantMetrics(tenant)), { initialValue: [] });
statusSig = toSignal(this.store.select(selectDeviceStatus(tenant, 'kiosk-42')));

latestKpiSig = computed(() => {
  const byKpi = new Map<string, number>();
  for (const m of this.metricsSig()) byKpi.set(m.kpi, m.value);
  return byKpi; // used by charts
});

Optimistic Updates with Rollbacks and Correlation IDs

// libs/telemetry/src/lib/state/optimistic.actions.ts
export const OptimisticActions = createActionGroup({
  source: 'Optimistic',
  events: {
    'Start': props<{ id: string; patch: Partial<Status>; correlationId: string }>(),
    'Apply Local': props<{ id: string; patch: Partial<Status>; correlationId: string }>(),
    'Rollback': props<{ id: string; prev: Status; correlationId: string; reason?: string }>(),
    'Commit': props<{ correlationId: string }>(),
  }
});

// effect wires Start -> local apply + send command
start$ = createEffect(() => this.actions$.pipe(
  ofType(OptimisticActions.start),
  tap(({ id, patch, correlationId }) => this.store.dispatch(OptimisticActions.applyLocal({ id, patch, correlationId }))),
  map(({ id, patch, correlationId }) => TelemetryActions.sendCommand({ payload: { type: 'setStatus', id, ...patch }, correlationId }))
));

// incoming acks reconcile
ack$ = createEffect(() => this.actions$.pipe(
  ofType(TelemetryActions.incomingEvent),
  filter(({ event }) => event.type === 'ack'),
  map(({ event }) => event as { type: 'ack'; correlationId: string; ok: boolean; reason?: string }),
  mergeMap(({ correlationId, ok, reason }) => ok
    ? of(OptimisticActions.commit({ correlationId }), TelemetryActions.commandAck({ correlationId, ok }))
    : of(OptimisticActions.rollback({ id: 'n/a', prev: {} as any, correlationId, reason }), TelemetryActions.commandAck({ correlationId, ok, reason }))
  )
));

// reducer keeps a small pending map keyed by correlationId
interface Pending { [correlationId: string]: { id: string; prev: Status } }

on(OptimisticActions.applyLocal, (s, { id, patch, correlationId }) => {
  const prev = statusAdapter.getSelectors().selectEntities(s.statuses)[id]!;
  return {
    ...s,
    statuses: statusAdapter.updateOne({ id, changes: patch }, s.statuses),
    pending: { ...s.pending, [correlationId]: { id, prev } }
  };
}),

on(OptimisticActions.rollback, (s, { correlationId }) => {
  const p = s.pending[correlationId];
  if (!p) return s;
  const { id, prev } = p;
  const next = statusAdapter.upsertOne(prev, s.statuses);
  const { [correlationId]: _, ...rest } = s.pending;
  return { ...s, statuses: next, pending: rest };
}),

on(OptimisticActions.commit, (s, { correlationId }) => {
  const { [correlationId]: _, ...rest } = s.pending;
  return { ...s, pending: rest };
})

Production Examples: Ads Analytics and Kiosk Telemetry

Dashboard UX: Backpressure, Virtualization, and PrimeNG

# tools/ci/render-guard.yml (conceptual)
checks:
  dashboard-home:
    maxRenders: 3  # per signal change
  campaign-table:
    maxRenders: 2

When to Hire an Angular Developer for Legacy Rescue (WebSocket State)

How an Angular Consultant Structures Typed Actions and Effects for Telemetry

Key Takeaways and What to Instrument Next