feat: add the cross-machine wake for Postgres — LISTEN/NOTIFY

CHANGELOG.md

@@ -6,6 +6,33 @@ This project follows semantic versioning for the published package. Sample,
 documentation, and CI-only changes may remain in `Unreleased` until the next
 package release is useful for users.
 
+## Unreleased
+
+- **Added the cross-machine wake for the Postgres dialect — `LISTEN`/`NOTIFY`.**
+  Completes the wake tiers from 0.4.0: `new PostgresOutboxStore({ wakeChannel })`
+  piggybacks `pg_notify` on the enqueue transaction (Postgres delivers it **on
+  commit** and drops it on rollback, so the wake is atomic with the event
+  becoming visible — no post-commit discipline needed), and the new
+  `PostgresWakeListener` (at `@nest-native/messaging/postgres`) holds a
+  dedicated `LISTEN` connection that feeds the worker's `OutboxWaker`,
+  reconnecting with a fixed delay when the connection drops. Best-effort by the
+  same contract as the other tiers: notifications missed during a reconnect gap
+  are not recovered — polling remains the delivery backstop, so a lost wake only
+  costs one poll interval, never an event. Channels are allow-listed to an
+  identifier-safe charset (`LISTEN` cannot be parameterized), capped at
+  Postgres's 63-byte identifier limit (beyond it `LISTEN` silently truncates
+  while `pg_notify` RAISES — which would abort the caller's business
+  transaction), and double-quoted, keeping `LISTEN` case-sensitivity aligned
+  with `pg_notify`'s exact-string channel. The listener survives the pg
+  end-during-connect edge (a client ended mid-connect never settles its
+  `connect()` promise — the session races it against the connection's own
+  `end`/`error` events so `stop()` cannot hang), guards a throwing custom
+  `WakeSignal` from crashing the worker, validates `reconnectDelayMs`, and the
+  documented factory sets `keepAlive: true` so a half-open LISTEN socket is
+  detected by the OS. Postgres-only by nature; SQLite/MySQL use the
+  `WakeSocket` tier. Verified against real Postgres: delivered on commit,
+  dropped on rollback.
+
 ## 0.4.0 - 2026-07-19
 
 - **Added `OutboxWaker` — an in-process wake for the worker loop.** The worker

packages/messaging/README.md

@@ -111,15 +111,38 @@ switching topology — single process (`OutboxWaker`) ↔ app + worker processes
 this covers every supported deployment: processes sharing a SQLite file are by
 definition on one machine.
 
-**Across machines** the wake would have to travel through shared infrastructure
-(e.g. Postgres `LISTEN`/`NOTIFY`) — a planned, dialect-specific follow-up. Those
-workers rely on polling today.
+**Across machines** the wake travels through the one thing every worker already
+shares — the database. On the **Postgres** dialect, `LISTEN`/`NOTIFY` carries it:
+
+```ts
+// producer side — one option on the store; pg_notify rides the enqueue
+// transaction, so Postgres delivers the wake ON COMMIT and drops it on
+// rollback (the signal is atomic with the event becoming visible)
+new PostgresOutboxStore({ wakeChannel: 'outbox_wake' })
+
+// worker side — a DEDICATED (non-pooled) LISTEN connection feeds the waker
+import { PostgresWakeListener } from '@nest-native/messaging/postgres';
+
+const listener = new PostgresWakeListener({
+  connect: () => new pg.Client({ connectionString, keepAlive: true }), // fresh client per attempt
+  channel: 'outbox_wake',
+  waker, // the same OutboxWaker passed to runWorkerLoop
+});
+listener.start();
+// on shutdown: await listener.stop();
+```
+
+The listener reconnects (default every 5s) when its connection drops;
+notifications missed during the gap are not recovered — polling remains the
+backstop, exactly as with the other tiers. `LISTEN`/`NOTIFY` is Postgres-only:
+SQLite and MySQL deployments use the socket tier above (for SQLite that is the
+whole story anyway — its processes share one machine by definition).
 
 ## Status & scope
 
 - **Drivers:** SQLite (better-sqlite3, sync), Postgres (`pg`, async), and MySQL (`mysql2`, async) via per-dialect stores.
 - **Transports:** in-process (default, `@nest-native/messaging/in-process` — no broker, at-least-once via the claimer) and Kafka (`@nest-native/kafka`).
-- **Latency:** the worker drains a backlog immediately and only idles at `pollIntervalMs`; an `OutboxWaker` cuts that idle wait in-process, and the `WakeSocket` pair carries the wake across processes on the same machine (see above).
-- **Roadmap:** a cross-machine wake (Postgres `LISTEN`/`NOTIFY`); additional transports. CDC (Debezium) is an intentional non-goal — this is the app-level outbox.
+- **Latency:** the worker drains a backlog immediately and only idles at `pollIntervalMs`; the wake tiers cut that idle wait — `OutboxWaker` in-process, the `WakeSocket` pair across processes on one machine, and Postgres `LISTEN`/`NOTIFY` across machines (see above).
+- **Roadmap:** additional transports. CDC (Debezium) is an intentional non-goal — this is the app-level outbox.
 
 Part of the [nest-native](https://github.com/nest-native) family. Not affiliated with the NestJS core team. MIT licensed.

packages/messaging/dialects/postgres/index.ts

@@ -3,3 +3,4 @@
 export * from './schema';
 export * from './outbox-store';
 export * from './inbox-store';
+export * from './wake';

packages/messaging/dialects/postgres/outbox-store.ts

@@ -8,16 +8,33 @@ import type {
   ResolvedClaimerConfig,
 } from '../../interfaces';
 import { outboxEvents } from './schema';
+import { assertValidWakeChannel } from './wake';
 
 type Db = NodePgDatabase<Record<string, never>>;
 
+export interface PostgresOutboxStoreOptions {
+  /**
+   * When set, `enqueue` also runs `pg_notify(wakeChannel, '')` on the same
+   * handle — inside the caller's transaction, so Postgres delivers the wake
+   * **on commit** and drops it on rollback: the signal is atomic with the event
+   * becoming visible. Pair it with a `PostgresWakeListener` on the workers.
+   */
+  wakeChannel?: string;
+}
+
 /**
  * Postgres (node-postgres) outbox store. Every method is **asynchronous** —
  * `enqueue` awaits the insert (call it with `await` inside an async
  * `@Transactional` body), and the claimer's batch claim runs in an async
  * transaction.
  */
 export class PostgresOutboxStore implements OutboxStore {
+  constructor(private readonly options: PostgresOutboxStoreOptions = {}) {
+    if (this.options.wakeChannel !== undefined) {
+      assertValidWakeChannel(this.options.wakeChannel);
+    }
+  }
+
   async enqueue(db: unknown, input: EnqueueInput<object>): Promise<OutboxEventRow> {
     const now = new Date().toISOString();
     const [row] = await (db as Db)
@@ -34,6 +51,13 @@ export class PostgresOutboxStore implements OutboxStore {
         createdAt: now,
       })
       .returning();
+    if (this.options.wakeChannel !== undefined) {
+      // pg_notify takes the channel as a plain string parameter (safely
+      // parameterized, unlike LISTEN's identifier) — delivered on commit.
+      await (db as Db).execute(
+        sql`select pg_notify(${this.options.wakeChannel}, '')`,
+      );
+    }
     return row;
   }
 

packages/messaging/dialects/postgres/wake.ts

@@ -0,0 +1,205 @@
+import type { WakeSignal } from '../../outbox-waker';
+
+/**
+ * The cross-MACHINE wake, for the Postgres dialect: workers on other hosts
+ * can't be reached by an in-memory `OutboxWaker` or a same-machine
+ * `WakeSocket…` pair, but they all share the database — so the database carries
+ * the signal. Producers `NOTIFY` (see `PostgresOutboxStore`'s `wakeChannel`
+ * option, which piggybacks `pg_notify` on the enqueue transaction — Postgres
+ * delivers it **on commit** and drops it on rollback, so the wake is atomic
+ * with the event becoming visible); each worker holds a `LISTEN` connection
+ * that feeds its `OutboxWaker`.
+ *
+ * Best-effort by the same contract as every other wake tier: the listener
+ * reconnects with a fixed delay when its connection drops, and any
+ * notification missed during the gap is NOT recovered — polling remains the
+ * delivery backstop, so a lost wake only costs one poll interval, never an
+ * event. Correctness never depends on the wake arriving.
+ */
+
+/**
+ * The slice of `pg.Client` the listener uses — structural, so tests fake it
+ * hermetically and any LISTEN-capable client satisfies it. Must be a
+ * DEDICATED, non-pooled connection: notifications arrive only on the exact
+ * connection that issued `LISTEN`, and a pool may recycle it.
+ */
+export interface WakeListenConnection {
+  // `Promise<unknown>`, not `Promise<void>`: pg.Client's promise overloads
+  // resolve with values (`connect()` → the client), and TS's void-return
+  // leniency does not pierce the Promise generic.
+  connect(): Promise<unknown>;
+  query(text: string): Promise<unknown>;
+  end(): Promise<unknown>;
+  on(event: 'notification', listener: () => void): unknown;
+  on(event: 'error', listener: (error: unknown) => void): unknown;
+  on(event: 'end', listener: () => void): unknown;
+}
+
+export interface PostgresWakeListenerOptions {
+  /**
+   * Factory for a fresh dedicated connection per attempt — e.g.
+   * `() => new pg.Client({ connectionString, keepAlive: true })`. A factory
+   * (not an instance) because every reconnect needs a new client. Set
+   * `keepAlive: true` (pg leaves it off by default): a LISTEN connection is
+   * pure-receive, so without TCP keepalives a half-open death (NAT/firewall
+   * idle eviction, a host vanishing without FIN/RST) is never detected — the
+   * listener would sit parked forever believing it is healthy while every
+   * wake silently falls back to the polling interval.
+   */
+  connect: () => WakeListenConnection;
+  /** The wake target — the same `OutboxWaker` passed to `runWorkerLoop`. */
+  waker: WakeSignal;
+  /** NOTIFY channel to listen on; must match the producer side. Default `outbox_wake`. */
+  channel?: string;
+  /** Delay before re-connecting after a drop or failed attempt. Default 5000ms. */
+  reconnectDelayMs?: number;
+  /** Connection/listen failures are reported here instead of thrown. */
+  onError?: (error: unknown) => void;
+}
+
+// LISTEN takes an identifier (it cannot be parameterized like pg_notify's
+// string argument), so the channel is allow-listed to a safe charset and then
+// double-quoted — quoting also keeps LISTEN case-sensitive, matching the exact
+// string the producer passes to pg_notify.
+const CHANNEL_PATTERN = /^[A-Za-z_][A-Za-z0-9_]*$/;
+
+export function assertValidWakeChannel(channel: string): void {
+  // 63 = NAMEDATALEN - 1, the Postgres identifier limit. Beyond it, LISTEN
+  // silently truncates while pg_notify RAISES — which would abort the caller's
+  // business transaction on every enqueue. The charset is ASCII-only, so
+  // characters equal bytes.
+  if (!CHANNEL_PATTERN.test(channel) || channel.length > 63) {
+    throw new Error(
+      `invalid wake channel ${JSON.stringify(channel)}: must match ${String(CHANNEL_PATTERN)} and be at most 63 characters`,
+    );
+  }
+}
+
+/**
+ * Holds a dedicated `LISTEN` connection and forwards every notification on the
+ * channel to `waker.notify()`. `start()` launches a supervision loop that
+ * reconnects (with `reconnectDelayMs` between attempts) until `stop()`;
+ * failures go to `onError`, never to the caller — the worker must boot and keep
+ * polling even when Postgres is briefly unreachable.
+ */
+export class PostgresWakeListener {
+  readonly #channel: string;
+  readonly #reconnectDelayMs: number;
+  #loop: Promise<void> | null = null;
+  #current: WakeListenConnection | null = null;
+  #stopped = false;
+  #abortSleep: (() => void) | null = null;
+
+  constructor(private readonly options: PostgresWakeListenerOptions) {
+    this.#channel = options.channel ?? 'outbox_wake';
+    assertValidWakeChannel(this.#channel);
+    this.#reconnectDelayMs = options.reconnectDelayMs ?? 5_000;
+    // Fail fast on a config that would turn a Postgres outage into a hot
+    // reconnect loop (0/negative/NaN all collapse to ~immediate retries).
+    if (!Number.isFinite(this.#reconnectDelayMs) || this.#reconnectDelayMs < 1) {
+      throw new Error(
+        `invalid reconnectDelayMs ${this.#reconnectDelayMs}: must be a finite number >= 1`,
+      );
+    }
+  }
+
+  /** Launch the supervision loop (idempotent — a second call is a no-op). */
+  start(): void {
+    this.#loop ??= this.#supervise();
+  }
+
+  /** End the current connection and stop reconnecting. Awaitable; idempotent. */
+  async stop(): Promise<void> {
+    this.#stopped = true;
+    this.#abortSleep?.();
+    const current = this.#current;
+    if (current) {
+      await this.#endQuietly(current); // emits 'end' → the parked session resolves
+    }
+    await this.#loop;
+  }
+
+  async #supervise(): Promise<void> {
+    while (!this.#stopped) {
+      try {
+        await this.#session();
+      } catch (error) {
+        this.options.onError?.(error);
+      }
+      if (!this.#stopped) {
+        await this.#sleep(this.#reconnectDelayMs);
+      }
+    }
+  }
+
+  /** One connection lifetime: connect, LISTEN, park until drop/stop. */
+  async #session(): Promise<void> {
+    const client = this.options.connect();
+    this.#current = client;
+    try {
+      // Park BEFORE connect(), and race connect against it: a client that
+      // stop() ends mid-connect emits 'end' but its connect() promise NEVER
+      // settles (pg skips the connect callback once `_ending` is set) — an
+      // un-raced `await client.connect()` would deadlock stop() forever. The
+      // early 'error' listener also keeps an async connection error from
+      // becoming an uncaught event that crashes the worker.
+      let dead = false;
+      const parked = new Promise<void>((resolve, reject) => {
+        client.on('error', (error) => {
+          dead = true;
+          reject(error);
+        });
+        client.on('end', () => {
+          dead = true;
+          resolve();
+        });
+      });
+      parked.catch(() => undefined); // consumed via the awaits below; never unhandled
+      client.on('notification', () => this.#forwardWake());
+      await Promise.race([client.connect(), parked]);
+      if (dead) {
+        return; // ended (stop, or a server drop) before LISTEN could be issued
+      }
+      await client.query(`LISTEN "${this.#channel}"`);
+      await parked;
+    } finally {
+      this.#current = null;
+      await this.#endQuietly(client);
+    }
+  }
+
+  // pg emits 'notification' synchronously from its socket-data handler; a
+  // throwing custom WakeSignal must be routed to onError, not thrown into that
+  // path (an uncaught exception there kills the worker process). The stock
+  // OutboxWaker cannot throw, but WakeSignal is a public structural seam.
+  #forwardWake(): void {
+    try {
+      this.options.waker.notify();
+    } catch (error) {
+      this.options.onError?.(error);
+    }
+  }
+
+  async #endQuietly(client: WakeListenConnection): Promise<void> {
+    try {
+      await client.end();
+    } catch (error) {
+      this.options.onError?.(error);
+    }
+  }
+
+  /** Reconnect backoff that `stop()` can cut short. */
+  #sleep(ms: number): Promise<void> {
+    return new Promise((resolve) => {
+      const timer = setTimeout(() => {
+        this.#abortSleep = null;
+        resolve();
+      }, ms);
+      this.#abortSleep = () => {
+        clearTimeout(timer);
+        this.#abortSleep = null;
+        resolve();
+      };
+    });
+  }
+}

packages/messaging/test/integration/round-trip.spec.ts

@@ -14,6 +14,7 @@ import {
   outboxEvents as pgOutboxEvents,
   PostgresInboxStore,
   PostgresOutboxStore,
+  PostgresWakeListener,
 } from '../../dialects/postgres';
 
 // Gated end-to-end tests against a REAL database. They skip unless the matching
@@ -200,4 +201,46 @@ describe('Postgres round-trip (real service)', { skip: !POSTGRES_URL }, () => {
     assert.equal((seen.rows as { c: number }[])[0].c, 1);
     void pgInboxEvents;
   });
+
+  test('LISTEN/NOTIFY wake: delivered on commit, dropped on rollback', async () => {
+    const pg = await import('pg');
+    let wakes = 0;
+    const listener = new PostgresWakeListener({
+      // A dedicated client, NOT the pool: notifications arrive only on the
+      // exact connection that issued LISTEN.
+      connect: () => new pg.Client({ connectionString: POSTGRES_URL, keepAlive: true }),
+      channel: 'outbox_wake_it',
+      waker: { notify: () => (wakes += 1) },
+    });
+    listener.start();
+    try {
+      const wakeStore = new PostgresOutboxStore({ wakeChannel: 'outbox_wake_it' });
+
+      // Give the listener a beat to establish LISTEN before the first NOTIFY.
+      await new Promise((resolve) => setTimeout(resolve, 300));
+
+      // Committed enqueue → exactly one wake arrives.
+      await db.transaction(async (tx) => {
+        await wakeStore.enqueue(tx, { topic: 'wake.commit', payload: { n: 1 } });
+      });
+      const deadline = Date.now() + 5_000;
+      while (wakes === 0 && Date.now() < deadline) {
+        await new Promise((resolve) => setTimeout(resolve, 10));
+      }
+      assert.equal(wakes, 1, 'a committed enqueue must wake the listener');
+
+      // Rolled-back enqueue → Postgres drops the notification with the tx.
+      await assert.rejects(
+        db.transaction(async (tx) => {
+          await wakeStore.enqueue(tx, { topic: 'wake.rollback', payload: { n: 2 } });
+          throw new Error('force rollback');
+        }),
+        /force rollback/,
+      );
+      await new Promise((resolve) => setTimeout(resolve, 500));
+      assert.equal(wakes, 1, 'a rolled-back enqueue must NOT wake the listener');
+    } finally {
+      await listener.stop();
+    }
+  });
 });