Production‑Grade Real‑Time NgRx in Angular 20+: Typed WebSockets, Optimistic Updates, and SignalStore Dashboards

The 9:17am Dashboard Spike

Why Typed Real‑Time State Matters in Angular 20+

Typed actions/effects ensure every message through your pipeline is validated at compile time. Signals prevent the zone.js re-render tax on hot paths. Add heartbeats and exponential backoff so reconnections don’t melt your APIs.

Implement Typed WebSocket Streams with NgRx

// contracts/models.ts
export type TenantId = string;
export type CorrelationId = string;

export interface TelemetryBase { ts: number; tenantId: TenantId; }
export interface CpuEvent extends TelemetryBase { type: 'cpu'; host: string; pct: number; }
export interface ErrorEvent extends TelemetryBase { type: 'error'; code: string; message: string; }
export type TelemetryEvent = CpuEvent | ErrorEvent;

export interface SetThresholdCmd { kind: 'set-threshold'; host: string; pct: number; cid: CorrelationId; }
export interface Ack { kind: 'ack'; cid: CorrelationId; }
export interface Nack { kind: 'nack'; cid: CorrelationId; reason: string; }
export type ServerMsg = TelemetryEvent | Ack | Nack;
export type ClientMsg = SetThresholdCmd;

// store/telemetry.actions.ts
import { createActionGroup, emptyProps, props } from '@ngrx/store';
import { TelemetryEvent, Ack, Nack, ClientMsg, CorrelationId } from '../contracts/models';

export const TelemetryActions = createActionGroup({
  source: 'Telemetry',
  events: {
    'Connect': emptyProps(),
    'Connected': emptyProps(),
    'Disconnected': props<{ reason?: unknown }>(),
    'HeartbeatTimeout': emptyProps(),

    'Event In': props<{ event: TelemetryEvent }>(),

    'Cmd Send': props<{ cmd: ClientMsg }>(),
    'Cmd Ack': props<{ ack: Ack }>(),
    'Cmd Nack': props<{ nack: Nack }>(),
    'Cmd Timeout': props<{ cid: CorrelationId }>(),
  }
});

// services/websocket.service.ts
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { defer, Observable, timer } from 'rxjs';
import { backoff } from 'rxjs-backoff'; // or write your own expo backoff
import { ServerMsg, ClientMsg } from '../contracts/models';

export class WebSocketService {
  private socket$?: WebSocketSubject<ServerMsg | ClientMsg>;
  messages$: Observable<ServerMsg> = defer(() => {
    this.socket$ = webSocket<ServerMsg | ClientMsg>({ url: this.url, deserializer: e => JSON.parse(e.data) });
    return this.socket$ as unknown as Observable<ServerMsg>;
  }).pipe(
    backoff({ initialInterval: 500, maxInterval: 10_000, resetOnSuccess: true })
  );

  constructor(private url: string) {}
  send(msg: ClientMsg) { this.socket$?.next(msg); }
  close() { this.socket$?.complete(); }
}

// store/telemetry.effects.ts
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { inject, Injectable } from '@angular/core';
import { TelemetryActions } from './telemetry.actions';
import { WebSocketService } from '../services/websocket.service';
import { filter, map, merge, mergeMap, of, race, take, takeUntil, tap, timer } from 'rxjs';
import { isTelemetryEvent } from '../contracts/guards';

@Injectable({ providedIn: 'root' })
export class TelemetryEffects {
  private actions$ = inject(Actions);
  private ws = new WebSocketService('/ws/telemetry');

  connect$ = createEffect(() =>
    this.actions$.pipe(
      ofType(TelemetryActions.connect),
      mergeMap(() => this.ws.messages$.pipe(
        tap({ error: err => console.error('socket error', err) }),
        map(msg => {
          if ('type' in msg) return TelemetryActions.eventIn({ event: msg });
          if (msg.kind === 'ack') return TelemetryActions.cmdAck({ ack: msg });
          if (msg.kind === 'nack') return TelemetryActions.cmdNack({ nack: msg });
          return TelemetryActions.disconnected({ reason: 'unknown-msg' });
        }),
        takeUntil(this.actions$.pipe(ofType(TelemetryActions.disconnected)))
      ))
    )
  );

  sendCmd$ = createEffect(() =>
    this.actions$.pipe(
      ofType(TelemetryActions.cmdSend),
      tap(a => this.ws.send(a.cmd)),
      mergeMap(a => {
        const cid = a.cmd.cid;
        const ack$ = this.actions$.pipe(ofType(TelemetryActions.cmdAck), filter(x => x.ack.cid === cid), take(1));
        const nack$ = this.actions$.pipe(ofType(TelemetryActions.cmdNack), filter(x => x.nack.cid === cid), take(1));
        const timeout$ = timer(5000).pipe(map(() => TelemetryActions.cmdTimeout({ cid })));
        return race(ack$, nack$, timeout$);
      })
    )
  );
}

// store/telemetry.reducer.ts
import { createReducer, on } from '@ngrx/store';
import { TelemetryActions } from './telemetry.actions';
import { TelemetryEvent } from '../contracts/models';

export interface DashboardState {
  events: TelemetryEvent[]; // windowed, not infinite
  pending: Record<string, { host: string; pct: number; ts: number }>; // by cid
  thresholdByHost: Record<string, number>;
}

const initial: DashboardState = { events: [], pending: {}, thresholdByHost: {} };

export const dashboardReducer = createReducer(
  initial,
  on(TelemetryActions.eventIn, (s, { event }) => ({
    ...s,
    events: [event, ...s.events].slice(0, 1000) // window
  })),
  // optimistic patch on send
  on(TelemetryActions.cmdSend, (s, { cmd }) => ({
    ...s,
    pending: { ...s.pending, [cmd.cid]: { host: cmd.host, pct: cmd.pct, ts: Date.now() } },
    thresholdByHost: { ...s.thresholdByHost, [cmd.host]: cmd.pct },
  })),
  // confirm
  on(TelemetryActions.cmdAck, (s, { ack }) => {
    const { [ack.cid]: _, ...rest } = s.pending; return { ...s, pending: rest };
  }),
  // rollback on NACK/timeout
  on(TelemetryActions.cmdNack, (s, { nack }) => {
    const p = s.pending[nack.cid]; if (!p) return s;
    const { [nack.cid]: _, ...rest } = s.pending;
    const current = { ...s.thresholdByHost };
    // simple rollback: remove optimistic if known previous exists (not shown here)
    delete current[p.host];
    return { ...s, pending: rest, thresholdByHost: current };
  }),
);

// store/telemetry.store.ts (SignalStore)
import { signalStore, withState, withComputed, patchState } from '@ngrx/signals';
import { computed, effect } from '@angular/core';
import { DashboardState } from './telemetry.reducer';

export const TelemetryStore = signalStore(
  withState<DashboardState>({ events: [], pending: {}, thresholdByHost: {} }),
  withComputed(({ events }) => ({
    cpuAvg: computed(() => {
      const slice = events().filter(e => e.type === 'cpu').slice(0, 300);
      const sum = slice.reduce((s, e: any) => s + e.pct, 0);
      return slice.length ? Math.round(sum / slice.length) : 0;
    }),
    tableRows: computed(() => events().slice(0, 500)),
  })),
);

Optimistic Updates with Correlation IDs and Rollback

// component example
submitThreshold(host: string, pct: number) {
  const cid = crypto.randomUUID();
  this.store.dispatch(TelemetryActions.cmdSend({ cmd: { kind: 'set-threshold', host, pct, cid } }));
}

<!-- PrimeNG DataTable w/ pending indicator -->
<p-table [value]="rows()">
  <ng-template pTemplate="body" let-row>
    <span>{{ row.host }}</span>
    <span [class.pending]="pendingByHost()[row.host]">{{ thresholdByHost()[row.host] }}%</span>
  </ng-template>
</p-table>

.pending { position: relative; }
.pending::after { content: '•'; color: var(--orange-500); margin-left: .25rem; }

Signals‑First Rendering with NgRx Pipes

// bridging selector to signal when not using SignalStore
import { toSignal } from '@angular/core/rxjs-interop';
this.cpuAvg = toSignal(this.store.select(selectCpuAvg), { initialValue: 0 });

Multi‑Tenant and Role‑Aware Streams

// effect guard for tenant context
const tenantId$ = this.store.selectSignal(selectTenantId);
...
map(msg => (msg.tenantId && msg.tenantId !== tenantId$()) ? null : msg),
filter(Boolean)

When to Hire an Angular Developer for Legacy Rescue

I’ve done this across airport kiosks (Docker hardware simulation for scanners/printers), telecom analytics, and insurance telematics. The result: resilient transport, smooth UIs, and measurable KPIs your PM can take to leadership.

How an Angular Consultant Approaches Signals Migration for NgRx Interop

# GitHub Actions fragment: run e2e + perf budgets
jobs:
  e2e:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with: { node-version: 20 }
      - run: npm ci
      - run: npm run test:ci && npm run e2e:ci
      - run: npm run lhci:assert # Core Web Vitals gates

Practical Outcomes and What to Measure Next