diff --git a/docs/runtime/sql.mdx b/docs/runtime/sql.mdx
index 230d4af311d7..99f9ac45a0a9 100644
--- a/docs/runtime/sql.mdx
+++ b/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:
 
diff --git a/packages/bun-types/sql.d.ts b/packages/bun-types/sql.d.ts
index 59681350ffb0..58910137ef27 100644
--- a/packages/bun-types/sql.d.ts
+++ b/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
diff --git a/src/bun.js/api/sql.classes.ts b/src/bun.js/api/sql.classes.ts
index ee1405ca47c6..b7353c50fef2 100644
--- a/src/bun.js/api/sql.classes.ts
+++ b/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"],
     }),
   );
 
diff --git a/src/js/bun/sql.ts b/src/js/bun/sql.ts
index dc436d367fef..f112af29eb12 100644
--- a/src/js/bun/sql.ts
+++ b/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: {
diff --git a/src/js/internal/sql/postgres.ts b/src/js/internal/sql/postgres.ts
index af4502cd9ad3..b40246e78b35 100644
--- a/src/js/internal/sql/postgres.ts
+++ b/src/js/internal/sql/postgres.ts
@@ -470,6 +470,14 @@ function onQueryFinish(this: PooledPostgresConnection, onClose: (err: Error) =>
   this.adapter.release(this);
 }
 
+async function resolvePostgresPassword(
+  password: Bun.MaybePromise<string> | string | undefined | (() => Bun.MaybePromise<string>),
+): Promise<string> {
+  if (typeof password === "function") password = password();
+  if (password && $isPromise(password)) password = await password;
+  return (password as string) || "";
+}
+
 class PooledPostgresConnection {
   private static async createConnection(
     options: Bun.SQL.__internal.DefinedPostgresOrMySQLOptions,
@@ -491,22 +499,14 @@ class PooledPostgresConnection {
       path,
     } = options;
 
-    let password: Bun.MaybePromise<string> | string | undefined | (() => Bun.MaybePromise<string>) = options.password;
-
     try {
-      if (typeof password === "function") {
-        password = password();
-      }
-
-      if (password && $isPromise(password)) {
-        password = await password;
-      }
+      const password = await resolvePostgresPassword(options.password);
 
       return createPostgresConnection(
         hostname,
         Number(port),
         username || "",
-        password || "",
+        password,
         database || "",
         // > The default value for sslmode is prefer. As is shown in the table, this
         // makes no sense from a security point of view, and it only promises
@@ -1037,6 +1037,8 @@ class PostgresAdapter
       return;
     }
 
+    this.#closeListen();
+
     let timeout = options?.timeout;
     if (timeout) {
       timeout = Number(timeout);
@@ -1397,6 +1399,375 @@ class PostgresAdapter
 
     return [query, binding_values];
   }
+
+  // --- LISTEN/NOTIFY ---
+  // A single dedicated connection is created and reused for all listeners
+  // on this adapter. It reconnects automatically with exponential backoff
+  // (250ms → 32s) when the connection drops, and re-issues LISTEN for every
+  // tracked channel. notify() goes through the regular pool via pg_notify().
+
+  #listenConnection: $ZigGeneratedClasses.PostgresSQLConnection | null = null;
+  #listenConnectPromise: Promise<$ZigGeneratedClasses.PostgresSQLConnection> | null = null;
+  // Per-channel listener sets. The presence of a key means LISTEN should be
+  // active for that channel on the dedicated connection.
+  #listenChannels: Map<string, Set<(payload: string) => void>> = new Map();
+  // Per-channel onlisten callbacks (fired on every successful LISTEN ack,
+  // including reconnects).
+  #listenOnlistenCallbacks: Map<string, Set<(state: { pid: number; secret: number }) => void>> = new Map();
+  // Shared by concurrent listen() calls on the same channel so they all await
+  // the same LISTEN ack instead of skipping it and resolving early.
+  #listenInFlight: Map<string, Promise<void>> = new Map();
+  #listenReconnectDelay: number = 250;
+  #listenReconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  // Track consecutive per-channel LISTEN failures during reconnect. A channel
+  // that fails this many times in a row is dropped to avoid an infinite retry
+  // loop on a permanently-misconfigured channel.
+  #listenChannelFailures: Map<string, number> = new Map();
+  readonly #listenMaxChannelFailures: number = 10;
+  // Shared state object returned from every listen() call. The same reference
+  // is handed out to all listeners and mutated in-place on each (re)connect, so
+  // user code holding `state` from a stale listen() always sees current pid/secret.
+  // This matches the postgres.js `state.pid` shape but with auto-update semantics.
+  #listenState: { pid: number; secret: number } = { pid: 0, secret: 0 };
+
+  #closeListen() {
+    if (this.#listenReconnectTimer) {
+      clearTimeout(this.#listenReconnectTimer);
+      this.#listenReconnectTimer = null;
+    }
+    const conn = this.#listenConnection;
+    this.#listenConnection = null;
+    // The in-flight create promise self-clears via .finally(), but null it here
+    // for symmetry so post-close synchronous reads don't see a stale promise.
+    this.#listenConnectPromise = null;
+    this.#listenChannels.clear();
+    this.#listenOnlistenCallbacks.clear();
+    this.#listenChannelFailures.clear();
+    this.#listenInFlight.clear();
+    this.#listenReconnectDelay = 250;
+    // Intentionally do NOT zero #listenState here — user code holding a `state`
+    // reference from a previous listen() should be able to inspect the last-known
+    // pid/secret after close (e.g. to log "shut down on backend N"). The next
+    // (re)connect overwrites these in place.
+    if (conn) {
+      try {
+        conn.close();
+      } catch {}
+    }
+  }
+
+  async #createListenConnection(): Promise<$ZigGeneratedClasses.PostgresSQLConnection> {
+    const info = this.connectionInfo;
+    const password = await resolvePostgresPassword(info.password);
+
+    // If close() ran while we were resolving the password, bail out.
+    if (this.closed) throw this.connectionClosedError();
+
+    const { promise, resolve, reject } = Promise.withResolvers<$ZigGeneratedClasses.PostgresSQLConnection>();
+    createPostgresConnection(
+      info.hostname,
+      Number(info.port),
+      info.username || "",
+      password,
+      info.database || "",
+      info.sslMode || SSLMode.disable,
+      info.tls || null,
+      info.query || "",
+      info.path || "",
+      (err, conn) => {
+        if (err) {
+          reject(wrapPostgresError(err));
+          return;
+        }
+        // If close() ran while the native side was connecting, the adapter
+        // is gone — close the orphan connection rather than leaking it.
+        if (this.closed) {
+          try {
+            conn.close();
+          } catch {}
+          reject(this.connectionClosedError());
+          return;
+        }
+        this.#listenConnection = conn;
+        this.#listenReconnectDelay = 250;
+        // Mutate the shared #listenState in place so every previously-returned
+        // listen() result sees the new connection's pid/secret on reconnect.
+        const connWithIds = conn as unknown as { processId: number; secretKey: number };
+        this.#listenState.pid = connWithIds.processId;
+        this.#listenState.secret = connWithIds.secretKey;
+        // Generated `.d.ts` marks `onnotification` readonly because the codegen
+        // produces a getter regardless of setter presence (same for onclose/onconnect).
+        // The Zig setter exists; cast through a writable shape rather than `any`.
+        (conn as { onnotification: (channel: string, payload: string) => void }).onnotification = (
+          channel: string,
+          payload: string,
+        ) => this.#dispatchNotification(channel, payload);
+        conn.ref();
+        resolve(conn);
+      },
+      _err => {
+        this.#listenConnection = null;
+        this.#scheduleListenReconnect();
+      },
+      0, // idleTimeout: never time out — listen connection must stay open
+      info.connectionTimeout ?? 30000,
+      0, // maxLifetime: never recycle
+      false, // useUnnamedPreparedStatements: irrelevant; only LISTEN/UNLISTEN run here
+    );
+    return promise;
+  }
+
+  async #ensureListenConnection(): Promise<$ZigGeneratedClasses.PostgresSQLConnection> {
+    if (this.#listenConnection) return this.#listenConnection;
+    if (!this.#listenConnectPromise) {
+      this.#listenConnectPromise = this.#createListenConnection().finally(() => {
+        this.#listenConnectPromise = null;
+      });
+    }
+    return this.#listenConnectPromise;
+  }
+
+  #dispatchNotification(channel: string, payload: string) {
+    const listeners = this.#listenChannels.get(channel);
+    if (!listeners) return;
+    for (const fn of listeners) {
+      try {
+        fn.$call(undefined, payload);
+      } catch {}
+    }
+  }
+
+  #scheduleListenReconnect() {
+    // Reconnect retries indefinitely as long as there are tracked channels and
+    // the adapter is open. There is no global attempt cap — if PG is permanently
+    // down, this will keep firing every ≤32s forever (capped delay below). Stops
+    // immediately when:
+    //   - the adapter is closed (#closeListen clears channels and the timer), or
+    //   - the last channel is unlistened (#listenChannels becomes empty), or
+    //   - all permanently-failing channels exceed #listenMaxChannelFailures and get
+    //     dropped, draining #listenChannels.
+    // This matches postgres.js, which also retries indefinitely.
+    if (this.closed || this.#listenChannels.size === 0 || this.#listenReconnectTimer) return;
+    // Apply ±25% jitter (multiplier in [0.75, 1.25]) to the base delay to avoid
+    // synchronized retry storms when many adapters in the same process lose their
+    // listen connection at once.
+    const jitter = 0.75 + Math.random() * 0.5;
+    const delayMs = Math.max(1, Math.floor(this.#listenReconnectDelay * jitter));
+    const timer = setTimeout(async () => {
+      this.#listenReconnectTimer = null;
+      if (this.closed || this.#listenChannels.size === 0) return;
+      // Snapshot before any await so #closeListen() cannot clear the map under us.
+      const channels = Array.from(this.#listenChannels.keys());
+      let anyChannelFailed = false;
+      try {
+        const conn = await this.#ensureListenConnection();
+        for (const channel of channels) {
+          // Channel may have been unlistened while we awaited the connection.
+          if (!this.#listenChannels.has(channel)) continue;
+          try {
+            await this.#runListenQuery(conn, `LISTEN ${this.#quoteChannel(channel)}`);
+            const failures = this.#listenChannelFailures.get(channel);
+            if (failures !== undefined) this.#listenChannelFailures.delete(channel);
+            const onlistens = this.#listenOnlistenCallbacks.get(channel);
+            if (onlistens) {
+              for (const fn of onlistens) {
+                try {
+                  fn.$call(undefined, this.#listenState);
+                } catch {}
+              }
+            }
+          } catch (err) {
+            // Per-channel LISTEN failed on a live connection (transient PG error,
+            // permissions blip, etc). Surface the error and track consecutive
+            // failures — drop the channel after #listenMaxChannelFailures so we
+            // don't retry forever for a permanently-misconfigured channel.
+            const failures = (this.#listenChannelFailures.get(channel) ?? 0) + 1;
+            this.#listenChannelFailures.set(channel, failures);
+            const errMsg = (err as Error)?.message ?? String(err);
+            if (failures >= this.#listenMaxChannelFailures) {
+              console.warn(
+                `bun:sql LISTEN to channel "${channel}" failed ${failures} times in a row; ` +
+                  `giving up and removing the subscription. Last error: ${errMsg}`,
+              );
+              this.#listenChannels.delete(channel);
+              this.#listenOnlistenCallbacks.delete(channel);
+              this.#listenChannelFailures.delete(channel);
+            } else {
+              console.warn(
+                `bun:sql LISTEN to channel "${channel}" failed (attempt ${failures}/${this.#listenMaxChannelFailures}): ${errMsg}`,
+              );
+              anyChannelFailed = true;
+            }
+          }
+        }
+        if (anyChannelFailed && this.#listenChannels.size > 0) {
+          // Connection is alive but at least one channel did not register.
+          // Schedule another tick; PG tolerates duplicate LISTENs as no-ops.
+          this.#listenReconnectDelay = Math.min(this.#listenReconnectDelay * 2, 32000);
+          this.#scheduleListenReconnect();
+        } else {
+          // All channels reconnected successfully — reset backoff for the next failure.
+          this.#listenReconnectDelay = 250;
+        }
+      } catch {
+        this.#listenReconnectDelay = Math.min(this.#listenReconnectDelay * 2, 32000);
+        this.#scheduleListenReconnect();
+      }
+    }, delayMs);
+    timer.unref?.();
+    this.#listenReconnectTimer = timer;
+  }
+
+  // PG channel identifiers must be double-quoted to preserve case and allow
+  // characters outside the unquoted-identifier grammar; embedded `"` doubles up.
+  #quoteChannel(channel: string): string {
+    return `"${channel.replaceAll('"', '""')}"`;
+  }
+
+  async #runListenQuery(conn: $ZigGeneratedClasses.PostgresSQLConnection, sql: string): Promise<void> {
+    const pendingValue = new SQLResultArray();
+    const handle = createPostgresQuery(sql, [], pendingValue, undefined, false, true);
+    const { promise, resolve, reject } = Promise.withResolvers<void>();
+    // Fake query object — bypasses the user-facing Query class because LISTEN/UNLISTEN
+    // don't fit the tagged-template path (no result rows, no Promise interface needed).
+    //
+    // CONTRACT (must stay in sync with src/js/internal/sql/postgres.ts initPostgres):
+    //   onResolvePostgresQuery (this file, ~line 237):
+    //     - is_last branch:    reads `query[_results]`, calls `query.resolve(...)`.
+    //                          May call `queries.indexOf(query)`+`splice` IF the
+    //                          connection has a JS-cached `queries` array — the
+    //                          dedicated listen connection does not, so this branch
+    //                          is skipped.
+    //     - !is_last branch:   reads `query[_handle]`, calls `setPendingValue` on it,
+    //                          mutates `query[_results]`. LISTEN/UNLISTEN produce a
+    //                          single CommandComplete + ReadyForQuery so this is not
+    //                          reached in practice — but we delegate setPendingValue
+    //                          to the real native handle just in case.
+    //   onRejectPostgresQuery (this file, ~line 300):
+    //     - calls `query.reject(err)`. May call `queries.indexOf(query)` (same
+    //       no-op as above for the listen connection).
+    //
+    // If either callback grows to access more fields (e.g. `query.cancel?.()`,
+    // `query.done()`, etc.), those fields MUST be added here or LISTEN/UNLISTEN
+    // will silently break. The native side never references this object directly —
+    // it only flows through the two JS callbacks above.
+    handle.run(conn, {
+      resolve: () => resolve(),
+      reject: (err: any) => reject(wrapPostgresError(err)),
+      [_results]: null,
+      [_handle]: {
+        setPendingValue: (v: SQLResultArray) => handle.setPendingValue(v),
+      },
+    });
+    return promise;
+  }
+
+  async listen(
+    channel: string,
+    onnotify: (payload: string) => void,
+    onlisten?: (state: { pid: number; secret: number }) => void,
+  ): Promise<{ state: { pid: number; secret: number }; unlisten: () => Promise<void> }> {
+    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 (!$isCallable(onnotify)) throw $ERR_INVALID_ARG_TYPE("onnotify", "function", onnotify);
+    if (onlisten !== undefined && !$isCallable(onlisten)) throw $ERR_INVALID_ARG_TYPE("onlisten", "function", onlisten);
+
+    if (!this.#listenChannels.has(channel)) this.#listenChannels.set(channel, new Set());
+    this.#listenChannels.get(channel)!.add(onnotify);
+
+    if (onlisten) {
+      if (!this.#listenOnlistenCallbacks.has(channel)) this.#listenOnlistenCallbacks.set(channel, new Set());
+      this.#listenOnlistenCallbacks.get(channel)!.add(onlisten);
+    }
+
+    try {
+      // Always issue LISTEN (idempotent server-side) and share the in-flight
+      // promise so concurrent callers can't resolve before the ack and can't
+      // miss the case where the connection dropped mid-handoff.
+      let inFlight = this.#listenInFlight.get(channel);
+      if (!inFlight) {
+        inFlight = (async () => {
+          const conn = await this.#ensureListenConnection();
+          await this.#runListenQuery(conn, `LISTEN ${this.#quoteChannel(channel)}`);
+        })().finally(() => {
+          if (this.#listenInFlight.get(channel) === inFlight) this.#listenInFlight.delete(channel);
+        });
+        this.#listenInFlight.set(channel, inFlight);
+      }
+      await inFlight;
+    } catch (err) {
+      // Roll back only the registrations this call made — siblings keep theirs.
+      const set = this.#listenChannels.get(channel);
+      set?.delete(onnotify);
+      if (onlisten) {
+        const onlistens = this.#listenOnlistenCallbacks.get(channel);
+        onlistens?.delete(onlisten);
+        if (onlistens && onlistens.size === 0) this.#listenOnlistenCallbacks.delete(channel);
+      }
+      if (set && set.size === 0) {
+        this.#listenChannels.delete(channel);
+        this.#listenOnlistenCallbacks.delete(channel);
+      }
+      throw err;
+    }
+
+    try {
+      onlisten?.(this.#listenState);
+    } catch {}
+
+    // unlistened is checked then set synchronously inside the closure; no await
+    // sits between the check and the assignment, so two simultaneous calls on a
+    // single isolate cannot both pass the gate. If an `await` is ever introduced
+    // before the flag flip, this becomes a TOCTOU bug — be careful.
+    let unlistened = false;
+    return {
+      state: this.#listenState,
+      unlisten: async () => {
+        if (unlistened) return;
+        unlistened = true;
+        // Drop this caller's onlisten — unlisten(channel, onnotify) only clears
+        // onlistenCallbacks on full teardown, so siblings would keep firing it.
+        if (onlisten) {
+          const onlistens = this.#listenOnlistenCallbacks.get(channel);
+          onlistens?.delete(onlisten);
+          if (onlistens && onlistens.size === 0) this.#listenOnlistenCallbacks.delete(channel);
+        }
+        await this.unlisten(channel, onnotify);
+      },
+    };
+  }
+
+  async unlisten(channel: string, onnotify?: (payload: string) => void): Promise<void> {
+    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 (onnotify !== undefined && !$isCallable(onnotify)) throw $ERR_INVALID_ARG_TYPE("onnotify", "function", onnotify);
+
+    if (!onnotify) {
+      this.#listenChannels.delete(channel);
+      this.#listenOnlistenCallbacks.delete(channel);
+      this.#listenChannelFailures.delete(channel);
+    } else {
+      this.#listenChannels.get(channel)?.delete(onnotify);
+      if (this.#listenChannels.get(channel)?.size === 0) {
+        this.#listenChannels.delete(channel);
+        this.#listenOnlistenCallbacks.delete(channel);
+        this.#listenChannelFailures.delete(channel);
+      }
+    }
+
+    if (this.#listenConnection && !this.#listenChannels.has(channel)) {
+      await this.#runListenQuery(this.#listenConnection, `UNLISTEN ${this.#quoteChannel(channel)}`);
+    }
+
+    // No remaining subscriptions — cancel any pending reconnect so it does not
+    // keep the event loop alive after the user is done listening.
+    if (this.#listenChannels.size === 0 && this.#listenReconnectTimer) {
+      clearTimeout(this.#listenReconnectTimer);
+      this.#listenReconnectTimer = null;
+    }
+  }
 }
 
 export default {
diff --git a/src/sql/postgres/PostgresRequest.zig b/src/sql/postgres/PostgresRequest.zig
index 9eb5085d045b..548007663129 100644
--- a/src/sql/postgres/PostgresRequest.zig
+++ b/src/sql/postgres/PostgresRequest.zig
@@ -378,6 +378,7 @@ pub fn onData(
             'H' => try connection.on(.CopyOutResponse, Context, reader),
             'c' => try connection.on(.CopyDone, Context, reader),
             'W' => try connection.on(.CopyBothResponse, Context, reader),
+            'A' => try connection.on(.NotificationResponse, Context, reader),
 
             else => {
                 debug("Unknown message: {c}", .{c});
diff --git a/src/sql/postgres/PostgresSQLConnection.zig b/src/sql/postgres/PostgresSQLConnection.zig
index a64937767463..9a4efd5b9cca 100644
--- a/src/sql/postgres/PostgresSQLConnection.zig
+++ b/src/sql/postgres/PostgresSQLConnection.zig
@@ -168,6 +168,45 @@ pub fn setOnClose(_: *PostgresSQLConnection, thisValue: jsc.JSValue, globalObjec
     js.oncloseSetCached(thisValue, globalObject, value);
 }
 
+pub fn getOnNotification(_: *PostgresSQLConnection, thisValue: jsc.JSValue, _: *jsc.JSGlobalObject) jsc.JSValue {
+    if (js.onnotificationGetCached(thisValue)) |value| {
+        return value;
+    }
+    return .js_undefined;
+}
+
+pub fn setOnNotification(_: *PostgresSQLConnection, thisValue: jsc.JSValue, globalObject: *jsc.JSGlobalObject, value: jsc.JSValue) void {
+    js.onnotificationSetCached(thisValue, globalObject, value);
+}
+
+/// Backend process ID delivered via BackendKeyData ('K') message during connection setup.
+/// Returns 0 before the message has been received. Useful for `pg_terminate_backend(pid)`
+/// and matches the `state.pid` field exposed by postgres.js.
+pub fn getProcessId(this: *PostgresSQLConnection, _: *jsc.JSGlobalObject) jsc.JSValue {
+    return jsc.JSValue.jsNumber(this.backend_key_data.process_id);
+}
+
+/// Backend secret key delivered via BackendKeyData ('K'). Required to issue a
+/// CancelRequest on a separate connection. Matches `state.secret` in postgres.js.
+pub fn getSecretKey(this: *PostgresSQLConnection, _: *jsc.JSGlobalObject) jsc.JSValue {
+    return jsc.JSValue.jsNumber(this.backend_key_data.secret_key);
+}
+
+pub fn dispatchNotification(this: *PostgresSQLConnection, channel: []const u8, payload: []const u8) void {
+    if (this.js_value == .zero) return;
+    if (this.vm.isShuttingDown()) return;
+    debug("dispatchNotification: channel={s} payload.len={d}", .{ channel, payload.len });
+    this.js_value.ensureStillAlive();
+    const callback = js.onnotificationGetCached(this.js_value) orelse return;
+    if (!callback.isCallable()) return;
+    callback.ensureStillAlive();
+    // cloneUTF8 copies into a ref-counted WTFStringImpl so the JSValue is safe
+    // to use after the NotificationResponse ByteLists are freed.
+    const channel_js = bun.String.cloneUTF8(channel).toJS(this.globalObject) catch return;
+    const payload_js = bun.String.cloneUTF8(payload).toJS(this.globalObject) catch return;
+    this.globalObject.queueMicrotask(callback, &[_]JSValue{ channel_js, payload_js });
+}
+
 pub fn setupTLS(this: *PostgresSQLConnection) void {
     debug("setupTLS", .{});
     const new_socket = this.socket.SocketTCP.socket.connected.upgrade(this.tls_ctx.?, this.tls_config.server_name) orelse {
@@ -1859,6 +1898,12 @@ pub fn on(this: *PostgresSQLConnection, comptime MessageType: @Type(.enum_litera
             defer this.updateRef();
             request.onResult("CLOSECOMPLETE", this.globalObject, this.js_value, false);
         },
+        .NotificationResponse => {
+            var notification: protocol.NotificationResponse = undefined;
+            try notification.decodeInternal(Context, reader);
+            defer notification.deinit();
+            this.dispatchNotification(notification.channel.slice(), notification.payload.slice());
+        },
         .CopyInResponse => {
             debug("TODO CopyInResponse", .{});
         },
diff --git a/src/sql/postgres/protocol/NotificationResponse.zig b/src/sql/postgres/protocol/NotificationResponse.zig
index 17229e596dc2..29bb8776d751 100644
--- a/src/sql/postgres/protocol/NotificationResponse.zig
+++ b/src/sql/postgres/protocol/NotificationResponse.zig
@@ -15,8 +15,8 @@ pub fn decodeInternal(this: *@This(), comptime Container: type, reader: NewReade
 
     this.* = .{
         .pid = try reader.int4(),
-        .channel = (try reader.readZ()).toOwned(),
-        .payload = (try reader.readZ()).toOwned(),
+        .channel = try (try reader.readZ()).toOwned(),
+        .payload = try (try reader.readZ()).toOwned(),
     };
 }
 
diff --git a/test/js/sql/sql.test.ts b/test/js/sql/sql.test.ts
index 25d9c77ec74f..2580b05eb96d 100644
--- a/test/js/sql/sql.test.ts
+++ b/test/js/sql/sql.test.ts
@@ -12384,5 +12384,334 @@ CREATE TABLE ${table_name} (
         expect(e.message).toContain("65535");
       }
     });
+    describe("LISTEN/NOTIFY", () => {
+      test("sql.listen and sql.notify round-trip a payload", async () => {
+        await using db = postgres(options);
+        const { promise, resolve } = Promise.withResolvers<string>();
+
+        await db.listen("test_listen_basic", payload => resolve(payload));
+        await db.notify("test_listen_basic", "hello-bun");
+
+        const received = await promise;
+        expect(received).toBe("hello-bun");
+        await db.unlisten("test_listen_basic");
+      });
+
+      test("sql.listen receives multiple notifications in order", async () => {
+        await using db = postgres(options);
+        const received: string[] = [];
+        let resolver: (() => void) | null = null;
+
+        await db.listen("test_listen_multi", payload => {
+          received.push(payload);
+          if (received.length >= 3) resolver?.();
+        });
+
+        const { promise, resolve } = Promise.withResolvers<void>();
+        resolver = resolve;
+
+        await db.notify("test_listen_multi", "one");
+        await db.notify("test_listen_multi", "two");
+        await db.notify("test_listen_multi", "three");
+
+        await promise;
+        expect(received).toEqual(["one", "two", "three"]);
+        await db.unlisten("test_listen_multi");
+      });
+
+      test("sql.listen invokes onlisten callback on registration", async () => {
+        await using db = postgres(options);
+        const { promise, resolve } = Promise.withResolvers<void>();
+
+        await db.listen("test_listen_onlisten", () => {}, resolve);
+        await promise;
+        await db.unlisten("test_listen_onlisten");
+      });
+
+      test("sql.listen returns state with pid/secret and unlisten", async () => {
+        await using db = postgres(options);
+        const result = await db.listen("test_listen_state", () => {});
+        expect(result).toBeObject();
+        expect(result.state).toBeDefined();
+        // state.pid should be a positive integer (PG backend PID, populated from BackendKeyData)
+        expect(typeof result.state.pid).toBe("number");
+        expect(result.state.pid).toBeGreaterThan(0);
+        // state.secret is the cancellation secret key
+        expect(typeof result.state.secret).toBe("number");
+        expect(typeof result.unlisten).toBe("function");
+        await result.unlisten();
+      });
+
+      test("sql.listen state is shared across listeners and updates on reconnect", async () => {
+        await using db = postgres(options);
+
+        // First subscription on the listen connection
+        const r1 = await db.listen("test_listen_state_shared", () => {});
+        const initialPid = r1.state.pid;
+        expect(initialPid).toBeGreaterThan(0);
+
+        // Second subscription should hand back the SAME state object reference,
+        // so users holding either reference always see the live backend pid/secret.
+        const r2 = await db.listen("test_listen_state_shared_2", () => {});
+        expect(r2.state).toBe(r1.state);
+        expect(r2.state.pid).toBe(initialPid);
+
+        // Drive a reconnect by terminating the listen backend from a pool query.
+        // onlisten fires once on registration AND again on reconnect — count fires
+        // and resolve only on the second (post-reconnect) firing.
+        const { promise: reconnected, resolve: resolveReconnect } = Promise.withResolvers<void>();
+        let onlistenFires = 0;
+        await db.listen(
+          "test_listen_state_shared_observer",
+          () => {},
+          () => {
+            if (++onlistenFires === 2) resolveReconnect();
+          },
+        );
+
+        // pg_terminate_backend may itself report an error on the kicking
+        // connection — narrow the catch to that one query so unrelated bugs
+        // (e.g. db being undefined) still surface as a real test failure.
+        await db.unsafe("SELECT pg_terminate_backend($1)", [initialPid]).catch(() => {});
+
+        await reconnected;
+
+        // Same object reference, but pid is now the *new* backend's PID.
+        expect(r1.state).toBe(r2.state);
+        expect(r1.state.pid).toBeGreaterThan(0);
+        expect(r1.state.pid).not.toBe(initialPid);
+
+        await db.unlisten("test_listen_state_shared");
+        await db.unlisten("test_listen_state_shared_2");
+        await db.unlisten("test_listen_state_shared_observer");
+      });
+
+      test("sql.listen reconnects and resumes notification delivery", async () => {
+        await using db = postgres(options);
+        const received: string[] = [];
+        const { promise: gotBefore, resolve: resolveBefore } = Promise.withResolvers<void>();
+        const { promise: gotAfter, resolve: resolveAfter } = Promise.withResolvers<void>();
+        const { promise: secondListen, resolve: resolveSecondListen } = Promise.withResolvers<void>();
+
+        let onlistenCount = 0;
+        const sub = await db.listen(
+          "test_listen_reconnect",
+          payload => {
+            received.push(payload);
+            if (payload === "before") resolveBefore();
+            if (payload === "after") resolveAfter();
+          },
+          () => {
+            onlistenCount++;
+            if (onlistenCount === 2) resolveSecondListen();
+          },
+        );
+        expect(onlistenCount).toBe(1);
+        const initialPid = sub.state.pid;
+
+        await db.notify("test_listen_reconnect", "before");
+        await gotBefore;
+
+        // Force the listen connection to drop server-side
+        await db.unsafe("SELECT pg_terminate_backend($1)", [initialPid]).catch(() => {});
+
+        // Wait for reconnect to issue LISTEN again (onlisten fires twice: original + reconnect)
+        await secondListen;
+        expect(sub.state.pid).not.toBe(initialPid);
+
+        // Notifications on the same channel should now be delivered via the new backend
+        await db.notify("test_listen_reconnect", "after");
+        await gotAfter;
+
+        expect(received).toEqual(["before", "after"]);
+        await sub.unlisten();
+      });
+
+      test("sql.listen reconnect re-fires onlisten for every channel", async () => {
+        await using db = postgres(options);
+        const fires: string[] = [];
+        const { promise: allReconnected, resolve } = Promise.withResolvers<void>();
+
+        let total = 0;
+        const onlisten = (channel: string) => () => {
+          fires.push(channel);
+          // 2 channels × 2 fires each (initial + after reconnect) = 4
+          if (++total === 4) resolve();
+        };
+
+        const r1 = await db.listen("test_reconnect_a", () => {}, onlisten("a"));
+        await db.listen("test_reconnect_b", () => {}, onlisten("b"));
+        const initialPid = r1.state.pid;
+
+        await db.unsafe("SELECT pg_terminate_backend($1)", [initialPid]).catch(() => {});
+
+        await allReconnected;
+
+        // Each channel's onlisten fired exactly twice (initial + post-reconnect)
+        expect(fires.filter(c => c === "a").length).toBe(2);
+        expect(fires.filter(c => c === "b").length).toBe(2);
+
+        await db.unlisten("test_reconnect_a");
+        await db.unlisten("test_reconnect_b");
+      });
+
+      test("sql.unlisten stops receiving notifications", async () => {
+        await using db = postgres(options);
+        const received: string[] = [];
+        const { promise: gotBefore, resolve: resolveBefore } = Promise.withResolvers<void>();
+        const { promise: gotBarrier, resolve: resolveBarrier } = Promise.withResolvers<void>();
+
+        await db.listen("test_listen_unlisten", payload => {
+          received.push(payload);
+          if (payload === "before") resolveBefore();
+        });
+        // Barrier channel shares the same listen connection; messages on a single
+        // connection are delivered in order, so when the barrier arrives we know
+        // any pending "after" message would already have been delivered.
+        await db.listen("test_listen_unlisten_barrier", () => resolveBarrier());
+
+        await db.notify("test_listen_unlisten", "before");
+        await gotBefore;
+
+        await db.unlisten("test_listen_unlisten");
+
+        await db.notify("test_listen_unlisten", "after");
+        await db.notify("test_listen_unlisten_barrier", "go");
+        await gotBarrier;
+
+        expect(received).toEqual(["before"]);
+
+        await db.unlisten("test_listen_unlisten_barrier");
+      });
+
+      test("multiple listeners on the same channel all receive notifications", async () => {
+        await using db = postgres(options);
+        const { promise: p1, resolve: r1 } = Promise.withResolvers<string>();
+        const { promise: p2, resolve: r2 } = Promise.withResolvers<string>();
+
+        const fn1 = (payload: string) => r1(payload);
+        const fn2 = (payload: string) => r2(payload);
+
+        await db.listen("test_listen_multi_listeners", fn1);
+        await db.listen("test_listen_multi_listeners", fn2);
+        await db.notify("test_listen_multi_listeners", "broadcast");
+
+        const [got1, got2] = await Promise.all([p1, p2]);
+        expect(got1).toBe("broadcast");
+        expect(got2).toBe("broadcast");
+
+        await db.unlisten("test_listen_multi_listeners");
+      });
+
+      test("sql.notify with pg_notify empty payload works", async () => {
+        await using db = postgres(options);
+        const { promise, resolve } = Promise.withResolvers<string>();
+
+        await db.listen("test_listen_empty", payload => resolve(payload));
+        await db.notify("test_listen_empty", "");
+
+        const received = await promise;
+        expect(received).toBe("");
+        await db.unlisten("test_listen_empty");
+      });
+
+      test("sql.listen rejects invalid channel names", async () => {
+        await using db = postgres(options);
+        // empty string
+        expect(db.listen("", () => {})).rejects.toThrow();
+        // null bytes are forbidden by PostgreSQL identifiers
+        expect(db.listen("with\0null", () => {})).rejects.toThrow();
+        // non-callable onnotify
+        expect(db.listen("ch", 42 as any)).rejects.toThrow();
+        // non-callable onlisten
+        expect(db.listen("ch", () => {}, 42 as any)).rejects.toThrow();
+      });
+
+      test("sql.notify validates arguments", async () => {
+        await using db = postgres(options);
+        expect(() => db.notify("", "payload")).toThrow();
+        expect(() => db.notify("with\0null", "payload")).toThrow();
+        expect(() => db.notify("ch", null as any)).toThrow();
+        expect(() => db.notify("ch", 42 as any)).toThrow();
+      });
+
+      test("sql.unlisten validates arguments", async () => {
+        await using db = postgres(options);
+        expect(db.unlisten("")).rejects.toThrow();
+        expect(db.unlisten("with\0null")).rejects.toThrow();
+        expect(db.unlisten("ch", 42 as any)).rejects.toThrow();
+      });
+
+      test("unlisten() returned from listen() is idempotent", async () => {
+        await using db = postgres(options);
+        const { unlisten } = await db.listen("test_double_unlisten", () => {});
+        await unlisten();
+        // Calling again must not throw and must not send a second UNLISTEN.
+        await unlisten();
+        await unlisten();
+      });
+
+      test("unlisten() returned from listen() removes only that listener", async () => {
+        await using db = postgres(options);
+        const { promise: gotBoth, resolve: resolveBoth } = Promise.withResolvers<void>();
+        let count = 0;
+        const received: string[] = [];
+
+        const fn1 = (payload: string) => {
+          received.push(`fn1:${payload}`);
+          if (++count === 2) resolveBoth();
+        };
+        const fn2 = (payload: string) => {
+          received.push(`fn2:${payload}`);
+          if (++count === 2) resolveBoth();
+        };
+
+        const sub1 = await db.listen("test_partial_unlisten", fn1);
+        await db.listen("test_partial_unlisten", fn2);
+
+        // Remove only fn1
+        await sub1.unlisten();
+
+        // Use a barrier channel for synchronization
+        const { promise: gotBarrier, resolve: resolveBarrier } = Promise.withResolvers<void>();
+        await db.listen("test_partial_unlisten_barrier", () => resolveBarrier());
+
+        await db.notify("test_partial_unlisten", "after_unlisten");
+        await db.notify("test_partial_unlisten_barrier", "go");
+        await gotBarrier;
+
+        // fn2 should still receive, fn1 should not
+        expect(received).toEqual(["fn2:after_unlisten"]);
+
+        await db.unlisten("test_partial_unlisten");
+        await db.unlisten("test_partial_unlisten_barrier");
+      });
+
+      test("close during in-flight listen does not leak the connection", async () => {
+        // Force a fresh adapter (no shared pool) and start listen + close in the same tick.
+        const db = postgres(options);
+        const listenPromise = db.listen("test_close_during_listen", () => {}).catch(() => {});
+        await db.close({ timeout: 0 });
+        await listenPromise; // should resolve/reject cleanly without leaking
+      });
+
+      test("sql.listen onlisten fires after server-side LISTEN is registered", async () => {
+        await using db = postgres(options);
+        const { promise: onlistenFired, resolve } = Promise.withResolvers<void>();
+        await db.listen("test_onlisten_timing", () => {}, resolve);
+        await onlistenFired;
+
+        // At this point, pg_listening_channels() on the listen connection
+        // must include our channel. Since we can't query the dedicated
+        // listen connection directly, the next-best assertion is that an
+        // immediate notify is received.
+        const { promise: gotPayload, resolve: resolvePayload } = Promise.withResolvers<string>();
+        await db.listen("test_onlisten_timing", p => resolvePayload(p));
+        await db.notify("test_onlisten_timing", "after_onlisten");
+        expect(await gotPayload).toBe("after_onlisten");
+
+        await db.unlisten("test_onlisten_timing");
+      });
+    });
   }); // Close "PostgreSQL tests" describe
 } // Close if (isDockerEnabled())