add sql listen, unlisten, notify for postgres

[versecafe]

What does this PR do?

Adds sql.listen, sql.unlisten, sql.notify support for postgres matching the postgres.js api

import { SQL } from "bun";
const sql = new SQL("postgres://localhost/mydb");

const { state, unlisten } = await sql.listen(
  "user_events",
  payload => console.log("got", JSON.parse(payload)),
  () => console.log("listening on backend pid", state.pid),
);

await sql.notify("user_events", JSON.stringify({ type: "signup", id: 42 }));
// -> listening on backend pid 84231
// -> got { type: "signup", id: 42 }

// Auto-reconnects with exponential backoff if the connection drops;
// onlisten fires again, state.pid updates in place.

await unlisten();   // remove just this listener
await sql.close();  // closes the dedicated listen connection

How did you verify your code works?

17 tests added and some manual checks

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

[versecafe]

could probably delete the commented out listen/notify pseudo code from robobun in sql.test.ts but didn't want to touch extra stuff

[coderabbitai]

Walkthrough

Adds PostgreSQL LISTEN/NOTIFY pub/sub to Bun SQL: listen(), unlisten(), and notify() methods, a lazily-created shared listen connection, automatic reconnect with exponential backoff and jitter, in-place mutable listener state updates on reconnect, and per-channel failure tracking with idempotent unlisten behavior.

Changes

Cohort / File(s)	Summary
Documentation docs/runtime/sql.mdx	Adds LISTEN/NOTIFY docs and runnable example; documents runtime behavior (lazy dedicated listen connection, reconnect/backoff, shared state semantics), notify() semantics, error handling, and removes prior "not implemented" markers.
Type Definitions packages/bun-types/sql.d.ts	Adds listen(), unlisten(), and notify() method signatures to the SQL class, including returned shared state and unlisten() promise.
Public JS Surface src/js/bun/sql.ts, src/js/bun/.../sql	Adds listen, unlisten, and notify on the default SQL object; validates channel/payload; forwards to adapter when supported or returns a standardized "not supported" rejected Promise; ensures lazy initialization.
Postgres adapter implementation src/js/internal/sql/postgres.ts	Implements LISTEN/NOTIFY subsystem: single reusable listen connection, per-channel handler sets and onlisten callbacks, coordinated registrations, exponential backoff + jitter and reconnect with re-issuing LISTEN, shared mutable { pid, secret } updated in-place, per-channel failure counting and drop logic, idempotent unlisten, and shutdown ordering to close listen subsystem first.
Connection classes & protocol src/bun.js/api/sql.classes.ts, src/sql/postgres/PostgresSQLConnection.zig, src/sql/postgres/PostgresRequest.zig, src/sql/postgres/protocol/NotificationResponse.zig	Adds onnotification getter/setter and read-only processId/secretKey to connection class surface; extends protocol handling for 'A' (Notification) messages, decodes and dispatches notifications to JS via microtask (with safe buffer copies), and tightens decode error handling.
Native password handling src/js/internal/sql/postgres.ts (helper)	Extracts resolvePostgresPassword() and ensures connection creation receives an eagerly-resolved password string.
Tests test/js/sql/sql.test.ts	Adds comprehensive LISTEN/NOTIFY tests: end-to-end delivery, handler registration/unregistration semantics, shared listener state (PID/secret), reconnect behavior, argument validation, empty payloads, and unlisten idempotency.

🚥 Pre-merge checks | ✅ 4

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'add sql listen, unlisten, notify for postgres' accurately and clearly summarizes the main changes in the PR, which adds three new methods to the SQL API for PostgreSQL.
Description check	✅ Passed	The PR description follows the required template with both sections completed. It provides a clear example of the new API, explains the functionality including auto-reconnect behavior, and documents verification through 17 added tests.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 8

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@packages/bun-types/sql.d.ts`:
- Around line 885-889: The listen declaration incorrectly types onlisten as
taking no arguments; update the listen signature so onlisten is typed to receive
the mutable listen state object (e.g. onlisten?: (state: { pid: number; secret:
number }) => void) so callers get the pid/secret on initial connect and
reconnect; keep the existing onnotify and return shape (state and unlisten)
unchanged and only change the onlisten parameter type in the listen function
declaration.

In `@src/js/bun/sql.ts`:
- Around line 940-956: Attach stable stubs for sql.listen, sql.unlisten, and
sql.notify even when pool.listen is missing so the API shape is consistent; if
pool.listen exists keep the real implementations (pool.listen, pool.unlisten,
sql.unsafe...), otherwise set sql.listen and sql.unlisten to functions that
immediately return a rejected Promise with a clear Error like "adapter doesn't
support LISTEN/NOTIFY", and set sql.notify to a stub that first validates
channel/payload the same way as the real notify (check non-empty string, no null
bytes, payload type) and then returns a rejected Promise with the same
adapter-unsupported Error; update the code paths around pool.listen, sql.listen,
sql.unlisten, and sql.notify (and the lazy default client code paths such as
lazyDefaultSQL / new SQL) so callers always see these methods but get a
consistent rejection when the adapter lacks LISTEN/NOTIFY.

In `@src/js/internal/sql/postgres.ts`:
- Line 1565: Extract a private helper to centralize channel quoting and replace
duplicated channel.replaceAll('"','""') occurrences: add a method like
`#quoteChannel`(channel) that returns the channel wrapped in double quotes with
internal quotes escaped (e.g. `"${...}"`), then update callers such as the
LISTEN/UNLISTEN invocations that call this.#runListenQuery(conn, `LISTEN
"${channel.replaceAll('"', '""')}"`) (and the similar UNLISTEN sites) to use
this.#quoteChannel(channel) instead (e.g. `LISTEN
${this.#quoteChannel(channel)}`) so all quoting logic is in one place.
- Around line 1430-1452: In `#closeListen`(), reset the backoff by setting
this.#listenReconnectDelay back to 250 (ms) when clearing timers/connection
state so a subsequent re-listen starts from the documented baseline; locate the
method named `#closeListen` and add the single assignment
this.#listenReconnectDelay = 250 alongside the existing clears for
`#listenReconnectTimer`, `#listenConnectPromise`, `#listenChannels`, and
`#listenChannelFailures`.
- Around line 1699-1712: The returned unlisten closure currently only forwards
onnotify to this.unlisten(channel, onnotify) and therefore leaves the onlisten
callback (registered in listen()) in `#listenOnlistenCallbacks`, causing it to
fire after reconnects; update the closure returned by listen() so it also
removes the specific onlisten callback from
this.#listenOnlistenCallbacks[channel] (remove only that callback from the Set,
and if the Set becomes empty delete the map entry) before/after calling
this.unlisten(channel, onnotify), mirroring the rollback cleanup logic in
listen() so the exact onlisten installed by this listen() call is removed even
when sibling listeners exist.
- Around line 1454-1515: The `#createListenConnection` method duplicates password
resolution and createPostgresConnection wiring that already exists in
PooledPostgresConnection.createConnection; extract a shared helper (e.g.,
resolvePassword(info)) to perform the password callback/promise unwrap used in
both places, and create a thin connection-factory function that encapsulates the
createPostgresConnection invocation and its success/error callbacks so both
`#createListenConnection` and PooledPostgresConnection.createConnection call the
same factory; update `#createListenConnection` to call resolvePassword(info) and
the shared factory (keeping the existing listen-specific logic such as setting
`#listenConnection`, `#listenState`, onnotification, ref(), and reconnect handling).
- Around line 1665-1693: The current race comes from using isNewChannel computed
after we already inserted this call's handler into this.#listenChannels, so
concurrent listen() calls can skip issuing LISTEN and leave a handler registered
without an active LISTEN; fix by detecting whether THIS call actually
created/added the channel (capture previous size/creation status before
inserting) and only skip `#runListenQuery` when a previous caller truly issued
LISTEN, e.g. read prevSize = this.#listenChannels.get(channel)?.size ?? 0, if
prevSize === 0 then this call is responsible to await
this.#ensureListenConnection() and await this.#runListenQuery(...), otherwise
just add handler and don’t assume LISTEN succeeded; alternatively always call
`#runListenQuery` (idempotent on server) and keep the existing rollback logic
around `#listenChannels` and `#listenOnlistenCallbacks` using those same symbols
(`#listenChannels`, `#listenOnlistenCallbacks`, `#ensureListenConnection`,
`#runListenQuery`) so failed LISTEN removes only this call's handlers.

In `@test/js/sql/sql.test.ts`:
- Around line 12388-12419: The LISTEN/NOTIFY tests (e.g., the tests named
"sql.listen and sql.notify round-trip a payload" and "sql.listen receives
multiple notifications in order") currently await Promise.withResolvers()
without a per-assertion timeout, which can hang the suite; update each test that
uses Promise.withResolvers() together with db.listen/db.notify (also at the
other noted ranges) to await the subscriber-resolved promise with a bounded
timeout: wrap the Promise.withResolvers().promise in a timeout helper (e.g.,
withTimeout) or race it against a short timeout and fail if it elapses, and
ensure the resolver is invoked from the db.listen callback when the expected
condition is met, then unlisten as before; reference db.listen, db.notify,
db.unlisten, Promise.withResolvers, and the test names to locate the places to
change.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 0c19f071-45b5-41c8-b170-bac9a5545dce

📥 Commits

Reviewing files that changed from the base of the PR and between 75b7475 and 05630a8.

📒 Files selected for processing (9)

• docs/runtime/sql.mdx
• packages/bun-types/sql.d.ts
• src/bun.js/api/sql.classes.ts
• src/js/bun/sql.ts
• src/js/internal/sql/postgres.ts
• src/sql/postgres/PostgresRequest.zig
• src/sql/postgres/PostgresSQLConnection.zig
• src/sql/postgres/protocol/NotificationResponse.zig
• test/js/sql/sql.test.ts

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/js/bun/sql.ts`:
- Around line 962-963: The helper unsupported() currently returns
Promise.reject(...) which is inconsistent and uses a tamperable public API;
change its implementation to return the JSC intrinsic Promise.$reject(new
Error("LISTEN/NOTIFY is not supported by this adapter (PostgreSQL only)")) so it
matches other rejections in this file (see uses of Promise.$reject) and keeps
private global usage consistent with the coding guidelines.

In `@src/js/internal/sql/postgres.ts`:
- Line 1556: The reconnect backoff timer (this.#listenReconnectTimer) can keep
the process alive after unlisten() because it isn't unref()'d; modify the code
that assigns this.#listenReconnectTimer = setTimeout(...) to call .unref() on
the returned timer handle (and keep the reference for potential clearTimeout),
and also update unlisten() to clear and null the timer when `#listenChannels`
becomes empty to be safe; reference the private field
this.#listenReconnectTimer, the unlisten() method, and the existing
conn.ref()/conn.unref() logic when adding the unref() and clearTimeout handling.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 3161425e-5110-4799-b363-2f05a7e006b5

📥 Commits

Reviewing files that changed from the base of the PR and between 05630a8 and 4659ea5.

📒 Files selected for processing (3)

• packages/bun-types/sql.d.ts
• src/js/bun/sql.ts
• src/js/internal/sql/postgres.ts

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

src/js/internal/sql/postgres.ts (1)

1035-1047: ⚠️ Potential issue | 🟡 Minor

Move #closeListen() after the options.timeout validation.

this.#closeListen() runs unconditionally at the top of close(), before Number(timeout) / range validation. If the caller passes an invalid timeout (NaN, negative, or > 2**31), the listen connection is already torn down and #listenChannels / #listenOnlistenCallbacks are cleared, but then the function throws and this.closed is never set. The listen subsystem is now permanently dead while the pool is still "open", and a second close() call also re-runs #closeListen() (no-op) and throws again on the same bad input.

Pre-PR, validation came before the only destructive step (this.closed = true + #close()). Restoring that ordering keeps close() failure-atomic w.r.t. the listen subsystem.

♻️ Suggested fix

   async close(options?: { timeout?: number }): Promise<void> {
     if (this.closed) {
       return;
     }
 
-    this.#closeListen();
-
     let timeout = options?.timeout;
     if (timeout) {
       timeout = Number(timeout);
       if (timeout > 2 ** 31 || timeout < 0 || timeout !== timeout) {
         throw $ERR_INVALID_ARG_VALUE("options.timeout", timeout, "must be a non-negative integer less than 2^31");
       }
+      this.#closeListen();
       this.closed = true;
       ...
     } else {
+      this.#closeListen();
       this.closed = true;
       ...
     }
   }

Or simply hoist the validation above #closeListen() and keep a single call site.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/js/internal/sql/postgres.ts` around lines 1035 - 1047, The close() method
currently calls the private `#closeListen`() before validating options.timeout,
which can destructively tear down listen state then throw and leave this.closed
false; move the timeout validation (Number conversion and range/NaN checks for
options.timeout) to run before calling this.#closeListen() so any invalid
timeout throws without altering listen state, then proceed to call
this.#closeListen() and the rest of the teardown (this.closed and `#close`) as
before; target the async close(...) function and its timeout handling code paths
to apply this reorder, keeping a single call site for this.#closeListen().

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Outside diff comments:
In `@src/js/internal/sql/postgres.ts`:
- Around line 1035-1047: The close() method currently calls the private
`#closeListen`() before validating options.timeout, which can destructively tear
down listen state then throw and leave this.closed false; move the timeout
validation (Number conversion and range/NaN checks for options.timeout) to run
before calling this.#closeListen() so any invalid timeout throws without
altering listen state, then proceed to call this.#closeListen() and the rest of
the teardown (this.closed and `#close`) as before; target the async close(...)
function and its timeout handling code paths to apply this reorder, keeping a
single call site for this.#closeListen().

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: ba79efac-17b5-49fd-b4dc-51fdfa5b81fe

📥 Commits

Reviewing files that changed from the base of the PR and between 4659ea5 and ba2089c.

📒 Files selected for processing (2)

• src/js/bun/sql.ts
• src/js/internal/sql/postgres.ts

[versecafe]

leaving this to be converted by robobun/claude to the rust codebase

[robobun]

Opened #32089 with this branch rebased onto main: the native side is now implemented in the Rust port (src/sql_jsc/postgres/) since the .zig files are no longer compiled. Your three commits are preserved unchanged on that branch, so the JS/types/docs/tests stay credited to you.

[robobun]

Thanks for this, @versecafe. As you suggested, this branch was carried over to the Rust port as #32089: it started from your commits (you are credited as co-author on the commits there), and it has since been reworked there following maintainer review (listen() now resolves to a subscription object with unlisten() and await using, sql.unlisten() and the state object were dropped at alii's request). The dedicated listen connection, reconnect with backoff, onlisten on every (re)subscribe and pg_notify based notify() all came from here.

The .zig files this PR modifies no longer exist on main, so closing this one in favor of #32089, which is the one under review now. If you want to weigh in on the API shape, that thread is the place.