Setting the file. One moment.
Subchapter 6.5
references/subscribing.mdMarkdown25 KBView on GitHub
Canonical page: https://neon.com/docs/realtime/guides/subscribing.md (opens in a new tab)
Summary: Subscribe to live queries from trusted backend code or the browser. Fetch sealed queries from your API, handle every callback and renew sealed queries before they expire.
Subscribe to live queries using the Realtime client.
Once you’ve defined and sealed a query, you can subscribe to it using the low‑level Realtime client (opens in a new tab).
Important: Client-side reactivity
Manual subscriptions are a relatively low-level API. Most apps and agents can use the higher‑level reactivity primitives that manage the subscription for you.
Workers, trusted agents and other backend processes can subscribe to a query directly.
However, to subscribe to a live query in an untrusted environment, you’ll need to expose a sealed query from your backend, either a standard API endpoint or using a server function (where SSR can embed the sealed query and the initial results).
Tip: Running example
The guide continues the running example (opens in a new tab) introduced in the Defining queries guide.
Add a url to createRealtime() (opens in a new tab) and it returns a direct server, which seals queries with your secret and subscribes to them itself.
import { createRealtime } from "@neon/realtime/server"
import { drizzleAdapter } from "@neon/realtime-drizzle"
import { messagesQuery } from "./queries"
// The worker acts for one known user, so its parameters are trusted constants
const SUPPORT_BOT_ID = "support-bot"
const SUPPORT_CHANNEL_ID = 2
const realtime = createRealtime({
url: process.env.NEON_REALTIME_URL!,
secret: process.env.NEON_REALTIME_SECRET!,
db: "neondb",
adapter: drizzleAdapter()
})
const query = messagesQuery(SUPPORT_BOT_ID, SUPPORT_CHANNEL_ID)
const subscription = await realtime.subscribe(query)
subscription.onChange(console.log)
process.once("SIGTERM", () => realtime.close())Warning: Untrusted parameters
Never fill a direct subscription’s parameters from untrusted input.
NEON_REALTIME_URL is written by neon env pull --service realtime, or shown on the Realtime page (opens in a new tab) in the Neon Console. Note the await: the direct server’s subscribe() is async, because it seals the query locally before it subscribes. The browser client’s subscribe() is synchronous. Otherwise it’s the same subscription, with the same options and callbacks as the client’s.
The direct server renews its own sealed queries and recovers from outages until you close the subscription or it moves to error. An error is final, so a long-running worker should watch for it with onStateChange and subscribe again. realtime.close() closes every subscription and the shared WebSocket. The direct server opens its WebSocket with the ws package, which you installed in setup, even on Node.js 22 and later.
Use direct subscriptions for workers, trusted agents and server-side fan-out.
Most queries depend on who’s asking and what they’re asking for.
If you’ve exposed your sealed query via an API endpoint, you can provide this context via the request data using your standard auth machinery, such as Neon’s Managed Better Auth (opens in a new tab):
import { createAuthClient } from "@neon/auth"
export const authClient = createAuthClient(import.meta.env.VITE_NEON_AUTH_URL)In our running example (opens in a new tab), we create a typed client for the API with Hono’s hc. It takes its routes from AppType, so every call is checked against the backend:
import type { SealedLiveQuery } from "@neon/realtime/client"
import { hc } from "hono/client"
import type { Message } from "../backend/schema"
import type { AppType } from "../backend/server"
import { authClient } from "./auth"
export const server = hc<AppType>(window.location.origin, {
headers: async (): Promise<Record<string, string>> => {
const { data } = await authClient.getSession()
const token = data?.session.token
return token ? { Authorization: `Bearer ${token}` } : {}
}
})
export async function fetchSealedMessages(channelId: number): Promise<SealedLiveQuery<Message>> {
const res = await server.api.v1.messages.live.$post({ json: { channelId } })
if (!res.ok) {
throw new Error(`Could not fetch the sealed query (${res.status})`)
}
return (await res.json()) as SealedLiveQuery<Message>
}Hono’s client types the response body as JSON, so the Date in created_at shows up as a string. The as restores the row type that the Realtime client decodes at runtime.
getSession() returns the cached session until its JWT is about to expire, then fetches a new one, so each request carries a valid token. Managed Better Auth tokens only last about 15 minutes. Without a session there’s no token, so the API answers 401.
window.location.origin assumes your dev server proxies /api to the API, as the example’s Vite config (opens in a new tab) does.
If your framework supports server functions (opens in a new tab), you can expose the sealed query from one instead of an API endpoint. Client code calls a server function like a local async function, so it can take the place of fetchSealedMessages() above. With SSR, the page can also embed the sealed query and the initial query results on the server.
Server functions run on your app’s origin, so they read the session from its cookie. See Exposing queries, Using server functions, the Server-side rendering guide and the TanStack (opens in a new tab) example for more details.
Important: Next.js uses Route Handlers
In Next.js, server functions are server actions, which run one at a time, so renew through a Route Handler (opens in a new tab) instead, as the Next.js (opens in a new tab) example shows.
Create one client for the whole app, because every subscription shares its connection:
import { createRealtimeClient } from "@neon/realtime/client"
import { drizzleParsers } from "@neon/realtime-drizzle/client"
export const client = createRealtimeClient({
url: import.meta.env.VITE_NEON_REALTIME_URL,
parsers: { ...drizzleParsers }
})The client opens its WebSocket lazily, on the first subscription, and multiplexes every subscription over that one connection. The first sealed query binds the connection to its database, so every subscription on a client must target the same db. The connection carries no user identity. Each subscription brings its own sealed query.
Now fetch the sealed query, subscribe to it and render every change:
import type { LiveQuerySnapshot, LiveQueryState } from "@neon/realtime/client"
import type { Message } from "../backend/schema"
import { fetchSealedMessages } from "./api"
import { client } from "./client"
const list = document.querySelector<HTMLUListElement>("#messages")!
const statusLine = document.querySelector<HTMLParagraphElement>("#status")!
// A real app takes the channel from its router
const channelId = 2
const sealedQuery = await fetchSealedMessages(channelId)
const subscription = client.subscribe(sealedQuery)
subscription.onChange(render)
window.addEventListener("beforeunload", () => {
subscription.unsubscribe()
client.close()
})
function render(snapshot: LiveQuerySnapshot<Message>) {
statusLine.textContent = describe(snapshot)
list.replaceChildren(...(snapshot.data ?? []).map(renderMessage))
}
function describe(state: LiveQueryState): string {
switch (state.status) {
case "connecting":
return "Loading messages"
case "live":
return "Up to date"
case "stale":
return "Updating. Messages may be out of date."
case "error":
return `Stopped: ${state.error.message}`
case "closed":
return "Closed"
}
}
function renderMessage(message: Message): HTMLLIElement {
const item = document.createElement("li")
item.textContent = `${message.created_at.toLocaleTimeString()} ${message.body}`
return item
}subscribe() returns a materialized subscription, which keeps the current rows for you. onChange gets a snapshot of the rows and the subscription’s state whenever either changes, so the render reads data, status and error from one consistent value. data is undefined until the first result arrives.
Rows arrive decoded, so created_at is a Date, as Drizzle’s Message type says. The client’s default parsers decode these Postgres types:
| Postgres type | Arrives as |
|---|---|
bool | boolean |
int2, int4, oid, float4, float8 | number |
json, jsonb | The parsed JSON value |
timestamptz | Date |
timestamp | Date in local time |
date | Date at local midnight |
bytea | Uint8Array |
int8, numeric | string, so their values stay exact |
Everything else, such as text and uuid | string, as Postgres formats it |
NULL arrives as null, and arrays of built-in types decode element by element.
The client above spreads drizzleParsers, so rows match Drizzle’s types: date stays a string and timestamp is read as UTC.
When you’re done with a subscription, call unsubscribe(). It moves the subscription to closed, so any listeners get a final closed update, and then stops all further callbacks. client.close() does the same for every subscription on the client, and closes the WebSocket.
A subscription has three callbacks. Each one takes a listener and returns a function that removes it:
| Callback | Listener arguments | Fires |
|---|---|---|
onChange | snapshot: LiveQuerySnapshot<Row> | After each state change, after each changes update while live, and when a new result is complete. Materialized subscriptions only. |
onStateChange | state: LiveQueryState | When the status changes. |
onUpdate | update: RawLiveQueryUpdate<Row> | With type: "rows" and rows, each { rowId, row }, when a complete, authoritative result replaces the previous one. With type: "changes", changes and batch: LiveQueryBatchInfo when committed transactions change the result. Each change is an upsert with a rowId and row, or a remove with only a rowId. |
Listeners run in the order you add them, and one that throws doesn’t stop the others. Each initial result and each committed transaction becomes visible atomically. Outside a listener, getSnapshot() and getState() return the current values.
The status tells you how far to trust the rows:
| Status | Meaning |
|---|---|
connecting | Waiting for a result: when it starts, including with initialData from a server render, after a reconnect, and after renewing an expired sealed query. Rows you already have stay. |
live | The rows are authoritative and changes are streaming. It stays live while the server replaces the whole result. |
stale | The sealed query expired and the subscription is waiting for a renewed one. The rows are usable but may be out of date. Render them normally. |
error | Automatic recovery has stopped. See handle errors. |
closed | Closed by unsubscribe() or client.close(). It doesn’t reopen. |
By default, subscriptions are materialized. The client keeps the rows and applies each update before onUpdate fires, so getSnapshot() in an update listener already includes it. If your code keeps its own rows, as TanStack DB does, pass { materialize: false } for a raw subscription. It has the same callbacks except onChange, and no getSnapshot(). Keep the rows in a map keyed by rowId, replace them on each rows update and apply each changes update’s upserts and removes:
import type { LiveQueryState, RawLiveQueryUpdate } from "@neon/realtime/client"
import type { Message } from "../backend/schema"
import { fetchSealedMessages } from "./api"
import { client } from "./client"
const channelId = 2
// Keyed by rowId, which is only valid until the next `rows` update
const rows = new Map<string, Message>()
const subscription = client.subscribe(await fetchSealedMessages(channelId), { materialize: false })
subscription.onUpdate((update: RawLiveQueryUpdate<Message>) => {
if (update.type === "rows") {
rows.clear()
for (const { rowId, row } of update.rows) {
rows.set(rowId, row)
}
return
}
for (const change of update.changes) {
if (change.type === "upsert") {
rows.set(change.rowId, change.row)
} else {
rows.delete(change.rowId)
}
}
console.log(`${rows.size} messages after transactions ${update.batch.txids.join(", ")}`)
})
subscription.onStateChange((state: LiveQueryState) => {
console.log(state.status, state.error?.message ?? "")
})A rowId is opaque and scoped to its subscription, and each rows update invalidates every earlier one. It isn’t a primary key, so don’t store it or match it against your own ids. A rows update can arrive at any time, for example after a reconnect, and it always replaces the whole result.
Each changes update also carries the ids of the Postgres transactions it covers, in update.batch.txids. The mutations guide uses them to confirm optimistic mutations.
Sealed queries are short-lived, so a subscription that outlives its sealed query needs a fresh one. Fetch a replacement before expiresAt and pass it to renew(). This scheduler renews 10 seconds before expiry, retries a failed fetch after 1 second, reschedules from each new expiresAt and stops if renew() fails:
import type { RawLiveQuerySubscription, SealedLiveQuery } from "@neon/realtime/client"
// The same timing the React and TanStack DB bindings use
const RENEW_EARLY_MS = 10_000
const RETRY_MS = 1_000
export function renewBeforeExpiry<Row>(
subscription: RawLiveQuerySubscription<Row>,
sealedQuery: SealedLiveQuery<Row>,
fetchSealedQuery: () => Promise<SealedLiveQuery<Row>>
): () => void {
let timer: ReturnType<typeof setTimeout> | undefined
let stopped = false
function schedule(delay: number) {
clearTimeout(timer)
if (!stopped) {
timer = setTimeout(renew, delay)
}
}
function scheduleBeforeExpiry(expiresAt: number) {
schedule(Math.max(0, expiresAt - Date.now() - RENEW_EARLY_MS))
}
async function renew() {
let next: SealedLiveQuery<Row>
try {
next = await fetchSealedQuery()
} catch {
schedule(RETRY_MS)
return
}
scheduleBeforeExpiry(next.expiresAt)
// A rejected renewal means the subscription is in `error` or the query changed
subscription.renew(next).catch(stop)
}
function stop() {
stopped = true
clearTimeout(timer)
}
scheduleBeforeExpiry(sealedQuery.expiresAt)
return stop
}Start it after subscribing, with a function that fetches the same query again, and stop it when you unsubscribe:
import type { LiveQuerySnapshot, LiveQueryState } from "@neon/realtime/client"
import type { Message } from "../backend/schema"
import { fetchSealedMessages } from "./api"
import { client } from "./client"
import { renewBeforeExpiry } from "./renewal"
const list = document.querySelector<HTMLUListElement>("#messages")!
const statusLine = document.querySelector<HTMLParagraphElement>("#status")!
// A real app takes the channel from its router
const channelId = 2
const sealedQuery = await fetchSealedMessages(channelId)
const subscription = client.subscribe(sealedQuery)
const stopRenewing = renewBeforeExpiry(subscription, sealedQuery, () => fetchSealedMessages(channelId))
subscription.onChange(render)
window.addEventListener("beforeunload", () => {
stopRenewing()
subscription.unsubscribe()
client.close()
})
function render(snapshot: LiveQuerySnapshot<Message>) {
statusLine.textContent = describe(snapshot)
list.replaceChildren(...(snapshot.data ?? []).map(renderMessage))
}
function describe(state: LiveQueryState): string {
switch (state.status) {
case "connecting":
return "Loading messages"
case "live":
return "Up to date"
case "stale":
return "Updating. Messages may be out of date."
case "error":
return `Stopped: ${state.error.message}`
case "closed":
return "Closed"
}
}
function renderMessage(message: Message): HTMLLIElement {
const item = document.createElement("li")
item.textContent = `${message.created_at.toLocaleTimeString()} ${message.body}`
return item
}renew() replaces the sealed query for the same exact query, and it keeps the subscription object, its listeners and its rows:
live, with no new rows update.stale. renew() moves it to connecting while the client subscribes again behind the same object, and to live when the new result arrives.If the replacement’s queryFingerprint doesn’t match, the promise that renew() returns rejects. A changed parameter, like a different channel, makes a different query, so subscribe to the new one and unsubscribe from the old.
You don’t write this yourself when you use the bindings. The React and TanStack DB bindings take a refreshQuery function and renew with the same timing, 10 seconds before expiry with a 1-second retry. They’re built on QueryRefreshController (opens in a new tab), which you can use if you’re writing a binding for another framework.
When automatic recovery stops, the subscription moves to error, and its error is a LiveQueryError (opens in a new tab) with:
code: a stable, machine-readable error codemessage: a description of the failureretryable: whether reconnecting or a fresh sealed query may recover itsqlState: the Postgres SQLSTATE code, when debug mode is on and the database supplied onecause: the underlying error, when there is oneA subscription never leaves error by itself, and renew() can’t bring it back. Unsubscribe, show the error, and subscribe again with a fresh sealed query, for example when the user retries.
Errors stay with their own subscription. A row that fails to decode, for example, puts only its own subscription into error, and the others on the client carry on.
If the connection itself fails, for example because of a fatal protocol error, every subscription on the client moves to error at once and the client can’t reconnect. Create a new client with createRealtimeClient() before you subscribe again.
The client is silent by default. To see what it’s doing, pass a logLevel to createRealtimeClient() (opens in a new tab): "error", "warn", "info" or "debug". Each level includes the ones before it.
import { createRealtimeClient } from "@neon/realtime/client"
import { drizzleParsers } from "@neon/realtime-drizzle/client"
export const client = createRealtimeClient({
url: import.meta.env.VITE_NEON_REALTIME_URL,
parsers: { ...drizzleParsers },
logLevel: import.meta.env.DEV ? "debug" : "warn"
})Entries go to the console unless you pass a logger (opens in a new tab), which receives each entry (opens in a new tab) instead. An entry has a level, a message, a timestamp and a stable event name, like connection_lost or subscription_failed. Match on event, not on message:
logger: (entry) => {
if (entry.event === "subscription_failed") reportError(entry.error)
}Entries never include sealed queries, SQL, parameters or rows. An entry’s error can hold an error from the server or from your own code, so handle it like any other error you log.
The React and TanStack DB bindings log through the client you give them. A direct server takes the same logLevel and logger options in createRealtime({ url }).
By default, subscription errors carry safe, generic messages. To see the database’s own error while you develop, set debugMode on the server that seals your queries:
import { createRealtime } from "@neon/realtime/server"
import { drizzleAdapter } from "@neon/realtime-drizzle"
export const realtime = createRealtime({
secret: process.env.NEON_REALTIME_SECRET!,
db: "neondb",
adapter: drizzleAdapter(),
debugMode: process.env.NODE_ENV === "development"
})Every query it seals then reports errors with the database’s full message, plus its SQLSTATE code in sqlState when there is one. It’s a backend option because it travels inside each sealed query.
Warning: Development only
Full error details can expose schema, table and column names to anyone who subscribes, including through any error message your UI shows. Never turn on debugMode in production.