CodingNic
Sign In Book Session

Backend Services

The Real-Time Layer

Backend Services 35 min read

The Real-Time Layer

Objectives

By the end of this chapter, you should be able to:

  • Push live positions to connected browsers over WebSocket
  • Explain why positions and alerts use two different pub/sub mechanisms inside the same server
  • Wire up Postgres LISTEN/NOTIFY correctly — including the one connection-pooling mistake that silently breaks it

💡 Why this matters: Get the LISTEN connection wrong and alerts don’t fail loudly — they just stop arriving, intermittently, once real traffic starts checking connections in and out of the pool. This lesson is as much about avoiding that bug as it is about building the feature.

Why Two Mechanisms, Not One

It would be simpler to route everything — positions, squawk alerts, overflights — through one pub/sub channel. SKYWATCH deliberately doesn’t, and the reason is a hard constraint: a Postgres NOTIFY payload is capped at 8000 bytes. A single poll cycle for a busy region can return dozens of aircraft, each with a dozen-plus fields — that JSON blows past 8000 bytes easily. Alerts, by contrast, are one small object each, and rare (an alert every few minutes at most, not every 15 seconds).

So: positions go over a plain in-process EventEmitter. The poller and the WebSocket layer live in the same Node process, so nothing needs to leave the process at all — no size limit, no serialization to Postgres and back, just a function call. Alerts go through real Postgres LISTEN/NOTIFY. That’s more machinery than an in-process emitter needs, but it buys something positions don’t need: decoupling from any single writer. If this app later ran the poller as a separate process from the API/WebSocket server (a very plausible next evolution), alerts would keep working with zero code changes, because they’re not relying on being in the same process — they’re relying on Postgres, which both processes already talk to. Positions would need a different solution at that point (Redis pub/sub, most likely) — but that’s a real cost paid only if and when the app actually needs it, not paid upfront for a scaling scenario that may never happen.

The one part of this that will bite you if you don’t know it going in: LISTEN is session-scoped. It only applies to the specific database connection it was issued on. If that connection comes from a pool, the pool can hand it back and give a different connection to the next thing that asks — and now you’re listening on a connection nobody’s using and missing notifications on the one your code thinks it’s using. The listener needs its own dedicated pg.Client, created once, never returned to any pool, for the lifetime of the process.

Positions: In-Process Pub/Sub

ts
// apps/server/src/events/bus.ts
import { EventEmitter } from "node:events";
import type { AircraftState } from "@skywatch/shared";

export interface PositionsEvent {
  regionId: number;
  ts: number;
  aircraft: AircraftState[];
}

class SkywatchEventBus extends EventEmitter {}
export const eventBus = new SkywatchEventBus();
eventBus.setMaxListeners(50); // default of 10 warns once more than a handful of WS clients connect

export function emitPositions(event: PositionsEvent) {
  eventBus.emit("positions", event);
}

export function onPositions(listener: (event: PositionsEvent) => void) {
  eventBus.on("positions", listener);
  return () => eventBus.off("positions", listener); // caller gets an unsubscribe function back
}

Returning the unsubscribe function directly from onPositions (instead of making every caller separately remember the exact listener reference to pass to .off()) is a small ergonomic choice that pays off in the WebSocket handler below, where it becomes a one-liner in the socket’s close handler.

Wire it into the poller from the previous lesson — one line, right after the fetch and before the (slower) DB write, so the live feed isn’t delayed by storage:

ts
// apps/server/src/poller/loop.ts (inside the per-region try block, right after fetchPoint)
import { emitPositions } from "../events/bus.js";
// ...
const data = await fetchPoint(region.lat, region.lon, region.radiusNm);
const aircraft = data.ac ?? [];
emitPositions({ regionId: region.id, ts: Date.now(), aircraft });
const stored = await storeSnapshots(aircraft);

Alerts: Postgres LISTEN/NOTIFY

ts
// apps/server/src/db/notify.ts
import { pool } from "./client.js";

/**
 * pg_notify() -- unlike a raw NOTIFY statement -- accepts parameters safely
 * and can be issued from any pooled connection (only LISTEN requires the
 * dedicated connection; see pgListener.ts).
 */
export async function notify(channel: string, payload: unknown): Promise<void> {
  const json = JSON.stringify(payload);
  if (json.length > 7800) {
    throw new Error(`notify payload for channel "${channel}" is ${json.length} bytes, too close to Postgres's 8000-byte NOTIFY limit`);
  }
  await pool.query("SELECT pg_notify($1, $2)", [channel, json]);
}

export const CHANNELS = { squawkAlert: "squawk_alert", overflight: "overflight" } as const;

The 7800-byte guard is deliberate headroom under the real 8000-byte limit — a throw here during development is far more useful than a silently-dropped notification in production because someone’s alert payload grew a field and tipped it over.

Both the listener below and the WebSocket endpoint later in this lesson need to agree on the shape of a squawk alert and an overflight event, and — once the endpoint exists — on the shape of every message type the socket can send at all. All three belong in one shared file:

ts
// packages/shared/src/events.ts
import type { AircraftState } from "./aircraft.js";

export interface SquawkAlertEvent {
  id: number;
  hex: string;
  flight: string | null;
  squawk: string; // "7500" | "7600" | "7700"
  lat: number | null;
  lon: number | null;
  altBaroFt: number | null;
  detectedAt: string; // ISO timestamp
}

export interface OverflightEvent {
  id: number;
  hex: string;
  flight: string | null;
  lat: number;
  lon: number;
  altBaroFt: number | null;
  distanceNm: number;
  detectedAt: string;
}

/**
 * Messages the WebSocket layer below pushes to connected browsers.
 * `positions` is the live feed; `squawk_alert` and `overflight` are pushed
 * the moment this lesson's LISTEN/NOTIFY plumbing delivers one.
 */
export type ServerToClientMessage =
  | { type: "connected"; ts: number }
  | { type: "positions"; ts: number; regionId: number; aircraft: AircraftState[] }
  | { type: "squawk_alert"; event: SquawkAlertEvent }
  | { type: "overflight"; event: OverflightEvent };

Add export * from "./events.js"; to packages/shared/src/index.ts. SquawkAlertEvent and OverflightEvent are needed immediately, by the listener below; ServerToClientMessage isn’t needed until “The WebSocket Endpoint” further down, but all three live in the same file since the union is defined in terms of the other two, and splitting them apart would just mean two files always changing together.

ts
// apps/server/src/db/pgListener.ts
import { EventEmitter } from "node:events";
import { Client } from "pg";
import type { OverflightEvent, SquawkAlertEvent } from "@skywatch/shared";
import { env } from "../env.js";
import { CHANNELS } from "./notify.js";

class PgListenerBus extends EventEmitter {}
export const pgListenerBus = new PgListenerBus();
pgListenerBus.setMaxListeners(100);

let client: Client | null = null;

export async function startPgListener(): Promise<void> {
  client = new Client({ connectionString: env.DATABASE_URL }); // NOT the pool -- see below
  await client.connect();

  client.on("notification", (msg) => {
    if (!msg.payload) return;
    try {
      const payload = JSON.parse(msg.payload) as SquawkAlertEvent | OverflightEvent;
      pgListenerBus.emit(msg.channel, payload);
    } catch (err) {
      console.error(`[pg-listener] failed to parse payload on channel "${msg.channel}":`, err);
    }
  });

  client.on("error", (err) => {
    // A dropped LISTEN connection means alerts silently stop flowing until
    // the process restarts -- surface loudly rather than swallow it.
    console.error("[pg-listener] connection error:", err);
  });

  await client.query(`LISTEN ${CHANNELS.squawkAlert}`);
  await client.query(`LISTEN ${CHANNELS.overflight}`);
  console.log(`[pg-listener] listening on "${CHANNELS.squawkAlert}", "${CHANNELS.overflight}"`);
}

export async function stopPgListener(): Promise<void> {
  if (client) {
    await client.end();
    client = null;
  }
}

new Client(...) here, not db/pool from db/client.ts — this is the one line in the whole file that matters most. A new pg.Client is a single dedicated connection; pool (a pg.Pool) hands out connections from a shared pool and can silently swap the underlying connection between queries. If you wrote pool.query("LISTEN ...") instead, it would appear to work in a quick manual test (the query succeeds!) and then intermittently, unpredictably stop delivering notifications once real traffic starts checking connections in and out of that same pool — exactly the kind of bug that’s miserable to track down because it doesn’t fail consistently. Get this right from the start rather than debugging it later.

Then re-export notify() from store.ts’s alert-writing functions (the previous lesson’s storeSquawkAlerts/storeOverflights), adding one await notify(...) call after each .returning():

ts
// apps/server/src/poller/store.ts (storeSquawkAlerts, updated)
import { CHANNELS, notify } from "../db/notify.js";
import type { SquawkAlertEvent } from "@skywatch/shared";

export async function storeSquawkAlerts(detections: SquawkDetection[]): Promise<SquawkAlertEvent[]> {
  const events: SquawkAlertEvent[] = [];
  for (const d of detections) {
    const [row] = await db.insert(squawkAlerts).values({ hex: d.hex, flight: d.flight, squawk: d.squawk, lat: d.lat, lon: d.lon, altBaroFt: d.altBaroFt }).returning();
    const event: SquawkAlertEvent = { id: row.id, hex: row.hex, flight: row.flight, squawk: row.squawk, lat: row.lat, lon: row.lon, altBaroFt: row.altBaroFt, detectedAt: row.detectedAt.toISOString() };
    events.push(event);
    await notify(CHANNELS.squawkAlert, event);
  }
  return events;
}

Apply the same shape to storeOverflights with CHANNELS.overflight. Both now return the created events (not just write them) — the poller’s log lines can use the returned events directly instead of the detections.

The WebSocket Endpoint

bash
cd apps/server
npm install @fastify/websocket
ts
// apps/server/src/ws/register.ts
import type { FastifyInstance } from "fastify";
import type { OverflightEvent, ServerToClientMessage, SquawkAlertEvent } from "@skywatch/shared";
import { onPositions } from "../events/bus.js";
import { pgListenerBus } from "../db/pgListener.js";
import { CHANNELS } from "../db/notify.js";

export function registerWebSocket(app: FastifyInstance): void {
  app.get("/ws", { websocket: true }, (socket) => {
    const send = (msg: ServerToClientMessage) => {
      if (socket.readyState === socket.OPEN) socket.send(JSON.stringify(msg));
    };

    send({ type: "connected", ts: Date.now() });

    const unsubscribePositions = onPositions((event) => {
      send({ type: "positions", ts: event.ts, regionId: event.regionId, aircraft: event.aircraft });
    });

    const onSquawkAlert = (event: SquawkAlertEvent) => send({ type: "squawk_alert", event });
    const onOverflight = (event: OverflightEvent) => send({ type: "overflight", event });
    pgListenerBus.on(CHANNELS.squawkAlert, onSquawkAlert);
    pgListenerBus.on(CHANNELS.overflight, onOverflight);

    socket.on("close", () => {
      unsubscribePositions();
      pgListenerBus.off(CHANNELS.squawkAlert, onSquawkAlert);
      pgListenerBus.off(CHANNELS.overflight, onOverflight);
    });

    socket.on("error", (err) => app.log.warn({ err }, "websocket client error"));
  });
}

One /ws endpoint multiplexes all three message types — positions tagged with regionId (so a client can filter to whichever region(s) it cares about; a later split-screen compare view leans on exactly this), plus the two alert types. There’s no server-side per-client subscription filtering: every connected client gets every watched region’s traffic and filters on the frontend. That’s a genuine simplification, correct at this app’s scale (a handful of regions, not hundreds of concurrently-watched airspaces across many clients) — worth knowing as a scaling limit, not worth solving prematurely.

The close handler unsubscribing from all three sources is what prevents a slow leak: without it, every WebSocket connection that ever opened and closed would leave its listener attached to eventBus/pgListenerBus forever, and emitPositions/pgListenerBus.emit would keep calling send() on a socket that’s long gone.

Wire both into index.ts:

ts
// apps/server/src/index.ts (add)
import websocket from "@fastify/websocket";
import { registerWebSocket } from "./ws/register.js";
import { startPgListener, stopPgListener } from "./db/pgListener.js";
// ...
await app.register(websocket);
registerWebSocket(app);
await startPgListener();
// in shutdown(), before app.close():
await stopPgListener();

Try It

  1. Start the server, then connect with any WebSocket client (wscat -c ws://localhost:4000/ws, or a browser console: new WebSocket("ws://localhost:4000/ws").onmessage = e => console.log(JSON.parse(e.data))).
  2. Confirm you see a {"type":"connected",...} message immediately, then a {"type":"positions",...} message roughly every 15 seconds as the poller cycles.
  3. To confirm the LISTEN/NOTIFY half specifically (rather than just positions), trigger a manual notification directly in psql while your WebSocket client is connected:
    sql
    SELECT pg_notify('squawk_alert', '{"id":1,"hex":"test","flight":null,"squawk":"7700","lat":null,"lon":null,"altBaroFt":null,"detectedAt":"2024-01-01T00:00:00.000Z"}');
  4. Confirm it arrives over the WebSocket within milliseconds. If it doesn’t arrive, the most likely cause is exactly the pooling mistake called out above — double check pgListener.ts is using new Client(...), not pool.

Recap

  • Positions travel over a plain in-process EventEmitter because a busy region’s aircraft data would blow past Postgres’s 8000-byte NOTIFY payload limit; alerts travel over real LISTEN/NOTIFY because they’re small, rare, and benefit from not being tied to the poller running in the same process.
  • LISTEN is session-scoped: the listener needs its own dedicated pg.Client, created once and never returned to a pool, for the life of the process. Using pool.query("LISTEN ...") instead looks like it works, then silently and intermittently stops delivering notifications once real traffic starts cycling connections through the pool.
  • One /ws endpoint multiplexes positions (tagged with regionId) and both alert types; there’s no server-side per-client filtering — every client gets every watched region’s traffic and filters on the frontend.
  • The socket’s close handler unsubscribes from all three event sources. Skip it, and every connection that ever opens and closes leaks a listener on eventBus/pgListenerBus forever.

Next lesson: bootstrapping the Next.js frontend, the store that holds live aircraft state, and the first Leaflet map showing it on screen.