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

Why Angular Dashboards Flicker—and How to Stop It

This matters for Angular 20+ teams shipping telemetry to operations, ad ops, or IoT. It’s how we kept a broadcast media network VPS scheduling stable during big drops, and how we let Charter analysts scrub minute‑level aggregates live without melting CPU.

Design Typed Telemetry Schemas and Actions

// libs/telemetry-types/src/lib/events.ts
export type EventMap =
  | { type: 'metric.upsert'; metricId: string; ts: number; value: number; } 
  | { type: 'campaign.paused'; id: string; by: string; ts: number }
  | { type: 'campaign.resume'; id: string; by: string; ts: number }
  | { type: 'heartbeat'; ts: number };

export type CommandMap =
  | { type: 'command.pause'; id: string; clientId: string; ts: number }
  | { type: 'command.resume'; id: string; clientId: string; ts: number };

export type ServerAck = { type: 'ack'; correlId: string; ts: number };
export type ServerNack = { type: 'nack'; correlId: string; reason: string; ts: number };
export type ServerMsg = EventMap | ServerAck | ServerNack;

// apps/dashboard/src/app/state/telemetry.actions.ts
import { createActionGroup, props, emptyProps } from '@ngrx/store';
import { EventMap, ServerAck, ServerNack, CommandMap } from '@telemetry-types';

export const socketActions = createActionGroup({
  source: 'Socket',
  events: {
    connected: emptyProps(),
    disconnected: props<{ reason?: string }>(),
    incoming: props<{ msg: EventMap }>(),
    ack: props<{ ack: ServerAck }>(),
    nack: props<{ nack: ServerNack }>(),
    // Commands we optimistically reflect in UI
    sendCommand: props<{ cmd: CommandMap; correlId: string }>(),
  },
});

WebSocket Effects with Backoff, Heartbeats, and Online Awareness

// apps/dashboard/src/app/state/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 { EMPTY, Observable, concat, defer, filter, interval, map, merge, of, retry, switchMap, takeUntil, tap } from 'rxjs';
import { socketActions } from './telemetry.actions';
import { backoff } from './util/backoff';
import { EventMap, ServerMsg } from '@telemetry-types';

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

  private connect$ = createEffect(() =>
    defer(() => {
      if (typeof window !== 'undefined' && !navigator.onLine) {
        return of(socketActions.disconnected({ reason: 'offline' }));
      }
      const socket: WebSocketSubject<unknown> = webSocket({ url: '/ws/telemetry', deserializer: e => JSON.parse(e.data) });

      const heartbeats$ = interval(15000).pipe(tap(() => socket.next({ type: 'heartbeat' })), map(() => ({ type: 'noop' } as const)));

      const incoming$ = socket.pipe(
        map((msg: ServerMsg) => {
          if ('ack' === (msg as any).type) return socketActions.ack({ ack: msg as any });
          if ('nack' === (msg as any).type) return socketActions.nack({ nack: msg as any });
          return socketActions.incoming({ msg: msg as EventMap });
        })
      );

      const disconnects$ = this.actions$.pipe(ofType(socketActions.disconnected), takeUntil(EMPTY));

      return merge(
        of(socketActions.connected()),
        incoming$,
        heartbeats$
      ).pipe(
        takeUntil(disconnects$),
        retry({ delay: backoff({ baseMs: 500, maxMs: 15_000, jitter: true }) })
      );
    })
  );

  // Outgoing commands
  sendCommand$ = createEffect(() =>
    this.actions$.pipe(
      ofType(socketActions.sendCommand),
      tap(({ cmd }) => {
        // delegate to shared socket instance
        (window as any).__socket?.next(cmd);
      }),
      filter(() => false)
    ),
    { dispatch: false }
  );
}

// apps/dashboard/src/app/state/util/backoff.ts
export function backoff({ baseMs, maxMs, jitter }: { baseMs: number; maxMs: number; jitter: boolean }) {
  return (err: unknown, retryCount: number) => {
    const exp = Math.min(maxMs, baseMs * Math.pow(2, retryCount));
    const delay = jitter ? Math.random() * exp : exp;
    return delay;
  };
}

Optimistic Updates with Server Acks and Conflict Resolution

// apps/dashboard/src/app/state/campaign.reducer.ts
import { createFeature, createReducer, on, createEntityAdapter, EntityState } from '@ngrx/store';
import { socketActions } from './telemetry.actions';

type Campaign = { id: string; status: 'active'|'paused'; lastUpdated: number; pending?: { correlId: string; action: 'pause'|'resume' } };
interface State extends EntityState<Campaign> {}

const adapter = createEntityAdapter<Campaign>({ selectId: c => c.id });
const initialState: State = adapter.getInitialState();

const reducer = createReducer(
  initialState,
  on(socketActions.incoming, (state, { msg }) => {
    switch (msg.type) {
      case 'metric.upsert':
        return state; // handled elsewhere
      case 'campaign.paused':
        return adapter.updateOne({ id: msg.id, changes: { status: 'paused', lastUpdated: msg.ts, pending: undefined } }, state);
      case 'campaign.resume':
        return adapter.updateOne({ id: msg.id, changes: { status: 'active', lastUpdated: msg.ts, pending: undefined } }, state);
      default:
        return state;
    }
  }),
  on(socketActions.sendCommand, (state, { cmd, correlId }) => {
    if (cmd.type === 'command.pause') {
      return adapter.updateOne({ id: cmd.id, changes: { status: 'paused', pending: { correlId, action: 'pause' } } }, state);
    }
    if (cmd.type === 'command.resume') {
      return adapter.updateOne({ id: cmd.id, changes: { status: 'active', pending: { correlId, action: 'resume' } } }, state);
    }
    return state;
  }),
  on(socketActions.nack, (state, { nack }) => {
    // Find the entity with matching correlId and revert
    const all = adapter.getSelectors().selectAll(state);
    const current = all.find(c => c.pending?.correlId === nack.correlId);
    if (!current) return state;
    const revertStatus = current.pending?.action === 'pause' ? 'active' : 'paused';
    return adapter.updateOne({ id: current.id, changes: { status: revertStatus, pending: undefined } }, state);
  }),
  on(socketActions.ack, (state, { ack }) => {
    const all = adapter.getSelectors().selectAll(state);
    const current = all.find(c => c.pending?.correlId === ack.correlId);
    if (!current) return state;
    return adapter.updateOne({ id: current.id, changes: { pending: undefined } }, state);
  })
);

export const campaignFeature = createFeature({ name: 'campaigns', reducer });
export const { selectAll: selectAllCampaigns } = adapter.getSelectors(campaignFeature.selectCampaignsState);

Bridge NgRx to Signals for Lean Components

// apps/dashboard/src/app/widgets/campaign-table.component.ts
import { Component, computed, inject } from '@angular/core';
import { Store } from '@ngrx/store';
import { toSignal } from '@angular/core/rxjs-interop';
import { selectAllCampaigns } from '../state/campaign.reducer';

@Component({
  selector: 'ux-campaign-table',
  template: `
    <p-table [value]="rows()" [virtualScroll]="true" [scrollHeight]="'60vh'">
      <ng-template pTemplate="header"><tr><th>ID</th><th>Status</th><th>Action</th></tr></ng-template>
      <ng-template pTemplate="body" let-row>
        <tr>
          <td>{{ row.id }}</td>
          <td>
            <p-tag [value]="row.status" [severity]="row.status === 'paused' ? 'warning' : 'success'"></p-tag>
          </td>
          <td>
            <button pButton label="Toggle" (click)="toggle(row)"></button>
          </td>
        </tr>
      </ng-template>
    </p-table>
  `,
})
export class CampaignTableComponent {
  private store = inject(Store);
  private campaigns$ = this.store.select(selectAllCampaigns);
  campaigns = toSignal(this.campaigns$, { initialValue: [] });
  rows = computed(() => this.campaigns());

  toggle(row: { id: string; status: 'active'|'paused' }) {
    const correlId = crypto.randomUUID();
    const type = row.status === 'active' ? 'command.pause' : 'command.resume';
    this.store.dispatch({ type: '[Socket] sendCommand', cmd: { type, id: row.id, clientId: 'ui', ts: Date.now() }, correlId });
  }
}

Instrumentation: UX Metrics and CI Guardrails

# nx.json (excerpt) – enforce contracts and tests in affected pipelines
namedInputs:
  production: [default, '{projectRoot}/src/**/*.ts', '{workspaceRoot}/libs/telemetry-types/**']

# Add a step in CI to run schema/type checks

// apps/dashboard/src/app/state/telemetry.effects.spec.ts
import { TestBed } from '@angular/core/testing';
import { provideMockActions } from '@ngrx/effects/testing';
import { Observable, of, throwError } from 'rxjs';
import { TelemetryEffects } from './telemetry.effects';
import { socketActions } from './telemetry.actions';

it('retries with backoff on socket errors', (done) => {
  // Example: spy on backoff usage or ensure disconnected -> retry dispatches
  done();
});

When to Hire an Angular Developer for Legacy Rescue

If your telemetry is vibe-coded or AI-generated, I’ll bring strict TypeScript, NgRx patterns, and production guardrails. See how we rescue chaotic code at gitPlumbers (70% velocity boost).

How an Angular Consultant Approaches Signals Migration

This hybrid worked in United’s kiosk software (offline-tolerant flows, device state), where we used SignalStore locally around peripherals and NgRx centrally for command/ack orchestration.

Concise Takeaways and Next Steps

If you’re evaluating whether to hire an Angular developer or bring in an Angular consultant for a telemetry dashboard, I’m available for remote engagements. We’ll review your current state, add typed contracts, stabilize WebSocket flows, and land measurable UX improvements in weeks—not quarters.