Directing the Real-Time Layer

Objectives

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

• Direct an AI assistant to push live positions to connected browsers over WebSocket
• Explain why positions and alerts need two different pub/sub mechanisms in the same server, well enough to catch an AI assistant that collapses them into one
• Verify Postgres LISTEN/NOTIFY is wired up correctly, including the one connection-pooling mistake that breaks it without ever throwing an error

💡 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. An AI assistant has no reason to know this failure mode exists unless you tell it, and reviewing its output here means specifically checking for the one line that prevents it.

Why This Is the Lesson to Slow Down On

It would be simpler to prompt for one pub/sub channel and route positions, squawk alerts, and overflights all through it. Don’t let an AI assistant simplify it that way, because the constraint forcing two mechanisms is real: a Postgres NOTIFY payload is capped at 8000 bytes, and a single poll cycle for a busy region can return dozens of aircraft, each with a dozen-plus fields, which blows past that limit easily. Alerts are the opposite shape: one small object each, rare, maybe one every few minutes. So positions travel over a plain in-process EventEmitter, since the poller and the WebSocket layer live in the same Node process and nothing needs to leave it at all, and alerts travel over real Postgres LISTEN/NOTIFY, which buys decoupling from any single writer at the cost of more machinery than an in-process emitter needs.

That second mechanism has a sharp edge an AI assistant will not know about unless the prompt says so directly: LISTEN is session-scoped, it only applies to the specific database connection it was issued on. Module 1’s client.ts exports both pool and db for exactly this reason, so that a pooled connection is always available for normal queries while a separate, dedicated connection stays free for the one thing pooling breaks. If an AI assistant reaches for pool.query("LISTEN ...") because pool is the connection object already in scope, the query succeeds, in a quick manual test everything looks fine, and then intermittently, unpredictably, notifications stop arriving once real traffic starts cycling connections through that same pool. This is worth naming explicitly in the prompt rather than trusting the AI to infer it, because the whole point is that it fails silently, not loudly.

The Prompt

What It Built

// 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
}

Wired into the poller loop, one line right after the fetch resolves and before the slower DB write:

// apps/server/src/poller/loop.ts (inside the per-region try block)
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);

// 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;

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

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

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

export type ServerToClientMessage =
  | { type: "connected"; ts: number }
  | { type: "positions"; ts: number; regionId: number; aircraft: AircraftState[] }
  | { type: "squawk_alert"; event: SquawkAlertEvent }
  | { type: "overflight"; event: OverflightEvent };

// 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) => {
    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;
  }
}

// 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;
}

(storeOverflights gets the same shape, with CHANNELS.overflight.)

// 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"));
  });
}

Wired into index.ts: register @fastify/websocket, call registerWebSocket(app), await startPgListener() after routes register, and await stopPgListener() inside the shutdown handler, before app.close().

Review This

Did the listener actually use a dedicated Client, or did it reach for the pooled pool? This is the one mistake in this entire module that produces zero errors anywhere while being completely broken. pool.query("LISTEN squawk_alert") is a perfectly valid query, it resolves successfully, and if you test it once, right after startup, with nothing else hitting the database, it can even appear to work, because the connection the pool happened to hand out for that query is still sitting idle and still technically listening. The failure only shows up once real traffic starts checking other connections in and out of the same pool, at which point the pool is free to hand that exact connection to some unrelated query, and your listener is now subscribed on a connection nobody is using while a completely different connection handles everything else. No exception, no log line, alerts just stop arriving. Check pgListener.ts for new Client({ connectionString: env.DATABASE_URL }), not pool imported from client.ts, and confirm that client is never passed into or returned to anything pool-managed. If it’s using pool, the fix is: “pgListener.ts is issuing LISTEN against the pooled pool export, it needs its own dedicated pg.Client, created and connected once, that’s never returned to a pool, for the life of the process.”

Is emitPositions actually called before the database write, or did it get placed after? The prompt asked for positions to reach the browser before the slower storage write happens, but it’s easy for an AI assistant to place emitPositions(...) after await storeSnapshots(aircraft) instead of before it, since both orderings produce identical behavior in isolation and neither one fails a test that just checks “did the WebSocket eventually receive a positions message.” The difference only shows up as added latency on every single poll cycle, the live feed silently waiting on a batched database insert it never needed to wait on, which defeats part of the reason positions use an in-process emitter at all instead of a slower pub/sub path. Check that emitPositions is called immediately after fetchPoint resolves, before storeSnapshots is awaited. If it’s in the wrong order, the fix is: “emitPositions needs to run right after the adsb.lol fetch resolves, before the storeSnapshots call, so the live feed to the browser isn’t delayed by the database write.”

Does the socket’s close handler actually unsubscribe from all three sources, or just some of them? It’s easy to remember to unsubscribe from eventBus (the thing most obviously tied to “the WebSocket connection”) and forget that pgListenerBus has two separate listeners registered too, one for each alert channel. A version that only calls unsubscribePositions() in the close handler still runs fine and still looks correct for as long as you’re only testing one WebSocket connection at a time, since nothing crashes when a listener sits around unused. It becomes a real leak the moment browsers open and close a lot of connections over the server’s lifetime, each one leaving two dangling listeners on pgListenerBus that never get cleaned up, and pgListenerBus.emit(...) keeps calling send() on sockets that are long gone. Check the close handler calls all three, unsubscribePositions(), pgListenerBus.off(CHANNELS.squawkAlert, onSquawkAlert), and pgListenerBus.off(CHANNELS.overflight, onOverflight). If any are missing, the fix is: “the WebSocket close handler is only unsubscribing from some of the three event sources it subscribed to on open, it needs to unsubscribe from eventBus and both pgListenerBus channels or every closed connection leaks a listener.”

Try It

1. Run the prompt above against your AI coding assistant, with the poller and detection engine from the previous lesson already running.
2. Read pgListener.ts, loop.ts, and ws/register.ts against the three checks above before connecting anything.
3. Start the server, then connect with a 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))).
4. Confirm you see a {"type":"connected",...} message immediately, then a {"type":"positions",...} message roughly every fifteen seconds as the poller cycles.
5. Confirm the LISTEN/NOTIFY half specifically, not just positions, by triggering a manual notification directly in psql while your WebSocket client is connected:
   sqlCopy

   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"}');
   

6. Confirm it arrives over the WebSocket within milliseconds. If it doesn’t arrive at all, and no error appeared anywhere, that’s the signature of the pooled-connection mistake above, go straight to pgListener.ts and check for new Client(...) versus pool.
7. Open and close several WebSocket connections in a row, then check the server logs or a debugger for growing listener counts on eventBus/pgListenerBus to confirm the close handler is actually cleaning up.

Recap

• Positions use an in-process EventEmitter because the poller and WebSocket server share a process and a busy region’s data would blow past Postgres’s 8000-byte NOTIFY limit; alerts use real LISTEN/NOTIFY because they’re small, rare, and shouldn’t be tied to the poller staying in the same process forever.
• LISTEN is session-scoped and needs its own dedicated pg.Client, never returned to a pool. Using the pooled pool export instead is the one mistake in this module that produces no error at all, it just stops working intermittently once real traffic exists. Module 1’s client.ts exports both pool and db for exactly this reason.
• emitPositions belongs right after the fetch resolves, before the slower database write, so the live feed isn’t waiting on storage it doesn’t need.
• The WebSocket close handler has three things to unsubscribe from, not one. Missing any of them is a slow leak that won’t show up until a lot of connections have opened and closed.

Next module: directing the frontend foundations, the design-token system, and the first live map.