Add sql.listen() and sql.notify() for PostgreSQL LISTEN/NOTIFY

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 variables
+- PostgreSQL `LISTEN` / `NOTIFY`
 
 ---
 
@@ -984,6 +985,63 @@ try {
 
 ---
 
+## LISTEN / NOTIFY (PostgreSQL)
+
+`sql.listen()` subscribes to a PostgreSQL notification channel and `sql.notify()` publishes to one. Processes sharing a database can use this as a lightweight message bus: cache invalidation, waking workers when rows are inserted (typically from a trigger calling `pg_notify`), fan-out of small events.
+
+```ts
+import { sql } from "bun";
+
+const subscription = await sql.listen("orders", payload => {
+  console.log("new order", JSON.parse(payload));
+});
+
+await sql.notify("orders", JSON.stringify({ id: 42 }));
+
+await subscription.unlisten();
+```
+
+`listen()` resolves once PostgreSQL has acknowledged the `LISTEN`, so a `notify()` issued after it resolves is delivered. The subscription it resolves to is also an async disposable:
+
+```ts
+{
+  await using subscription = await sql.listen("orders", handleOrder);
+  await doWork();
+} // unlistened here
+```
+
+### How it works
+
+- All subscriptions on a client share one dedicated connection. The first `listen()` opens it and removing the last subscription closes it, so a client that never listens never pays for it, and unlistening everything lets the process exit without `sql.close()`.
+- While anything is subscribed, that connection keeps the process alive, like a listening server.
+- If the connection drops, it is re-established with exponential backoff (250ms doubling to 32s, with jitter) and every channel is re-subscribed. PostgreSQL only delivers to connected listeners, so notifications sent in between are lost; the optional third argument to `listen()` runs on the initial subscribe and after every reconnect, which is the place to catch up:
+
+```ts
+await sql.listen("orders", handleOrder, async () => {
+  for (const order of await sql`SELECT * FROM orders WHERE processed = false`) handleOrder(order);
+});
+```
+
+- Every `listen()` call is its own subscription. Several on one channel share a single server-side `LISTEN`, each callback receives every notification, and each handle's `unlisten()` removes only what its own call registered. A callback (either argument) that throws is reported as an uncaught exception and stays subscribed.
+- Channel names are quoted as identifiers for you; like any PostgreSQL identifier they are limited to 63 bytes, and longer names are rejected rather than silently truncated. PostgreSQL limits payloads to 8000 bytes by default.
+
+### `notify()`
+
+`notify()` is an ordinary query (`SELECT pg_notify($1, $2)`) on whichever handle you call it through. On `sql` it uses the pool; inside `sql.begin()` it runs in the transaction, so PostgreSQL delivers it on `COMMIT` and drops it on `ROLLBACK`, which is how to announce a change only once it is visible:
+
+```ts
+await sql.begin(async tx => {
+  const [order] = await tx`INSERT INTO orders ${sql(data)} RETURNING id`;
+  await tx.notify("orders", String(order.id)); // delivered when the transaction commits
+});
+```
+
+The payload is optional: `sql.notify("cache-invalidated")` is PostgreSQL's bare `NOTIFY`. Reserved and transaction handles also have `listen()`; it always uses the client's shared listen connection.
+
+`LISTEN`/`NOTIFY` is PostgreSQL-only; on MySQL and SQLite these methods reject.
+
+---
+
 ## Prepared Statements
 
 By default, Bun's SQL client creates named prepared statements for queries it can infer are static, which is faster. To disable this, set `prepare: false` in the connection options:
@@ -1303,8 +1361,6 @@ We haven't implemented `LOAD DATA INFILE` support yet.
 We haven't implemented these yet:
 
 - `COPY` support
-- `LISTEN` support
-- `NOTIFY` support
 
 We also haven't implemented some of the more uncommon features like:
 

packages/bun-types/sql.d.ts

@@ -433,6 +433,17 @@ declare module "bun" {
      */
     type Options = SQLiteOptions | PostgresOrMySQLOptions;
 
+    /** One registration made by {@link SQL.listen}. */
+    interface ListenSubscription extends AsyncDisposable {
+      readonly channel: string;
+      /**
+       * Remove this registration. Resolves once the channel is no longer
+       * subscribed, or immediately when other registrations on it remain.
+       * Idempotent; `await using` calls it at the end of the scope.
+       */
+      unlisten(): Promise<void>;
+    }
+
     /**
      * A pending SQL query. Extends `Promise`, so it can be awaited, and adds
      * methods to control how it runs.
@@ -959,6 +970,57 @@ declare module "bun" {
      * ```
      */
     file<T = any>(filename: string, values?: any[] | Record<string, any>): SQL.Query<T>;
+
+    /**
+     * Subscribe to a PostgreSQL `LISTEN` channel. Resolves once the server has
+     * acknowledged the subscription, with a handle that removes it again.
+     *
+     * Every call is its own registration: several on one channel share a
+     * single server-side subscription and each receives every notification.
+     * All of them share one dedicated connection, opened by the first
+     * `listen()` and closed when the last registration is removed. If it
+     * drops, it is reconnected with exponential backoff and every channel is
+     * re-subscribed; `onlisten` runs again each time.
+     *
+     * A throwing `onnotify` or `onlisten` is reported as an uncaught exception.
+     *
+     * @param channel - Channel name, quoted for you; at most 63 bytes, the
+     * PostgreSQL identifier limit
+     * @param onnotify - Receives each notification's payload
+     * @param onlisten - Runs once the `LISTEN` is acknowledged, initially and
+     * after every reconnect
+     *
+     * @example
+     * ```ts
+     * const subscription = await sql.listen("events", payload => console.log(payload));
+     * await sql.notify("events", "hello");
+     * await subscription.unlisten();
+     * ```
+     *
+     * @example
+     * ```ts
+     * await using subscription = await sql.listen("events", handle);
+     * ```
+     */
+    listen(
+      channel: string,
+      onnotify: (payload: string) => void,
+      onlisten?: () => void,
+    ): Promise<SQL.ListenSubscription>;
+
+    /**
+     * Send a PostgreSQL `NOTIFY` via `pg_notify`. Runs as a normal query on
+     * this handle, so on a `sql.begin()` transaction it is delivered on commit
+     * and discarded on rollback. Omitting `payload` sends an empty one, like a
+     * bare `NOTIFY channel`.
+     *
+     * @example
+     * ```ts
+     * await sql.notify("events", JSON.stringify({ id: 1 }));
+     * await sql.notify("cache-invalidated");
+     * ```
+     */
+    notify(channel: string, payload?: string): Promise<void>;
   }
 
   /**

src/js/bun/sql.ts

@@ -237,6 +237,43 @@ const SQL: typeof Bun.SQL = function SQL(
     }
   }
 
+  const listenable = "listen" in pool ? pool : null;
+  function validateChannel(channel: unknown): asserts channel is string {
+    if (typeof channel !== "string" || channel.length === 0) {
+      throw $ERR_INVALID_ARG_VALUE("channel", channel, "must be a non-empty string");
+    }
+    if (channel.includes("\0")) {
+      throw $ERR_INVALID_ARG_VALUE("channel", channel, "must not contain null bytes");
+    }
+    // PostgreSQL truncates longer identifiers, so notifications would arrive under a different name.
+    if (Buffer.byteLength(channel) > 63) {
+      throw $ERR_INVALID_ARG_VALUE("channel", channel, "must be at most 63 bytes");
+    }
+  }
+  function validateCallback(name: string, fn: unknown, optional: boolean) {
+    if (!$isCallable(fn) && !(optional && fn === undefined)) {
+      throw $ERR_INVALID_ARG_TYPE(name, "function", fn);
+    }
+  }
+  const listenUnsupported = () =>
+    Promise.$reject(new Error("LISTEN/NOTIFY is not supported by this adapter (PostgreSQL only)"));
+  const listen: Bun.SQL["listen"] = async (channel, onnotify, onlisten) => {
+    validateChannel(channel);
+    validateCallback("onnotify", onnotify, false);
+    validateCallback("onlisten", onlisten, true);
+    return listenable ? listenable.listen(channel, onnotify, onlisten) : listenUnsupported();
+  };
+  // .execute(): queries are lazy, and notify() must send even when not awaited.
+  function makeNotify(target: { unsafe: Bun.SQL["unsafe"] }): Bun.SQL["notify"] {
+    return (channel, payload) => {
+      validateChannel(channel);
+      if (payload === undefined) payload = "";
+      else if (typeof payload !== "string") throw $ERR_INVALID_ARG_TYPE("payload", "string", payload);
+      if (!listenable) return listenUnsupported();
+      return target.unsafe("SELECT pg_notify($1, $2)", [channel, payload]).execute() as unknown as Promise<void>;
+    };
+  }
+
   function onReserveConnected(this: Query<any, any>, err: Error | null, pooledConnection) {
     const { resolve, reject } = this;
 
@@ -317,6 +354,8 @@ const SQL: typeof Bun.SQL = function SQL(
     // this matchs the behavior of the postgres package
     reserved_sql.reserve = () => sql.reserve();
     reserved_sql.array = sql.array;
+    reserved_sql.listen = listen;
+    reserved_sql.notify = makeNotify(reserved_sql);
     function onTransactionFinished(transaction_promise: Promise<any>) {
       reservedTransaction.delete(transaction_promise);
     }
@@ -592,6 +631,8 @@ const SQL: typeof Bun.SQL = function SQL(
     // this matchs the behavior of the postgres package
     transaction_sql.reserve = () => sql.reserve();
     transaction_sql.array = sql.array;
+    transaction_sql.listen = listen;
+    transaction_sql.notify = makeNotify(transaction_sql);
 
     transaction_sql.connect = () => {
       if (state.connectionState & ReservedConnectionState.closed) {
@@ -938,6 +979,8 @@ const SQL: typeof Bun.SQL = function SQL(
   sql.transaction = sql.begin;
   sql.distributed = sql.beginDistributed;
   sql.end = sql.close;
+  sql.listen = listen;
+  sql.notify = makeNotify(sql);
   return sql;
 };
 
@@ -1017,6 +1060,14 @@ defaultSQLObject.flush = (...args: Parameters<typeof lazyDefaultSQL.flush>) => {
   ensureDefaultSQL();
   return lazyDefaultSQL.flush(...args);
 };
+defaultSQLObject.listen = (...args: Parameters<typeof lazyDefaultSQL.listen>) => {
+  ensureDefaultSQL();
+  return lazyDefaultSQL.listen(...args);
+};
+defaultSQLObject.notify = (...args: Parameters<typeof lazyDefaultSQL.notify>) => {
+  ensureDefaultSQL();
+  return lazyDefaultSQL.notify(...args);
+};
 //define lazy properties
 defineProperties(defaultSQLObject, {
   options: {