Real‑Time NgRx for Telemetry Dashboards in Angular 20+: WebSockets, Optimistic Updates, and Typed Effects (Why to Hire an Angular Consultant)

The Real‑Time Angular Dashboard That Jitters: Why NgRx + WebSockets Fix It

Why Angular 20 Teams Need Typed NgRx for Telemetry Streams

Here’s the short version: typed NgRx keeps your pipeline honest, Signals keep your UI smooth, and a few ops‑grade patterns keep it online. If you need an Angular expert who’s shipped this at scale, I’m available for remote engagements.

Architecture Blueprint: NgRx, WebSockets, Optimistic Updates, and SignalStore

Comparison: delivery channels for ‘live’ data

For dashboards that must accept user commands (filters, edits, control signals) and reflect server acks, WebSockets + NgRx effects win.

Nx Workspace Layout for Real‑Time State

# Generate libraries (example)
nx g @nx/angular:lib dashboard-feature --directory=app --standalone
nx g @nx/angular:lib dashboard-data-access --directory=app
nx g @nx/angular:lib shared-schemas --directory=app

# Add NgRx in data-access
nx g @ngrx/schematics:store State --module app/dashboard-data-access/src/lib/data-access.module.ts --creators --minimal

Typed Event Schemas and Actions

// app/shared-schemas/src/lib/telemetry.events.ts
export type ServerEvent =
  | { type: 'telemetry.batch.v1'; seq: number; items: TelemetryPoint[] }
  | { type: 'telemetry.upsert.v1'; item: TelemetryPoint }
  | { type: 'ack.v1'; correlationId: string }
  | { type: 'nack.v1'; correlationId: string; reason: string };

export type ClientCommand =
  | { type: 'filter.update.v1'; filter: Filter }
  | { type: 'point.update.v1'; id: string; patch: Partial<TelemetryPoint>; correlationId: string };

export interface TelemetryPoint { id: string; ts: number; value: number; source: string }
export interface Filter { site?: string; region?: string; campaign?: string }

export const isServerEvent = (m: unknown): m is ServerEvent =>
  typeof m === 'object' && m !== null && 'type' in (m as any);

// app/dashboard-data-access/src/lib/telemetry.actions.ts
import { createActionGroup, props, emptyProps } from '@ngrx/store';
import { ServerEvent, ClientCommand, TelemetryPoint, Filter } from '@app/shared-schemas';

export const socketActions = createActionGroup({
  source: 'Socket',
  events: {
    'Connect': emptyProps(),
    'Connected': props<{ sessionId: string }>(),
    'Disconnected': props<{ code?: number; reason?: string }>(),
    'Error': props<{ error: unknown }>(),
    'Incoming': props<{ event: ServerEvent }>(),
    'Send': props<{ cmd: ClientCommand }>(),
  }
});

export const telemetryActions = createActionGroup({
  source: 'Telemetry',
  events: {
    'UpsertMany': props<{ items: TelemetryPoint[] }>(),
    'UpsertOne': props<{ item: TelemetryPoint }>(),
    'Ack': props<{ correlationId: string }>(),
    'Nack': props<{ correlationId: string; reason: string }>(),
    'ApplyFilter': props<{ filter: Filter }>(),
    'OptimisticPatch': props<{ id: string; patch: Partial<TelemetryPoint>; correlationId: string }>(),
    'RollbackPatch': props<{ id: string; correlationId: string }>(),
  }
});

Reducers with EntityAdapter and Latency Compensation

// app/dashboard-data-access/src/lib/telemetry.reducer.ts
import { createReducer, on } from '@ngrx/store';
import { createEntityAdapter, EntityState } from '@ngrx/entity';
import { telemetryActions } from './telemetry.actions';
import { TelemetryPoint } from '@app/shared-schemas';

interface OptimisticLog {
  [correlationId: string]: { id: string; before: TelemetryPoint };
}

export interface TelemetryState extends EntityState<TelemetryPoint> {
  filter: Partial<{ site: string; region: string; campaign: string }>;
  optimistic: OptimisticLog;
}

const adapter = createEntityAdapter<TelemetryPoint>({ selectId: (p) => p.id });
const initialState: TelemetryState = adapter.getInitialState({ filter: {}, optimistic: {} });

export const telemetryReducer = createReducer(
  initialState,
  on(telemetryActions.UpsertMany, (s, { items }) => adapter.upsertMany(items, s)),
  on(telemetryActions.UpsertOne, (s, { item }) => adapter.upsertOne(item, s)),
  on(telemetryActions.ApplyFilter, (s, { filter }) => ({ ...s, filter })),
  on(telemetryActions.OptimisticPatch, (s, { id, patch, correlationId }) => {
    const before = s.entities[id];
    if (!before) return s;
    const after = { ...before, ...patch };
    const withEntity = adapter.upsertOne(after, s);
    return { ...withEntity, optimistic: { ...withEntity.optimistic, [correlationId]: { id, before } } };
  }),
  on(telemetryActions.Ack, (s, { correlationId }) => {
    const { [correlationId]: _, ...rest } = s.optimistic; // drop log
    return { ...s, optimistic: rest };
  }),
  on(telemetryActions.RollbackPatch, (s, { id, correlationId }) => {
    const entry = s.optimistic[correlationId];
    if (!entry) return s;
    const withEntity = adapter.upsertOne(entry.before, s);
    const { [correlationId]: _, ...rest } = withEntity.optimistic;
    return { ...withEntity, optimistic: rest };
  })
);

export const { selectAll: selectAllPoints, selectEntities: selectPointMap } = adapter.getSelectors();

Effects for WebSocket Lifecycle and Exponential Backoff

// app/dashboard-data-access/src/lib/telemetry.effects.ts
import { inject, Injectable } from '@angular/core';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { Store } from '@ngrx/store';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { concatMap, delayWhen, filter, map, merge, mergeMap, of, retryWhen, scan, switchMap, takeUntil, tap, timer } from 'rxjs';
import { socketActions, telemetryActions } from './telemetry.actions';
import { isServerEvent, ServerEvent, ClientCommand } from '@app/shared-schemas';

@Injectable({ providedIn: 'root' })
export class TelemetryEffects {
  private actions$ = inject(Actions);
  private store = inject(Store);

  private connect$ = createEffect(() =>
    this.actions$.pipe(
      ofType(socketActions.Connect),
      switchMap(() => {
        const ws = webSocket({ url: 'wss://api.example.com/telemetry', deserializer: (e) => JSON.parse(e.data) });

        const incoming$ = ws.pipe(
          filter(isServerEvent),
          map((event: ServerEvent) => socketActions.Incoming({ event })),
          retryWhen((errors) =>
            errors.pipe(
              scan((acc, err) => ({ count: acc.count + 1, err }), { count: 0, err: null as unknown }),
              delayWhen((s) => timer(Math.min(30000, 1000 * Math.pow(2, s.count)) + Math.floor(Math.random() * 500)))
            )
          )
        );

        const outgoing$ = this.actions$.pipe(
          ofType(socketActions.Send),
          tap(({ cmd }) => ws.next(cmd as ClientCommand)),
          filter(() => false)
        );

        const disconnect$ = this.actions$.pipe(ofType(socketActions.Disconnected));

        return merge(incoming$, outgoing$).pipe(takeUntil(disconnect$));
      })
    )
  );

  handleIncoming$ = createEffect(() =>
    this.actions$.pipe(
      ofType(socketActions.Incoming),
      mergeMap(({ event }) => {
        switch (event.type) {
          case 'telemetry.batch.v1':
            return of(telemetryActions.UpsertMany({ items: event.items }));
          case 'telemetry.upsert.v1':
            return of(telemetryActions.UpsertOne({ item: event.item }));
          case 'ack.v1':
            return of(telemetryActions.Ack({ correlationId: event.correlationId }));
          case 'nack.v1':
            return of(telemetryActions.Nack({ correlationId: event.correlationId }));
          default:
            return of();
        }
      })
    )
  );
}

Optimistic Updates with Outbox and Correlation IDs

// app/dashboard-data-access/src/lib/optimistic.effects.ts
import { Injectable, inject } from '@angular/core';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { Store } from '@ngrx/store';
import { delay, map, merge, mergeMap, of, race, take, timer } from 'rxjs';
import { socketActions, telemetryActions } from './telemetry.actions';

@Injectable({ providedIn: 'root' })
export class OptimisticEffects {
  private actions$ = inject(Actions);
  private store = inject(Store);

  // Dispatch a command and apply optimistic change
  sendPatch$ = createEffect(() =>
    this.actions$.pipe(
      ofType(telemetryActions.OptimisticPatch),
      map(({ id, patch, correlationId }) =>
        socketActions.Send({ cmd: { type: 'point.update.v1', id, patch, correlationId } })
      )
    )
  );

  // Rollback on NACK or timeout
  watchdog$ = createEffect(() =>
    this.actions$.pipe(
      ofType(telemetryActions.OptimisticPatch),
      mergeMap(({ id, correlationId }) =>
        race(
          this.actions$.pipe(ofType(telemetryActions.Ack), take(1)),
          this.actions$.pipe(ofType(telemetryActions.Nack), take(1)),
          timer(5000).pipe(map(() => ({ type: 'timeout' as const })))
        ).pipe(
          map((signal) => {
            if ((signal as any).type === 'timeout') {
              return telemetryActions.RollbackPatch({ id, correlationId });
            }
            if ((signal as any).correlationId !== correlationId) {
              // ignore unrelated ack/nack, rely on timeout
              return { type: '[noop] IrrelevantAck' } as any;
            }
            return signal.type === 'nack.v1'
              ? telemetryActions.RollbackPatch({ id, correlationId })
              : telemetryActions.Ack({ correlationId });
          })
        )
      )
    )
  );
}

Bridging to Signals and PrimeNG UI

// app/dashboard-data-access/src/lib/telemetry.selectors.ts
import { createFeatureSelector, createSelector } from '@ngrx/store';
import { TelemetryState, selectAllPoints } from './telemetry.reducer';

export const selectTelemetryState = createFeatureSelector<TelemetryState>('telemetry');
export const selectFilter = createSelector(selectTelemetryState, (s) => s.filter);
export const selectPoints = createSelector(selectTelemetryState, selectAllPoints);
export const selectFilteredPoints = createSelector(
  selectPoints,
  selectFilter,
  (points, f) => points.filter(p => (!f.site || p.source === f.site))
);

// app/dashboard-feature/src/lib/telemetry-table.component.ts
import { ChangeDetectionStrategy, Component, computed, inject } from '@angular/core';
import { Store } from '@ngrx/store';
import { selectSignal } from '@ngrx/store';
import { selectFilteredPoints } from '@app/dashboard-data-access';

@Component({
  selector: 'app-telemetry-table',
  standalone: true,
  templateUrl: './telemetry-table.component.html',
  changeDetection: ChangeDetectionStrategy.OnPush
})
export class TelemetryTableComponent {
  private store = inject(Store);
  points = this.store.selectSignal(selectFilteredPoints);
  total = computed(() => this.points().length);
}

<!-- app/dashboard-feature/src/lib/telemetry-table.component.html -->
<p-table
  [value]="points()"
  [virtualScroll]="true"
  [rows]="50"
  [scrollHeight]="'60vh'"
  [style]="{ fontSize: '12px' }"
>
  <ng-template pTemplate="header">
    <tr><th>Time</th><th>Source</th><th>Value</th></tr>
  </ng-template>
  <ng-template pTemplate="body" let-row>
    <tr>
      <td>{{ row.ts | date:'mediumTime' }}</td>
      <td>{{ row.source }}</td>
      <td>{{ row.value | number:'1.0-2' }}</td>
    </tr>
  </ng-template>
</p-table>

Testing, Telemetry, and CI Guardrails

// sample effect test (Jasmine + marble)
import { TestBed } from '@angular/core/testing';
import { provideMockActions } from '@ngrx/effects/testing';
import { hot, cold } from 'jasmine-marbles';
import { TelemetryEffects } from './telemetry.effects';
import { socketActions, telemetryActions } from './telemetry.actions';

it('emits UpsertMany on batch event', () => {
  const actions$ = hot('-a', { a: socketActions.Incoming({ event: { type: 'telemetry.batch.v1', seq: 1, items: [{ id: '1', ts: 1, value: 10, source: 's' }] } }) });
  TestBed.configureTestingModule({ providers: [TelemetryEffects, provideMockActions(() => actions$)] });
  const effects = TestBed.inject(TelemetryEffects);
  expect(effects.handleIncoming$).toBeObservable(cold('-b', { b: telemetryActions.UpsertMany({ items: [{ id: '1', ts: 1, value: 10, source: 's' }] }) }));
});

# CI snippet: run tests + budgets
name: ci
on: [push, pull_request]
jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with: { node-version: 20 }
      - run: npm ci
      - run: npx nx run-many -t lint,test --parallel=3
      - run: npx nx build app --configuration=production
      - run: npx lighthouse-ci http://localhost:4200 --budget-path=./budgets.json

Comparison: RxJS‑Only vs NgRx Store vs SignalStore for Real‑Time

Teams often blend: global NgRx store for shared streams; SignalStore for feature‑local cohesion and component Signals. That’s my default on enterprise dashboards.

Case Snapshots From the Field

If your team needs to stabilize a chaotic codebase or modernize to Angular 20+, see how I can help you stabilize your Angular codebase via gitPlumbers—rescue chaotic code without a rewrite.

When to Hire an Angular Developer for Legacy Rescue

For modernization projects (AngularJS→Angular, RxJS 7/8, TypeScript strict), I’ve done the upgrades and shipped zero‑downtime releases. See live products and proof at NG Wave, IntegrityLens, and SageStepper.

How an Angular Consultant Approaches Signals Migration in NgRx Dashboards

// SignalStore façade example
import { signalStore, withState, patchState, withComputed, withMethods } from '@ngrx/signals';
import { inject } from '@angular/core';
import { Store } from '@ngrx/store';
import * as sel from '../data-access/telemetry.selectors';

interface VMState { filterText: string }

export const useTelemetryStore = signalStore(
  withState<VMState>({ filterText: '' }),
  withComputed((store) => ({
    points: inject(Store).selectSignal(sel.selectFilteredPoints),
    count: () => store.points().length,
  })),
  withMethods((store) => ({ setFilterText(v: string) { patchState(store, { filterText: v }); } }))
);

Takeaways and Next Steps

If you’re considering hiring Angular help for a high‑stakes dashboard—telemetry, multi‑tenant analytics, or kiosks—bring in experience early. I’ve done this work across telecom, airlines, insurance telematics, and IoT. Let’s review your state pipeline and ship a plan in a week.

FAQs: Real‑Time NgRx, WebSockets, and Hiring