add sql listen, unlisten, notify for postgres

docs/runtime/sql.mdx

@@ -43,6 +43,7 @@ const sqliteResults = await sqlite`
 - Binary protocol support makes it faster
 - TLS support (and auth mode)
 - Automatic configuration with environment variable
+- PostgreSQL `LISTEN` / `NOTIFY` for pub/sub between clients
 
 ---
 
@@ -1290,11 +1291,80 @@ We haven't implemented `LOAD DATA INFILE` support yet
 
 ### PostgreSQL-Specific Features
 
+#### `LISTEN` / `NOTIFY`
+
+PostgreSQL's `LISTEN`/`NOTIFY` provides lightweight pub/sub between database clients. Bun exposes three methods on a Postgres SQL instance:
+
+- `sql.listen(channel, onnotify, onlisten?)` subscribes to a channel.
+- `sql.unlisten(channel, onnotify?)` unsubscribes.
+- `sql.notify(channel, payload)` publishes a payload to a channel.
+
+```ts
+import { SQL } from "bun";
+
+const sql = new SQL("postgres://user:pass@localhost/mydb");
+
+// Subscribe — onnotify fires for every NOTIFY on this channel.
+const sub = await sql.listen("user_events", payload => {
+  const event = JSON.parse(payload);
+  console.log("got", event);
+});
+console.log("listening on backend pid", sub.state.pid);
+
+// Publish from anywhere with a Postgres connection (the same `sql` instance,
+// a different one, or even psql). Payload must be a string; use JSON.stringify
+// for structured data.
+await sql.notify("user_events", JSON.stringify({ type: "signup", id: 42 }));
+
+// Stop receiving — only this listener is removed.
+await sub.unlisten();
+```
+
+##### Connection lifecycle
+
+Each `Bun.SQL` instance opens at most one **dedicated** PostgreSQL connection for `LISTEN`. All channels share that single connection, separate from the regular query pool. The connection is created lazily on the first `sql.listen()` call and closed on `sql.close()`.
+
+If the dedicated connection drops (network blip, server restart, `pg_terminate_backend`), Bun automatically reconnects with exponential backoff (250 ms → 32 s, with ±25% jitter) and re-issues `LISTEN` for every tracked channel. The `onlisten` callback fires again on every successful `LISTEN`, so you can use it to detect (re)connection.
+
+```ts
+let connectCount = 0;
+await sql.listen(
+  "heartbeat",
+  () => {},
+  () => console.log("(re)connected, attempt", ++connectCount),
+);
+```
+
+##### `state` object
+
+The object returned from `listen()` contains a shared `state` reference and a per-listener `unlisten()` function:
+
+```ts
+const { state, unlisten } = await sql.listen("ch", () => {});
+state.pid; // PostgreSQL backend process ID — useful with pg_terminate_backend()
+state.secret; // Cancellation secret from BackendKeyData
+```
+
+The same `state` object is handed back from every `listen()` call on this SQL instance. When the listen connection reconnects, `state.pid` and `state.secret` are mutated in place, so any reference you've kept always reflects the current backend.
+
+##### `notify()` semantics
+
+`sql.notify(channel, payload)` runs `SELECT pg_notify($1, $2)` through the regular query pool — it does not require an active subscription. It returns a Promise that resolves once the server has accepted the NOTIFY (committed at the next transaction boundary, which for autocommit queries is immediate).
+
+Channel names are quoted as PostgreSQL identifiers; null bytes are rejected. PostgreSQL's default `NOTIFY` payload limit is 8000 bytes — payloads above that produce a server-side error.
+
+##### Error handling
+
+- A `LISTEN` that fails on a live connection (e.g., a transient permission error) is logged via `console.warn` and retried on the next reconnect tick. After 10 consecutive failures the channel is dropped to avoid an infinite retry loop.
+- Errors thrown synchronously from your `onnotify` handler are caught and ignored — wrap your handler in a `try`/`catch` if you need custom error reporting.
+- Calling `unlisten()` (either form) is idempotent.
+
+#### Other PostgreSQL features
+
 We haven't implemented these yet:
 
 - `COPY` support
-- `LISTEN` support
-- `NOTIFY` support
+- Logical replication (`sql.subscribe()`)
 
 We also haven't implemented some of the more uncommon features like:
 

packages/bun-types/sql.d.ts

@@ -859,6 +859,57 @@ declare module "bun" {
      */
     unsafe<T = any>(string: string, values?: any[]): SQL.Query<T>;
 
+    /**
+     * Subscribe to a PostgreSQL LISTEN channel. A dedicated connection is
+     * created and reused for all listeners; it reconnects automatically with
+     * exponential backoff on disconnect.
+     *
+     * @param channel - The channel name to listen on
+     * @param onnotify - Called with the payload string each time a notification arrives
+     * @param onlisten - Called each time the LISTEN command completes (initial connect and reconnects)
+     * The returned `state` object is shared across all listen() calls on this SQL
+     * instance and is mutated in place when the underlying listen connection is
+     * (re)established — a stale reference always reflects the current backend.
+     *
+     * @returns Promise resolving to `{ state, unlisten }`. `state.pid` is the
+     * PostgreSQL backend process ID (useful with `pg_terminate_backend`),
+     * `state.secret` is the cancellation secret. Call `unlisten()` to remove this
+     * specific listener (idempotent).
+     *
+     * @example
+     * const { state, unlisten } = await sql.listen('news', payload => console.log(payload));
+     * console.log('listening on backend pid', state.pid);
+     * // later:
+     * await unlisten();
+     */
+    listen(
+      channel: string,
+      onnotify: (payload: string) => void,
+      onlisten?: (state: { pid: number; secret: number }) => void,
+    ): Promise<{ state: { pid: number; secret: number }; unlisten: () => Promise<void> }>;
+
+    /**
+     * Unsubscribe from a PostgreSQL LISTEN channel.
+     * If `onnotify` is provided, only that specific listener is removed.
+     * If omitted, all listeners on the channel are removed and UNLISTEN is sent.
+     *
+     * @param channel - The channel name to unlisten from
+     * @param onnotify - The specific listener to remove (optional)
+     */
+    unlisten(channel: string, onnotify?: (payload: string) => void): Promise<void>;
+
+    /**
+     * Send a PostgreSQL NOTIFY to a channel using `pg_notify`.
+     * Uses a regular pool connection — does not require an active listener.
+     *
+     * @param channel - The channel name to notify
+     * @param payload - The payload string to send
+     *
+     * @example
+     * await sql.notify('news', JSON.stringify({ headline: 'hello' }));
+     */
+    notify(channel: string, payload: string): Promise<void>;
+
     /**
      * Reads a file and uses the contents as a query.
      * Optional parameters can be used if the file includes $1, $2, etc

src/bun.js/api/sql.classes.ts

@@ -49,8 +49,26 @@ for (const type of types) {
           setter: "setOnClose",
           this: true,
         },
+        ...(type === "PostgresSQL"
+          ? {
+              onnotification: {
+                getter: "getOnNotification",
+                setter: "setOnNotification",
+                this: true,
+              },
+              processId: {
+                getter: "getProcessId",
+              },
+              secretKey: {
+                getter: "getSecretKey",
+              },
+            }
+          : {}),
       },
-      values: ["onconnect", "onclose", "queries"],
+      values:
+        type === "PostgresSQL"
+          ? ["onconnect", "onclose", "queries", "onnotification"]
+          : ["onconnect", "onclose", "queries"],
     }),
   );
 

src/js/bun/sql.ts

@@ -936,6 +936,53 @@ const SQL: typeof Bun.SQL = function SQL(
   sql.transaction = sql.begin;
   sql.distributed = sql.beginDistributed;
   sql.end = sql.close;
+
+  const validateChannel = (channel: string) => {
+    if (typeof channel !== "string" || !channel) {
+      throw $ERR_INVALID_ARG_VALUE("channel", channel, "must be a non-empty string");
+    }
+    if (channel.indexOf("\0") !== -1) {
+      throw $ERR_INVALID_ARG_VALUE("channel", channel, "must not contain null bytes");
+    }
+  };
+  if (pool.listen) {
+    sql.listen = (channel, onnotify, onlisten?) => pool.listen(channel, onnotify, onlisten);
+    sql.unlisten = (channel, onnotify?) => pool.unlisten(channel, onnotify);
+    // notify uses a regular pool connection via parameterized query — no dedicated listen connection
+    sql.notify = (channel, payload) => {
+      validateChannel(channel);
+      if (typeof payload !== "string") {
+        throw $ERR_INVALID_ARG_TYPE("payload", "string", payload);
+      }
+      return sql.unsafe("SELECT pg_notify($1, $2)", [channel, payload]);
+    };
+  } else {
+    // Stubs for adapters without LISTEN/NOTIFY — keep the API shape uniform
+    // and surface a clear error instead of "sql.listen is not a function".
+    const unsupported = () =>
+      Promise.$reject(new Error("LISTEN/NOTIFY is not supported by this adapter (PostgreSQL only)"));
+    sql.listen = (channel, onnotify, onlisten?) => {
+      validateChannel(channel);
+      if (!$isCallable(onnotify)) throw $ERR_INVALID_ARG_TYPE("onnotify", "function", onnotify);
+      if (onlisten !== undefined && !$isCallable(onlisten)) {
+        throw $ERR_INVALID_ARG_TYPE("onlisten", "function", onlisten);
+      }
+      return unsupported();
+    };
+    sql.unlisten = (channel, onnotify?) => {
+      validateChannel(channel);
+      if (onnotify !== undefined && !$isCallable(onnotify)) {
+        throw $ERR_INVALID_ARG_TYPE("onnotify", "function", onnotify);
+      }
+      return unsupported();
+    };
+    sql.notify = (channel, payload) => {
+      validateChannel(channel);
+      if (typeof payload !== "string") throw $ERR_INVALID_ARG_TYPE("payload", "string", payload);
+      return unsupported();
+    };
+  }
+
   return sql;
 };
 
@@ -1015,6 +1062,18 @@ defaultSQLObject.flush = (...args: Parameters<typeof lazyDefaultSQL.flush>) => {
   ensureDefaultSQL();
   return lazyDefaultSQL.flush(...args);
 };
+defaultSQLObject.listen = (...args) => {
+  ensureDefaultSQL();
+  return lazyDefaultSQL.listen(...args);
+};
+defaultSQLObject.unlisten = (...args) => {
+  ensureDefaultSQL();
+  return lazyDefaultSQL.unlisten(...args);
+};
+defaultSQLObject.notify = (...args) => {
+  ensureDefaultSQL();
+  return lazyDefaultSQL.notify(...args);
+};
 //define lazy properties
 defineProperties(defaultSQLObject, {
   options: {