neon-realtime

<!-- Generated by scripts/sync-skill-docs.mjs from website@thruflo/realtime-2-docs:24a2fd708 (docs/realtime/guides/mutations). Do not edit: change the docs and run npm run sync:docs. sha256:14dae781f49028d99a30a01a01614a6432e3e0de7a6acc5c0b61b06c4824c7fa -->
Canonical page: https://neon.com/docs/realtime/guides/mutations.md

> Summary: Write through your own backend, return the transaction id, apply the change optimistically and drop the optimistic state once the change arrives on the live query. Without a framework, with React useOptimistic or with TanStack DB.

# Optimistic mutations

Make apps and agents super fast, with instant local writes.

## Contents

- [How optimistic mutations work](#how-optimistic-mutations-work)
  - [Automatic management of optimistic state](#automatic-management-of-optimistic-state)
- [Running example](#running-example)
  - [Returning the `txid` from your backend](#returning-the-txid-from-your-backend)
  - [Calling the routes from the frontend](#calling-the-routes-from-the-frontend)
  - [Shared mutation helpers](#shared-mutation-helpers)
- [Using optimistic mutations](#using-optimistic-mutations)
  - [Without a framework](#without-a-framework)
  - [With React's `useOptimistic` hook](#with-reacts-useoptimistic-hook)
  - [With TanStack DB collections](#with-tanstack-db-collections)
- [Render the app](#render-the-app)

Apply instant local writes using [optimistic mutations](#how-optimistic-mutations-work) that [automatically discard their optimistic state](#automatic-management-of-optimistic-state) when the changes [sync in through your live query subscriptions](subscribing.md#listen-for-changes).

## How optimistic mutations work

Normally, with online systems, you first send writes to the server and then display them to the user once the server has written to the database and responded successfully. Depending on network connectivity and system uptime, this can be slow or broken.

With optimistic mutations, you first display the write to the user and only then send to the server in the background. This makes apps feel super-fast and responsive, as well as resilient to connectivity or downtime issues.

### Automatic management of optimistic state

Realtime allows you to automate the management and lifecycle of your optimistic state, by monitoring the transactions coming through your live query subscriptions. For each write:

1. apply the change to local, optimistic state, so the UI updates straight away
2. send the write through your backend, with your normal validation, authorization and backend logic, which returns the [ID of the transaction](https://www.postgresql.org/docs/current/transaction-id.html) that made it
3. wait for that transaction ID to arrive on the subscription using [`awaitTxId()`](https://neon.com/docs/realtime/sdks/typescript.md#client-materializedlivequerysubscriptionawaittxid)
4. drop the optimistic state
5. if the write fails, or its transaction ID doesn't arrive before the timeout, drop the optimistic state and show an error

`awaitTxId()` remembers recently seen transaction IDs, so it resolves even when the change arrives before your backend's HTTP response. It also resolves after the subscription's `onChange`, so the synced rows already include the change when the optimistic state drops, which means nothing flickers.

## Running example

Let's return to our running example, flesh out the backend server, call its new routes from the frontend and add some shared frontend helpers.

### Returning the `txid` from your backend

Each write route takes a batch of changes and writes the whole batch in one SQL statement, which returns the ID of its transaction. So there's one txid to wait for however many rows change, and you get it without another database round trip.

Add the three write routes to `backend/server.ts`, chained onto the same app so `AppType` carries them to the frontend. They follow the `/api/v1/messages/live` route from the [queries guide](queries.md#exposing-an-endpoint) and the `/api/v1/inbox/live` route from the [reactivity guide](reactivity.md#seal-a-broader-query):

```ts filename="backend/server.ts"
import { and, eq, inArray, ne, sql } from "drizzle-orm"
import { Hono } from "hono"
import { HTTPException } from "hono/http-exception"
import { validator } from "hono/validator"

import { requireUser } from "./auth"
import { db } from "./db"
import { channelMembers, messages } from "./schema"

export type MutationResponse =
  | { readonly kind: "committed"; readonly txid: string }
  | { readonly kind: "unchanged" }

// The id of the transaction that made the change, read in the write's own RETURNING
const txid = sql<string>`pg_current_xact_id()::text`

const mutationResponse = (row?: { txid: string }): MutationResponse =>
  row ? { kind: "committed", txid: row.txid } : { kind: "unchanged" }

const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i

function isUuid(value: unknown): value is string {
  return typeof value === "string" && UUID.test(value)
}

function isBody(value: unknown): value is string {
  return typeof value === "string" && value.trim() !== ""
}

const app = new Hono()
  // ...the sealing routes: /api/v1/messages/live and /api/v1/inbox/live
  .post(
    "/api/v1/messages",
    validator("json", (value, c) => {
      const batch: unknown = value?.batch

      if (
        !Array.isArray(batch) ||
        batch.length === 0 ||
        !batch.every(
          (item): item is { id: string; channelId: number; body: string } =>
            isUuid(item?.id) && Number.isInteger(item?.channelId) && isBody(item?.body)
        )
      ) {
        return c.json({ error: "batch must be a non-empty array of { id, channelId, body }" }, 400)
      }

      return { batch }
    }),
    async (c) => {
      const user = await requireUser(c)
      const { batch } = c.req.valid("json")
      const channelIds = [...new Set(batch.map((message) => message.channelId))]
      const rows = batch.map(({ id, channelId, body }) => ({ id, channel_id: channelId, body }))

      // One statement: the batch is written only if the user is a member of every channel in it
      const [sent] = await db
        .insert(messages)
        .select(
          db
            .select({
              id: sql`batch.id`.as("id"),
              channel_id: sql`batch.channel_id`.as("channel_id"),
              author_id: sql`${user.id}`.as("author_id"),
              body: sql`batch.body`.as("body"),
              created_at: sql`now()`.as("created_at"),
            })
            .from(sql`jsonb_to_recordset(${JSON.stringify(rows)}::jsonb) as batch(id uuid, channel_id integer, body text)`)
            .where(
              eq(
                db.$count(
                  channelMembers,
                  and(eq(channelMembers.user_id, user.id), inArray(channelMembers.channel_id, channelIds))
                ),
                channelIds.length
              )
            )
        )
        .returning({ txid })

      if (!sent) {
        throw new HTTPException(403, { message: "Not a member of this channel" })
      }

      return c.json(mutationResponse(sent))
    }
  )
  .patch(
    "/api/v1/messages",
    validator("json", (value, c) => {
      const batch: unknown = value?.batch

      if (
        !Array.isArray(batch) ||
        batch.length === 0 ||
        !batch.every((item): item is { id: string; body: string } => isUuid(item?.id) && isBody(item?.body))
      ) {
        return c.json({ error: "batch must be a non-empty array of { id, body }" }, 400)
      }

      return { batch }
    }),
    async (c) => {
      const user = await requireUser(c)
      const { batch } = c.req.valid("json")

      const [edited] = await db
        .update(messages)
        .set({ body: sql`edit.body` })
        .from(sql`jsonb_to_recordset(${JSON.stringify(batch)}::jsonb) as edit(id uuid, body text)`)
        .where(and(eq(messages.id, sql`edit.id`), eq(messages.author_id, user.id), ne(messages.body, sql`edit.body`)))
        .returning({ txid })

      return c.json(mutationResponse(edited))
    }
  )
  .delete(
    "/api/v1/messages",
    validator("json", (value, c) => {
      const batch: unknown = value?.batch

      if (!Array.isArray(batch) || batch.length === 0 || !batch.every(isUuid)) {
        return c.json({ error: "batch must be a non-empty array of uuids" }, 400)
      }

      return { batch }
    }),
    async (c) => {
      const user = await requireUser(c)
      const { batch } = c.req.valid("json")

      const [deleted] = await db
        .delete(messages)
        .where(and(inArray(messages.id, batch), eq(messages.author_id, user.id)))
        .returning({ txid })

      return c.json(mutationResponse(deleted))
    }
  )

export type AppType = typeof app

export default app
```

Return the txid from the write itself. Each route puts a `txid` field in its write's `RETURNING`, and that field reads [`pg_current_xact_id()`](https://www.postgresql.org/docs/current/functions-info.html#FUNCTIONS-INFO-SNAPSHOT). A single statement runs in its own transaction, so that's the ID of the transaction that made the change. It comes back with the write, in one round trip, and it works with any Postgres driver. The `::text` cast returns the txid as a decimal string, so its 64-bit value stays exact in JavaScript. Every row a statement returns carries the same txid, so each route takes the first.

Write each batch in one statement, so there's one txid:

- **Send** passes the whole batch as one JSON parameter, and `jsonb_to_recordset()` turns it into typed rows (`id uuid, channel_id integer, body text`) that the insert selects from. The `WHERE` counts the user's `channel_members` rows for the batch's channels and compares the count with the number of distinct channels. If the user isn't a member of every channel, nothing's written and no row comes back, so the route throws a 403. It's all or nothing.
- **Edit** runs `UPDATE … FROM jsonb_to_recordset(…)`, which joins each message to its new body. It only changes messages the user wrote, so only the author can edit a message, and the `<>` check skips edits that leave the text as it was.
- **Delete** runs `DELETE … WHERE id IN (…)`, restricted to messages the user wrote, so only the author can delete a message.

In Drizzle, the send's `insert().select()` needs every column in table order (so it selects `created_at` as `now()`), and each `sql` field needs an `.as()` alias.

**Warning: Don't open a transaction just to read the txid**

Wrapping a write in `db.transaction()` and then running `select pg_current_xact_id()` costs four round trips (`BEGIN`, the write, the select and `COMMIT`) and holds a pooled connection for all of them. When a write genuinely needs several statements, keep the transaction and put the txid in the `RETURNING` of a write that always changes a row. Every statement in a transaction shares its txid, but a write that matches nothing returns no row, and so no txid. The [Mastra example](https://neon.com/docs/realtime/examples/mastra.md#start-agents-from-your-writes) shows this with two inserts.

When an edit or a delete batch matches no rows, the statement returns no row, so `mutationResponse()` returns `{ kind: "unchanged" }` instead of a txid. That happens when every edit leaves the text as it was, or when every message is already gone or someone else wrote it. Nothing changed, so there's nothing to wait for.

**Important: Always pass a timeout**

`awaitTxId()` keeps waiting while the subscription reconnects or waits for a renewed sealed query, and it never settles if the txid doesn't reach the query. With a timeout, it rejects, the optimistic change drops and the UI shows that the change couldn't be confirmed. The write may still have committed. It also rejects if the subscription errors or closes.

### Calling the routes from the frontend

Add a helper for each route to `frontend/api.ts`, next to `fetchSealedMessages()` from the [subscribing guide](subscribing.md#fetch-from-an-api-endpoint). Add `MutationResponse` to the file's type import from `../backend/server`, so each helper returns the route's own response type:

```ts filename="frontend/api.ts"
import type { AppType, MutationResponse } from "../backend/server"

export async function sendMessages(
  batch: Pick<Message, "id" | "channel_id" | "body">[]
): Promise<MutationResponse> {
  const res = await server.api.v1.messages.$post({
    json: { batch: batch.map(({ id, channel_id, body }) => ({ id, channelId: channel_id, body })) }
  })

  if (!res.ok) {
    throw new Error(`Could not send the messages (${res.status})`)
  }

  return res.json()
}

export async function editMessages(batch: { id: string; body: string }[]): Promise<MutationResponse> {
  const res = await server.api.v1.messages.$patch({ json: { batch } })

  if (!res.ok) {
    throw new Error(`Could not edit the messages (${res.status})`)
  }

  return res.json()
}

export async function deleteMessages(ids: string[]): Promise<MutationResponse> {
  const res = await server.api.v1.messages.$delete({ json: { batch: ids } })

  if (!res.ok) {
    throw new Error(`Could not delete the messages (${res.status})`)
  }

  return res.json()
}
```

Each helper takes an array, so one call can send any number of changes. The `hc` client types each call from its route's validators, so a wrong field is a compile error.

### Shared mutation helpers

All three approaches below share the helpers in `frontend/mutations.ts`: the mutation and message types, `newMessage()`, the `applyMutation()` reducer and `writeMutation()`, which sends one mutation through the route helpers you just added to `frontend/api.ts`, as a batch of one:

```ts filename="frontend/mutations.ts"
import type { Message } from "../backend/schema"
import type { MutationResponse } from "../backend/server"
import { deleteMessages, editMessages, sendMessages } from "./api"

export type MessageMutation =
  | { readonly type: "send"; readonly message: Message }
  | { readonly type: "edit"; readonly id: string; readonly body: string }
  | { readonly type: "delete"; readonly id: string }

export type OptimisticMessage = Message & {
  readonly pending?: boolean;
}

// The server sets its own author and timestamp, but keeps the client's id
export function newMessage(channelId: number, authorId: string, body: string): Message {
  return { id: crypto.randomUUID(), channel_id: channelId, author_id: authorId, body, created_at: new Date() }
}

export function applyMutation(
  messages: readonly OptimisticMessage[],
  mutation: MessageMutation
): readonly OptimisticMessage[] {
  switch (mutation.type) {
    case "send":
      // Once the sent message has synced, show the synced row only
      if (messages.some((message) => message.id === mutation.message.id)) {
        return messages
      }

      // Newest first, to match the query's order
      return [{ ...mutation.message, pending: true }, ...messages]
    case "edit":
      return messages.map((message) =>
        message.id === mutation.id ? { ...message, body: mutation.body, pending: true } : message
      )
    case "delete":
      return messages.filter((message) => message.id !== mutation.id)
  }
}

export function writeMutation(mutation: MessageMutation): Promise<MutationResponse> {
  switch (mutation.type) {
    case "send":
      return sendMessages([mutation.message])
    case "edit":
      return editMessages([{ id: mutation.id, body: mutation.body }])
    case "delete":
      return deleteMessages([mutation.id])
  }
}
```

A send carries the whole new row, with an id the client picks with `crypto.randomUUID()`. The synced row has the same id, so the optimistic row never needs swapping for a real one, and `applyMutation()` skips a pending send once the synced rows include it. That way, a sent message never shows twice.

## Using optimistic mutations

Pick [without a framework](#without-a-framework) to see the mechanism or for apps that don't use React, [React's `useOptimistic` hook](#with-reacts-useoptimistic-hook) for React apps that keep optimistic state locally in a component, or [TanStack DB collections](#with-tanstack-db-collections) for apps that sync into a normalized client store.

### Without a framework

Without a framework, you keep the optimistic state yourself. `frontend/optimistic-messages.ts` keeps an overlay of pending mutations on top of the subscription's synced rows, and merges the two every time either side changes:

```ts filename="frontend/optimistic-messages.ts"
import type { LiveQueryState, MaterializedLiveQuerySubscription } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { applyMutation, type MessageMutation, type OptimisticMessage, writeMutation } from "./mutations"

export function optimisticMessages(
  subscription: MaterializedLiveQuerySubscription<Message>,
  render: (messages: readonly OptimisticMessage[], state: LiveQueryState, error: string | undefined) => void
) {
  // A Map keeps insertion order, so mutations apply in the order they were made
  const pending = new Map<number, MessageMutation>()
  let nextKey = 0
  let error: string | undefined
  let stopped = false

  function update() {
    if (stopped) {
      return
    }

    const snapshot = subscription.getSnapshot()
    const synced: readonly OptimisticMessage[] = snapshot.data ?? []

    render([...pending.values()].reduce(applyMutation, synced), snapshot, error)
  }

  async function mutate(mutation: MessageMutation) {
    const key = nextKey++

    error = undefined
    pending.set(key, mutation)
    update()

    try {
      const result = await writeMutation(mutation)

      if (result.kind === "committed") {
        await subscription.awaitTxId(result.txid, 10_000)
      }
    } catch {
      error = "Couldn't confirm the change"
    } finally {
      pending.delete(key)
      update()
    }
  }

  const unsubscribe = subscription.onChange(update)

  update()

  return {
    mutate,
    stop: () => {
      stopped = true
      unsubscribe()
    }
  }
}
```

Each mutation leaves the overlay once its txid arrives, or once it fails. Either way, the list falls back to the synced rows, which by then either include the change or show that it didn't happen.

Call `optimisticMessages()` with the subscription you created in the [subscribing guide](subscribing.md#subscribe-to-the-sealed-query) and a function that draws the list. It replaces `subscription.onChange(render)` from that guide, and passes the subscription's state through, so the page can show its status line as before. Call `mutate()` from your UI, with `newMessage()` for a send. It needs the signed-in user's id, which you read from the [Managed Better Auth](https://neon.com/docs/auth/overview.md) session, as `frontend/main.tsx` does in [render the app](#render-the-app). Show Edit and Delete only on the user's own messages, and disable them on pending messages, so you don't start a new change on a message whose last change isn't confirmed yet. The routes still enforce authorship. Call `stop()` before you unsubscribe.

### With React's `useOptimistic` hook

In React, `useOptimistic` holds the overlay for you and each write runs in its own transition. This is the optimistic version of `frontend/Messages.tsx` from the [reactivity guide](reactivity.md#uselivequery). It runs the same live query and takes the same props, plus `channelId` and the signed-in user's `userId`, which `Channel` passes in (shown after the component). It reuses the types and helpers from `frontend/mutations.ts`, and `LocalTime` from the same guide:

```tsx filename="frontend/Messages.tsx"
import { type FormEvent, startTransition, useEffect, useOptimistic, useState } from "react"
import { useLiveQuery } from "@neon/realtime-react"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { LocalTime } from "./LocalTime"
import {
  applyMutation,
  type MessageMutation,
  newMessage,
  type OptimisticMessage,
  writeMutation
} from "./mutations"

const NO_MESSAGES: readonly OptimisticMessage[] = []

interface Props {
  sealedQuery: SealedLiveQuery<Message>;
  refreshQuery: () => Promise<SealedLiveQuery<Message>>;
  initialData?: readonly Message[];
  channelId: number;
  userId: string;
}

export function Messages({ sealedQuery, refreshQuery, initialData, channelId, userId }: Props) {
  const { data, status, error, utils } = useLiveQuery(sealedQuery, { refreshQuery, initialData })

  const [messages, addOptimistic] = useOptimistic<readonly OptimisticMessage[], MessageMutation>(
    data ?? NO_MESSAGES,
    applyMutation
  )

  const [mutationError, setMutationError] = useState<string>()

  useEffect(() => {
    return utils.onUpdate((update) => {
      if (update.type === "changes") {
        console.debug(`Applied ${update.changes.length} changes from transactions ${update.batch.txids.join(", ")}`)
      }
    })
  }, [utils])

  function mutate(mutation: MessageMutation) {
    setMutationError(undefined)

    startTransition(async () => {
      addOptimistic(mutation)

      try {
        const result = await writeMutation(mutation)

        if (result.kind === "committed") {
          await utils.awaitTxId(result.txid, 10_000)
        }
      } catch {
        setMutationError("Couldn't confirm the change")
      }
    })
  }

  function send(event: FormEvent<HTMLFormElement>) {
    event.preventDefault()

    const form = event.currentTarget
    const body = String(new FormData(form).get("body") ?? "").trim()

    if (body) {
      form.reset()
      mutate({ type: "send", message: newMessage(channelId, userId, body) })
    }
  }

  function edit(message: OptimisticMessage) {
    const body = window.prompt("Edit message", message.body)?.trim()

    if (body) {
      mutate({ type: "edit", id: message.id, body })
    }
  }

  if (status === "error") {
    return <p role="alert">Live updates stopped: {error.message}</p>
  }

  if (data === undefined) {
    return <p>Loading messages</p>
  }

  return (
    <section>
      <form onSubmit={send}>
        <input name="body" placeholder="Message" required />
        <button type="submit">Send</button>
      </form>

      {status === "stale" && <p role="status">Updating</p>}
      {mutationError && <p role="alert">{mutationError}</p>}

      <ul>
        {messages.map((message) => (
          <li key={message.id} aria-busy={message.pending}>
            <LocalTime date={message.created_at} />
            <p>{message.body}</p>
            {message.author_id === userId && (
              <>
                <button disabled={message.pending} onClick={() => edit(message)} type="button">
                  Edit
                </button>
                <button
                  disabled={message.pending}
                  onClick={() => mutate({ type: "delete", id: message.id })}
                  type="button"
                >
                  Delete
                </button>
              </>
            )}
          </li>
        ))}
      </ul>
    </section>
  )
}
```

`mutate()` runs each mutation in its own transition, so independent writes run concurrently. The transition applies the mutation optimistically, writes through your backend and, if the write committed, waits for its txid with [`utils.awaitTxId()`](https://neon.com/docs/realtime/sdks/typescript.md#react-uselivequeryutils). When the transition ends, React drops its optimistic change. By then the change is in `data`, with the same id for a sent message, so the list doesn't flicker. If the write fails or isn't confirmed in time, the optimistic change drops the same way, the message reverts and the error line shows.

Edit and Delete show only on the user's own messages, and the routes still enforce authorship. They're disabled on pending messages, so you don't start a new change on a message whose last change isn't confirmed yet.

In `frontend/Channel.tsx` from the [reactivity guide](reactivity.md#changing-query-parameters), give `Channel` a `userId` prop and pass it to `Messages` with `channelId`. Read `userId` from the [Managed Better Auth](https://neon.com/docs/auth/overview.md) session where you render `Channel`, as `frontend/main.tsx` does in [render the app](#render-the-app):

```tsx filename="frontend/Channel.tsx"
// ...imports and useSealedQuery() as in the reactivity guide

export function Channel({ channelId, userId }: { channelId: number; userId: string }) {
  const sealed = useSealedQuery(channelId)
  // Renewals must be for the same query, so refresh with the same channel
  const refreshQuery = useCallback(() => fetchSealedMessages(channelId), [channelId])

  if (sealed?.sealedQuery) {
    return (
      <Messages
        sealedQuery={sealed.sealedQuery}
        refreshQuery={refreshQuery}
        channelId={channelId}
        userId={userId}
      />
    )
  }

  if (sealed?.error) {
    return <p role="alert">Couldn't open this channel</p>
  }

  return <p>Loading messages</p>
}
```

### With TanStack DB collections

For more control, keep optimistic state in a [TanStack DB](https://tanstack.com/db/latest/docs/overview) collection. Every query and component that reads the collection then sees the same optimistic changes, and they stay consistent across client-side joins. Failed writes roll back automatically, and one transaction can span several collections.

Add mutation handlers to `createMessagesCollection()` in `frontend/collections.ts`, from the [reactivity guide](reactivity.md#create-the-collection). TanStack DB calls each handler once per transaction, with all of its mutations, so each handler sends them to your backend in one request. If the write committed, the handler waits for its txid before it returns, so TanStack DB keeps the optimistic state until the change has synced:

```ts filename="frontend/collections.ts"
import { createCollection } from "@tanstack/react-db"
import { realtimeCollectionOptions } from "@neon/realtime-tanstack"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { deleteMessages, editMessages, fetchSealedInbox, sendMessages } from "./api"
import { client } from "./client"

export function createMessagesCollection(sealedQuery: SealedLiveQuery<Message>) {
  return createCollection(
    realtimeCollectionOptions({
      id: "messages",
      client,
      query: sealedQuery,
      refreshQuery: fetchSealedInbox,
      getKey: (message) => message.id,
      onInsert: async ({ transaction, collection }) => {
        const result = await sendMessages(transaction.mutations.map(({ modified }) => modified))

        if (result.kind === "committed") {
          await collection.utils.awaitTxId(result.txid, 10_000)
        }
      },
      onUpdate: async ({ transaction, collection }) => {
        const result = await editMessages(transaction.mutations.map(({ modified: { id, body } }) => ({ id, body })))

        if (result.kind === "committed") {
          await collection.utils.awaitTxId(result.txid, 10_000)
        }
      },
      onDelete: async ({ transaction, collection }) => {
        const result = await deleteMessages(transaction.mutations.map(({ original }) => original.id))

        if (result.kind === "committed") {
          await collection.utils.awaitTxId(result.txid, 10_000)
        }
      }
    })
  )
}

export type MessagesCollection = ReturnType<typeof createMessagesCollection>
```

While a handler runs, TanStack DB holds back synced changes. When the handler returns, it drops the optimistic state and applies the synced rows together, so a sent message's optimistic row gives way to its synced row, which has the same id, in a single step. If a handler throws, the optimistic change rolls back. [`collection.utils.awaitTxId()`](https://neon.com/docs/realtime/sdks/typescript.md#tanstack-db-realtimecollectionutilsawaittxid) resolves once the transaction has entered TanStack DB's sync queue, and it takes the same explicit timeout.

Then write through the collection. This is the optimistic version of `frontend/Inbox.tsx` from the [reactivity guide](reactivity.md#query-the-collection). It runs the same local query, and its `insert()`, `update()` and `delete()` calls apply each change locally at once and call the matching handler:

```tsx filename="frontend/Inbox.tsx"
import type { FormEvent } from "react"
import { eq, useLiveQuery as useCollectionQuery } from "@tanstack/react-db"

import type { Message } from "../backend/schema"
import type { MessagesCollection } from "./collections"
import { newMessage } from "./mutations"

interface Props {
  messagesCollection: MessagesCollection;
  channelId: number;
  userId: string;
}

export function Inbox({ messagesCollection, channelId, userId }: Props) {
  const { data, isReady, isError } = useCollectionQuery({
    query: (q) =>
      q
        .from({ message: messagesCollection })
        .where(({ message }) => eq(message.channel_id, channelId))
        .orderBy(({ message }) => message.created_at, "desc")
  })

  function send(event: FormEvent<HTMLFormElement>) {
    event.preventDefault()

    const form = event.currentTarget
    const body = String(new FormData(form).get("body") ?? "").trim()

    if (body) {
      form.reset()
      messagesCollection.insert(newMessage(channelId, userId, body))
    }
  }

  function edit(message: Message) {
    const body = window.prompt("Edit message", message.body)?.trim()

    if (body) {
      messagesCollection.update(message.id, (draft) => {
        draft.body = body
      })
    }
  }

  if (isError) {
    return <p role="alert">Couldn't sync messages</p>
  }

  if (!isReady) {
    return <p>Loading messages</p>
  }

  return (
    <section>
      <form onSubmit={send}>
        <input name="body" placeholder="Message" required />
        <button type="submit">Send</button>
      </form>

      <ul>
        {data.map((message) => (
          <li key={message.id} aria-busy={message.$hasPendingWrites}>
            {message.body}
            {message.author_id === userId && (
              <>
                <button disabled={message.$hasPendingWrites} onClick={() => edit(message)} type="button">
                  Edit
                </button>
                <button
                  disabled={message.$hasPendingWrites}
                  onClick={() => messagesCollection.delete(message.id)}
                  type="button"
                >
                  Delete
                </button>
              </>
            )}
          </li>
        ))}
      </ul>
    </section>
  )
}
```

`$hasPendingWrites` is `true` while a row has a pending optimistic change, which disables its buttons, so you don't start a new change on a message whose last change isn't confirmed yet. Edit and Delete show only on the user's own messages, and the routes still enforce authorship. Each call returns a TanStack DB transaction, and its `when("settled")` promise rejects with the handler's error if the write fails and the change rolls back, so you can await it to show an error.

An optimistic insert needs a whole `Message`, so the component also takes the signed-in user's id. It only fills in the optimistic row, because the server takes the author from the verified token. Read it from the [Managed Better Auth](https://neon.com/docs/auth/overview.md) session where the app creates the collection, as `frontend/main.tsx` does in [render the app](#render-the-app).

The [TanStack example](https://neon.com/docs/realtime/examples/tanstack.md) puts these handlers into a full app with server-side rendering, and the [Next.js example](https://neon.com/docs/realtime/examples/nextjs.md) shows the React binding in a server-rendered app.

## Render the app

`frontend/main.tsx` from the [reactivity guide](reactivity.md#query-the-collection) renders both views inside `RealtimeProvider`. Read the signed-in user's id from the session and pass it to `Channel` and `Inbox`:

```tsx filename="frontend/main.tsx"
import { StrictMode } from "react"
import { createRoot } from "react-dom/client"
import { RealtimeProvider } from "@neon/realtime-react"

import { fetchSealedInbox } from "./api"
import { authClient } from "./auth"
import { Channel } from "./Channel"
import { client } from "./client"
import { createMessagesCollection } from "./collections"
import { Inbox } from "./Inbox"

const root = createRoot(document.getElementById("root")!)
const { data: session } = await authClient.getSession()

if (!session) {
  root.render(<p>Sign in to see the messages.</p>)
} else {
  const userId = session.user.id
  const messagesCollection = createMessagesCollection(await fetchSealedInbox())

  root.render(
    <StrictMode>
      <RealtimeProvider client={client}>
        <Channel channelId={2} userId={userId} />
        <Inbox messagesCollection={messagesCollection} channelId={2} userId={userId} />
      </RealtimeProvider>
    </StrictMode>
  )
}
```

Signed out, it renders a message instead of the views. Sign-in isn't part of these guides: the [running example](https://github.com/neondatabase/examples/tree/main/realtime/guides-chat-app) shows one way to add it.

<!-- Generated by scripts/sync-skill-docs.mjs from website@thruflo/realtime-nextjs-drizzle-kysely:78f8f132f (docs/realtime/guides/queries). Do not edit: change the docs and run npm run sync:docs. sha256:c4d6f1dc7acdcb310b0fc39744dd73fccf51b4aaa88abf7d8684b925d06535c1 -->
Canonical page: https://neon.com/docs/realtime/guides/queries.md

> Summary: Define live queries in your backend with raw SQL or Drizzle, fill in their parameters from the authenticated request, and seal them so clients can subscribe without being able to read or change them.

# Defining queries

Define live SQL queries that you can subscribe to.

## Contents

- [Setup](#setup)
- [Defining live queries](#defining-live-queries)
  - [Table requirements](#table-requirements)
  - [Using raw SQL](#using-raw-sql)
  - [Using a query builder or ORM](#using-a-query-builder-or-orm)
- [Sealing queries](#sealing-queries)
  - [Renewing sealed queries](#renewing-sealed-queries)
- [Authorization](#authorization)
  - [Sealing authorized queries](#sealing-authorized-queries)
  - [Live authorization](#live-authorization)
  - [Why not row-level security](#why-not-row-level-security)
- [Exposing queries](#exposing-queries)
  - [Exposing an endpoint](#exposing-an-endpoint)
  - [Using server functions](#using-server-functions)

Realtime works by defining and [subscribing](subscribing.md) to live SQL queries. Queries are designed to be [defined securely](#defining-live-queries) in your backend code and [subscribed‑to safely](#exposing-queries) in your frontend.

Define live queries using [raw SQL](#using-raw-sql) or a [query builder or ORM](#using-a-query-builder-or-orm). [Seal the queries](#sealing-queries) and [expose](#exposing-queries) them to your frontend via an [API endpoint](#exposing-an-endpoint) or a [server function](#using-server-functions), using your normal [authorization](#authorization) context.

## Setup

The guides all share one running example: a chat app where users read the messages in the channels they belong to. You can see the [source code in the examples repo](https://github.com/neondatabase/examples/tree/main/realtime/guides-chat-app) or expand the steps below.

<details>

<summary>Setup steps</summary>

Install [Drizzle](https://orm.drizzle.team) with the [node-postgres](https://node-postgres.com) driver (`pg`), [Hono](https://hono.dev), `ws`, and `jose`, which validates [Better Auth](https://neon.com/docs/auth/overview.md) tokens:

```bash filename="Terminal"
npm install drizzle-orm pg hono jose ws
npm install -D @types/pg @types/ws
```

Install the Realtime [TypeScript SDK](https://neon.com/docs/realtime/sdks/typescript.md) and its [Drizzle adapter](https://neon.com/docs/realtime/sdks/typescript.md#drizzle):

```bash filename="Terminal"
npm install @neon/realtime @neon/realtime-drizzle
```

Install the [Managed Better Auth](https://neon.com/docs/auth/overview.md) client for the frontend:

```bash filename="Terminal"
npm install @neondatabase/auth
```

Add a `neon.ts` file that enables Managed Better Auth and [Realtime](https://neon.com/docs/realtime/quickstart.md#enable-realtime) on your branch:

```ts filename="neon.ts"
import { defineConfig } from "@neon/config/v1"

export default defineConfig({
  auth: true,
  preview: { realtime: { allowedOrigins: ["http://localhost:5173"] } }
})
```

List every origin your browser app runs on in `allowedOrigins`, for example `http://localhost:3000` for the SSR frameworks.

Install the config package, link the directory to your Neon project and deploy the config:

```bash filename="Terminal"
npm install -D @neon/config
neon link
neon deploy
```

`neon deploy` enables Managed Better Auth and Realtime, waits until they're ready, and writes the four server variables below to `.env.local`. If Realtime isn't ready before the deploy stops waiting, run `neon env pull` once it is.

These are the environment variables on the server:

- `DATABASE_URL`: your Lakebase Postgres [connection string](https://neon.com/docs/connect/connect-intro.md)
- `NEON_REALTIME_SECRET`: the secret from the [Realtime page](https://neon.com/docs/realtime/quickstart.md#enable-realtime) in the Neon Console
- `NEON_AUTH_BASE_URL`: your [Managed Better Auth](https://neon.com/docs/auth/overview.md) URL
- `NEON_REALTIME_URL`: the URL from the same Realtime page

And these in the frontend, with the same values as `NEON_AUTH_BASE_URL` and `NEON_REALTIME_URL`:

- `VITE_NEON_AUTH_URL`
- `VITE_NEON_REALTIME_URL`

Vite expands `$VAR` in env files, and `neon env pull` keeps lines it doesn't manage, so point them at the server variables:

```bash filename="Terminal"
echo 'VITE_NEON_AUTH_URL=$NEON_AUTH_BASE_URL' >> .env.local
echo 'VITE_NEON_REALTIME_URL=$NEON_REALTIME_URL' >> .env.local
```

Define the example schema and apply your usual [Drizzle migrations](https://neon.com/docs/guides/drizzle-migrations.md):

```ts filename="backend/schema.ts"
import { integer, pgTable, primaryKey, text, timestamp, uuid } from "drizzle-orm/pg-core"

export const messages = pgTable(
  "messages", {
    id: uuid("id").primaryKey(),
    channel_id: integer("channel_id").notNull(),
    author_id: text("author_id").notNull(),
    body: text("body").notNull(),
    created_at: timestamp("created_at", { withTimezone: true }).notNull().defaultNow()
  }
)

export const channelMembers = pgTable(
  "channel_members", {
    channel_id: integer("channel_id").notNull(),
    user_id: text("user_id").notNull()
  },
  (table) => [primaryKey({ columns: [table.channel_id, table.user_id] })]
)

export type Message = typeof messages.$inferSelect
```

Configure your Drizzle client (`db`):

```ts filename="backend/db.ts"
import { drizzle } from "drizzle-orm/node-postgres"
import { Pool } from "pg"
import * as schema from "./schema"

const pool = new Pool({
  // Neon URLs say sslmode=require. Asking pg for verify-full keeps today's checks and skips its warning
  connectionString: process.env.DATABASE_URL!.replace("sslmode=require", "sslmode=verify-full")
})
export const db = drizzle(pool, { schema })

// So an idle connection error, for example when the compute scales to zero, doesn't crash the API
pool.on("error", (error: Error) => console.error(error))
```

Great, that's the main setup to follow along with the guides. Or for the full boilerplate, see the [example source code](https://github.com/neondatabase/examples/tree/main/realtime/guides-chat-app) on GitHub.

</details>

## Defining live queries

Neon Realtime supports any **deterministic SQL query** that is [supported by Postgres](https://neon.com/postgresql/tutorial/select).

This includes [where clauses](https://neon.com/postgresql/tutorial/where), [order by](https://neon.com/postgresql/tutorial/order-by) and [limit](https://neon.com/postgresql/tutorial/limit), [joins](https://neon.com/postgresql/tutorial/joins), aggregates, [group by](https://neon.com/postgresql/tutorial/group-by) and distinct, functions (built-in and user-defined), extensions, window functions, [sub-queries](https://neon.com/postgresql/tutorial/subquery), etc.

The only limitations are that:

1. queries must be **read-only**, i.e.: must not have side effects
2. any functions used in your queries must be **deterministic**

So, for example, you can't `SELECT` over updated rows and you can't use functions like `now()` or `gen_random_uuid()`.

### Table requirements

Tables need a [stable replica identity](https://www.postgresql.org/docs/current/sql-altertable.html#SQL-ALTERTABLE-REPLICA-IDENTITY). If your tables have a primary key, they're fine.

**Optional: REPLICA IDENTITY FULL:**

Setting `REPLICA IDENTITY FULL` can make live queries marginally faster, at the cost of writing the whole old row to the Postgres WAL on every update and delete. That may save Realtime a lookup when it processes those changes. If latency matters for a write-heavy table, benchmark both settings with your own workload before you switch.

```sql
ALTER TABLE messages REPLICA IDENTITY FULL;
```

### Using raw SQL

Create a `RealtimeServer` instance (`secret` is the `NEON_REALTIME_SECRET` generated when [enabling Realtime](https://neon.com/docs/realtime/quickstart.md#enable-realtime) and `db` should be the `dbname` in your [Postgres connection string](https://neon.com/docs/connect/connect-from-any-app.md#get-a-connection-string-from-the-neon-console)):

```ts filename="backend/realtime-raw.ts"
import { createRealtime } from "@neon/realtime/server"

export const realtime = createRealtime({
  secret: process.env.NEON_REALTIME_SECRET!,
  db: "neondb"
})
```

`createRealtime()` also takes a [`debugMode`](subscribing.md#show-full-database-errors) option.

Then define queries using [`rawSql`](https://neon.com/docs/realtime/sdks/typescript.md#backend-rawsql). For example:

```ts filename="backend/queries-raw.ts"
import { rawSql } from "@neon/realtime/server"

import type { Message } from "./schema"

export const allMessages = rawSql<Message>(
  `SELECT author_id, body, channel_id, created_at, id FROM messages`
)
```

A raw SQL query has to be a single `SELECT` statement with no `;`. With at most 32 KiB of SQL and at most 256 parameters of up to 16 KiB each.

#### Query parameters

When your query has parameters, wrap it in a function so they can be provided when using it. Put placeholders (like `$1`, `$2`, etc.) in the SQL and pass the values in the array.

For example:

```ts filename="backend/queries-raw.ts"
import { rawSql } from "@neon/realtime/server"

import type { Message } from "./schema"

export function messagesQuery(userId: string, channelId: number) {
  return rawSql<Message>(
    `SELECT messages.author_id, messages.body, messages.channel_id,
       messages.created_at, messages.id FROM messages
     INNER JOIN channel_members ON channel_members.channel_id = messages.channel_id
     WHERE channel_members.user_id = $1 AND messages.channel_id = $2
     ORDER BY messages.created_at DESC
     LIMIT 50`,
    [userId, channelId]
  )
}
```

**Warning: SQL injection**

Never use literal interpolation for user input like `` `WHERE user_id = '${userId}'` ``. Always use placeholders like `` `WHERE user_id = $1` `` instead.

#### Parameter types

You can pass strings, numbers, booleans, bigints and `null` as they are. Other value types need a [`pgParam`](https://neon.com/docs/realtime/sdks/typescript.md#backend-pgparam) helper to specify the Postgres type. A `Date` could be a `date`, a `timestamp` or a `timestamptz`, so wrap it in `pgParam.date()`, `pgParam.timestamp()` or `pgParam.timestamptz()`.

For example, this filters messages `since` a [timestamp with timezone](https://neon.com/postgresql/tutorial/timestamp#introduction-to-postgresql-timestamp):

```ts filename="backend/recent-messages.ts" {5,10,13}
import { pgParam, rawSql } from "@neon/realtime/server"

import type { Message } from "./schema"

export function messagesSinceQuery(userId: string, channelId: number, since: Date) {
  return rawSql<Message>(
    `SELECT messages.author_id, messages.body, messages.channel_id,
       messages.created_at, messages.id FROM messages
     INNER JOIN channel_members ON channel_members.channel_id = messages.channel_id
     WHERE channel_members.user_id = $1 AND messages.channel_id = $2 AND messages.created_at > $3
     ORDER BY messages.created_at DESC
     LIMIT 50`,
    [userId, channelId, pgParam.timestamptz(since)]
  )
}
```

There are helpers for dates, JSON, `bytea` and arrays, such as `pgParam.array("text", ["a"])`. `pgParam.text()` covers any other named type, for example `pgParam.text("uuid", id)`.

### Using a query builder or ORM

Realtime supports an adapter pattern to convert any typed query object to raw SQL. This works by passing an adapter to [`createRealtime`](https://neon.com/docs/realtime/sdks/typescript.md#backend-createrealtime).

The rest of these guides use the Drizzle versions of `backend/realtime.ts` and `backend/queries.ts` below in place of the raw SQL files above.

For example, configure Realtime to use the [`drizzleAdapter`](https://neon.com/docs/realtime/sdks/typescript.md#drizzle-drizzleadapter):

```ts filename="backend/realtime.ts" {2,7}
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()
})
```

You can now define queries using Drizzle:

```ts filename="backend/queries.ts"
import { and, desc, eq, getTableColumns } from "drizzle-orm"

import { db } from "./db"
import { channelMembers, messages } from "./schema"

export function messagesQuery(userId: string, channelId: number) {
  return db
    .select(getTableColumns(messages))
    .from(messages)
    .innerJoin(channelMembers, eq(channelMembers.channel_id, messages.channel_id))
    .where(and(eq(channelMembers.user_id, userId), eq(messages.channel_id, channelId)))
    .orderBy(desc(messages.created_at))
    .limit(50)
}
```

See the [Drizzle](https://neon.com/docs/realtime/examples/drizzle.md) and [Kysely](https://neon.com/docs/realtime/examples/kysely.md) examples for more details.

**Tip: Column name casing**

The Drizzle adapter needs each selected key to match its column name. That's why the schema uses snake_case keys and the join selects `getTableColumns(messages)`.

The [Drizzle example](https://neon.com/docs/realtime/examples/drizzle.md) shows how to alias fields to match the column key.

## Sealing queries

Once you've defined a query, if you're using it in trusted backend code, you can just [subscribe to it right away](subscribing.md#subscribe-in-trusted-backend-code). However, most of the time, you'll want to subscribe to live queries in an untrusted environment, such as a frontend app or agent runtime.

This is where **sealing** comes in. Sealing signs and encrypts the exact query, its parameters and the target database into a short-lived capability (similar to a JWT), so it can be used safely in an untrusted environment.

**Warning: Keep your Realtime secret on the server**

Anyone with `NEON_REALTIME_SECRET` can seal any query. Treat it like your `DATABASE_URL`, and never ship it in a browser bundle. In a Vite app, that means never giving it a `VITE_` prefix. To replace it, run [`neon realtime rotate-secret`](https://neon.com/docs/cli/realtime.md#rotate-secret).

To seal a query you pass your query instance, e.g.:

```ts filename="backend/server.ts"
const query = messagesQuery(userId, channelId)
```

To the [`seal`](https://neon.com/docs/realtime/sdks/typescript.md#backend-realtimeserverseal) method of the `RealtimeServer` instance you created above:

```ts filename="backend/server.ts"
const sealedQuery = await realtime.seal({ query })
```

It's local [Web Crypto](https://developer.mozilla.org/en-US/docs/Web/API/Web_Crypto_API) work with no network call, so it's cheap to run on every request. It returns a [`SealedLiveQuery<Row>`](https://neon.com/docs/realtime/sdks/typescript.md#backend-sealedlivequery), which has three fields:

- `capability`: the encrypted query
- `expiresAt`: when it expires, in Unix milliseconds
- `queryFingerprint`: a stable identity for the exact query

The client can subscribe to a sealed query, but it can't read or change it. Your secret, your SQL and the parameter values bound to it are not readable anywhere in it. The [Live SQL API](https://neon.com/docs/realtime/how-it-works.md#live-sql-api) verifies the capability itself, so subscribing doesn't need a call back to your server.

**Important: Sealed queries are bearer tokens**

Sealed queries are typically user-specific and sealed per request, where they can bake in parameters like the authenticated user ID.

Anyone holding a sealed query can subscribe to it until it expires. Send it over HTTPS, and keep it out of URLs, logs and persistent browser storage.

### Renewing sealed queries

Sealed queries are short-lived (currently 60 seconds). To keep a subscription running, renew its sealed query before `expiresAt`, or the subscription goes `stale` when it expires. The [reactivity bindings](reactivity.md) renew for you when you pass them a `refreshQuery` function. The [subscribing guide](subscribing.md#renew-the-sealed-query) shows how to renew manually.

A renewal has to be for the **same** query. Different parameter values make a different query, with a different fingerprint. So, for example, to switch channels in the chat app, you would seal the new query and subscribe again.

## Authorization

Once you've sealed a query, you'll want to get it to your frontend to subscribe to.

You do this by [exposing the query](#exposing-queries) from a **request handler** in your backend. So that's where you authenticate the user and decide what they can see.

### Sealing authorized queries

The flow is as follows:

1. read the auth context from the request
2. authorize access using arbitrary business logic (e.g.: your auth middleware)
3. construct the query, passing in any relevant parameters from the auth context (e.g.: filtering or joining on the authenticated `userId`)
4. seal the query so these can't be changed

This allows you to authorize live query subscriptions using existing web auth logic, middleware and frameworks, such as [Neon Auth](https://neon.com/docs/auth/overview.md).

### Live authorization

In our running example, [`messagesQuery()`](#using-a-query-builder-or-orm) joins through `channel_members` and filters on the `userId`, so it only returns messages from channels the user belongs to:

```ts
.where(and(eq(channelMembers.user_id, userId), eq(messages.channel_id, channelId)))
```

The join is **part of the live query**, so it stays live. If you remove someone from a channel, they'll immediately lose access to its messages.

That's why Realtime authorizes the **query**, not the connection. Your request handler checks who's asking and seals a query that only matches the rows that the user is allowed to see. The authorization is **defined** at request time but it **dynamically adapts** to your data in realtime.

**Warning: Never trust user input from the client**

Take authorization context from a verified session or token and validate everything else (like `channelId`) that the client sends in the request data.

### Why not row-level security

Postgres [row-level security](https://neon.com/docs/guides/row-level-security.md) (RLS) is often slow and becomes a bottleneck. It's also awkward to develop with, because your access rules live in migrations, rather than your application code.

It's simpler and more efficient to reuse the auth logic you already have in your backend.

Many existing apps do use RLS, though. So, while it isn't our recommended approach to auth, we have designed Neon Realtime's architecture to be able to support it and we aim to implement support before GA.

## Exposing queries

There are two main ways to expose a [sealed](#sealing-queries), [authorized](#authorization) query to your frontend:

1. [exposing an endpoint](#exposing-an-endpoint) from your backend API
2. [using server functions](#using-server-functions) (often with [SSR](ssr.md))

These are both **request handlers**. They run in a trusted environment in your backend with [auth context available](#sealing-authorized-queries) from the request.

With Vite and Hono, the browser's auth client talks to Managed Better Auth directly, so its session cookie never reaches your API. The browser sends the session's JWT as a bearer token, and the API verifies it. Next.js and TanStack Start mount Managed Better Auth on your app's own origin, so server code reads the session from its cookie and the browser holds no token. Either way, the Realtime connection carries no user credential: the sealed query is the authorization.

### Exposing an endpoint

Expose the `sealedQuery` from an API endpoint, using any web framework.

Our running example uses [Hono](https://hono.dev) and [Managed Better Auth](https://neon.com/docs/auth/overview.md). The client [sends its auth token](subscribing.md#fetch-from-an-api-endpoint) in an `Authorization: Bearer` header. The JWT is verified against your [Managed Better Auth JWKS](https://neon.com/docs/auth/guides/plugins/jwt.md#verify-a-token) in a `requireUser` helper:

```ts filename="backend/auth.ts"
import type { Context } from "hono"
import { HTTPException } from "hono/http-exception"
import { createRemoteJWKSet, jwtVerify } from "jose"

const authUrl = process.env.NEON_AUTH_BASE_URL!
const origin = new URL(authUrl).origin
const jwks = createRemoteJWKSet(new URL(`${authUrl}/.well-known/jwks.json`))

async function verifyUserId(token: string): Promise<string | undefined> {
  try {
    const { payload } = await jwtVerify(token, jwks, { issuer: origin, audience: origin })

    return payload.sub
  } catch {
    return undefined
  }
}

export async function getUser(headers: Headers): Promise<{ id: string } | undefined> {
  const header = headers.get("Authorization")
  const token = header?.startsWith("Bearer ") ? header.slice("Bearer ".length) : undefined
  const userId = token ? await verifyUserId(token) : undefined

  return userId ? { id: userId } : undefined
}

export async function requireUser(c: Context): Promise<{ id: string }> {
  const user = await getUser(c.req.raw.headers)

  if (!user) {
    throw new HTTPException(401, { message: "Unauthorized" })
  }

  return user
}
```

We then use this auth helper in a route to securely build and expose the `sealedQuery`:

```ts filename="backend/server.ts" {20,23}
import { Hono } from "hono"
import { validator } from "hono/validator"

import { requireUser } from "./auth"
import { messagesQuery } from "./queries"
import { realtime } from "./realtime"

const app = new Hono().post(
  "/api/v1/messages/live",
  validator("json", (value, c) => {
    const channelId: unknown = value?.channelId

    if (typeof channelId !== "number" || !Number.isInteger(channelId)) {
      return c.json({ error: "channelId must be an integer" }, 400)
    }

    return { channelId }
  }),
  async (c) => {
    const user = await requireUser(c)
    const { channelId } = c.req.valid("json")

    const query = messagesQuery(user.id, channelId)
    const sealedQuery = await realtime.seal({ query })

    return c.json(sealedQuery)
  }
)

export type AppType = typeof app

export default app
```

The `validator()` middleware checks the JSON body before the handler runs and types it, so the frontend's [`hc` client](subscribing.md#fetch-from-an-api-endpoint) knows this endpoint takes `{ channelId: number }`. The handler takes the user's ID from `requireUser()` and the `channelId` from the validated body, builds the query, seals it and returns it as JSON. Chain every route onto `new Hono()` in one expression, because `AppType` only carries the routes chained into it.

`seal()` doesn't depend on request or response types and a sealed query is plain JSON. The [subscribing guide](subscribing.md#fetch-from-an-api-endpoint) shows how to fetch it from the client and subscribe to it.

### Using server functions

If you're using a framework that supports [server functions](https://react.dev/reference/rsc/server-functions), you can return the `sealedQuery` from one. For example, with [TanStack Start](https://neon.com/docs/realtime/examples/tanstack.md) you can [create a server function](https://tanstack.com/start/latest/docs/framework/react/guide/server-functions).

The server function reads the session from the Managed Better Auth cookie with `requireUser`, a server-only middleware. The [SSR guide](ssr.md#send-both-to-the-browser) sets it up. The server function then seals the query for `context.user.id`:

```ts filename="src/server-functions.ts"
import { createServerFn } from "@tanstack/react-start"

import { messagesQuery } from "./backend/queries"
import { realtime } from "./backend/realtime"
import { requireUser } from "./session-middleware"

export const sealMessages = createServerFn({ method: "POST" })
  .middleware([requireUser])
  .validator((data: { channelId: number }) => {
    if (!Number.isInteger(data.channelId)) {
      throw new Error("channelId must be an integer")
    }

    return data
  })
  .handler(({ data, context }) =>
    realtime.seal({ query: messagesQuery(context.user.id, data.channelId) })
  )
```

The browser calls `sealMessages` to get the sealed query, and calls it again to [renew it](#renewing-sealed-queries).

To seal the query while the page renders on the server, see the [Server-side rendering](ssr.md) guide and the [Next.js](https://neon.com/docs/realtime/examples/nextjs.md) and [TanStack](https://neon.com/docs/realtime/examples/tanstack.md) examples.

<!-- Generated by scripts/sync-skill-docs.mjs from website@thruflo/realtime-3-integration:f26c1f629 (docs/realtime/guides/reactivity). Do not edit: change the docs and run npm run sync:docs. sha256:f6b3f6d6112f189444e30239f81d46c9e1431a61c2242286807d01190137ab87 -->
Canonical page: https://neon.com/docs/realtime/guides/reactivity.md

> Summary: Bind live queries to React component state with useLiveQuery, change their parameters and handle errors, or sync a broader live query into a TanStack DB collection and derive instant, local queries from it.

# Client-side reactivity

Bind live queries to component and agent state.

## Contents

- [React](#react)
  - [RealtimeProvider](#realtimeprovider)
  - [useLiveQuery](#uselivequery)
  - [Changing query parameters](#changing-query-parameters)
  - [Handling errors](#handling-errors)
- [TanStack DB](#tanstack-db)
  - [Seal a broader query](#seal-a-broader-query)
  - [Create the collection](#create-the-collection)
  - [Query the collection](#query-the-collection)
  - [Manage optimistic state](#manage-optimistic-state)
- [Other frameworks](#other-frameworks)

Build reactive apps and agents using higher-level reactivity primitives that manage live query [subscriptions](subscribing.md) for you.

**Important: Framework support**

Realtime [currently](#other-frameworks) has integrations for [React](#react) and [TanStack DB](#tanstack-db).

The [React binding](#react) keeps a component's state in step with a live query, and the [TanStack DB binding](#tanstack-db) syncs live queries into a [reactive client-side store](https://tanstack.com/db).

This page continues the [running example](https://github.com/neondatabase/examples/tree/main/realtime/guides-chat-app) from the [queries](queries.md) and [subscribing](subscribing.md) guides. It uses the `client` from `frontend/client.ts` and `fetchSealedMessages()` from `frontend/api.ts`.

## React

The `@neon/realtime-react` binding exports a [`RealtimeProvider`](https://neon.com/docs/realtime/sdks/typescript.md#react-realtimeprovider), a [`useLiveQuery()` hook](https://neon.com/docs/realtime/sdks/typescript.md#react-uselivequery) and a [`useRealtimeClient()` hook](https://neon.com/docs/realtime/sdks/typescript.md#react-userealtimeclient) that returns the provider's client. It works with React 18 and 19.

Install it alongside the client SDK:

```bash filename="Terminal"
npm install @neon/realtime-react
```

### RealtimeProvider

Wrap your app in `<RealtimeProvider>`. This allows all of your `useLiveQuery()` subscriptions to use the same [`RealtimeClient`](https://neon.com/docs/realtime/sdks/typescript.md#client-realtimeclient) and WebSocket connection.

```tsx filename="frontend/main.tsx"
import { StrictMode } from "react"
import { createRoot } from "react-dom/client"
import { RealtimeProvider } from "@neon/realtime-react"

import { client } from "./client"

createRoot(document.getElementById("root")!).render(
  <StrictMode>
    <RealtimeProvider client={client}>
      {/* Your app here */}
    </RealtimeProvider>
  </StrictMode>
)
```

`client` is the module singleton from `frontend/client.ts`, created once for the whole app. If you render on the server, create the client in the browser instead, as the [server-side rendering guide](ssr.md#open-the-connection-in-the-browser) explains.

### useLiveQuery

[`useLiveQuery()`](https://neon.com/docs/realtime/sdks/typescript.md#react-uselivequery) takes a `sealedQuery` and a `refreshQuery` (and, optionally, [`initialData` from SSR](ssr.md)) and returns the query's current rows and state.

For example, here we define a `Messages` component with `useLiveQuery`:

```tsx filename="frontend/Messages.tsx"
import { useEffect } from "react"
import { useLiveQuery } from "@neon/realtime-react"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { LocalTime } from "./LocalTime"

interface Props {
  sealedQuery: SealedLiveQuery<Message>;
  refreshQuery: () => Promise<SealedLiveQuery<Message>>;
  initialData?: readonly Message[];
}

export function Messages({ sealedQuery, refreshQuery, initialData }: Props) {
  const { data, status, error, utils } = useLiveQuery(sealedQuery, { refreshQuery, initialData })

  useEffect(() => {
    return utils.onUpdate((update) => {
      if (update.type === "changes") {
        console.debug(`Applied ${update.changes.length} changes from transactions ${update.batch.txids.join(", ")}`)
      }
    })
  }, [utils])

  if (status === "error") {
    return <p role="alert">Live updates stopped: {error.message}</p>
  }

  if (data === undefined) {
    return <p>Loading messages</p>
  }

  return (
    <section>
      {status === "stale" && <p role="status">Updating</p>}
      <ul>
        {data.map((message) => (
          <li key={message.id}>
            <LocalTime date={message.created_at} />
            <p>{message.body}</p>
          </li>
        ))}
      </ul>
    </section>
  )
}
```

Which uses this `LocalTime` component to format the time [after hydration](ssr.md#keep-row-shapes-identical):

```tsx filename="frontend/LocalTime.tsx"
import { useEffect, useState } from "react"

export function LocalTime({ date }: { date: Date }) {
  const [formatted, setFormatted] = useState<string>()
  useEffect(() => setFormatted(date.toLocaleTimeString()), [date])

  return <time dateTime={date.toISOString()}>{formatted}</time>
}
```

`useLiveQuery` subscribes when the component mounts, re-renders it whenever the results change and unsubscribes when it unmounts.

In this example, the sealed query is a `SealedLiveQuery<Message>`, so `data` is a `readonly Message[]`, with `created_at` decoded to a `Date`. It's `undefined` until the first result arrives. `status` is the subscription's [state](subscribing.md#listen-for-changes): the component shows a loading line until the first result and a small indicator while it's `stale`.

#### refreshQuery

`refreshQuery` is how the `useLiveQuery` hook [renews the sealed query](subscribing.md#renew-the-sealed-query).

Sealed queries are [short-lived](queries.md#renewing-sealed-queries), so the hook calls it 10 seconds before `expiresAt`. If a call fails, the hook retries every second until one succeeds. The callback has to return a sealed query for the same exact query, so `Channel` builds it from the same `channelId` the sealed query was sealed with, as [changing query parameters](#changing-query-parameters) shows. Changing only the callback doesn't restart the subscription, and `useCallback` keeps it stable across renders. Without `refreshQuery`, nothing renews the sealed query, and the rows go `stale` when it expires.

#### initialData

The hook also accepts an optional `initialData` parameter, for example from a server render. See the [server-side rendering guide](ssr.md#render-the-initial-rows-straight-away) for more details.

#### utils

The `utils` in the return value give you access to the hook's underlying [materialized subscription](https://neon.com/docs/realtime/sdks/typescript.md#react-uselivequeryutils).

This provides `getSnapshot()`, `getState()`, `awaitRows()`, the `onChange`, `onUpdate` and `onStateChange` listeners, `renew()` and `awaitTxId()`. It's stable for the life of the subscription, so register listeners in an effect and return the remover, as the `onUpdate` listener does.

Note that a `changes` update has already been applied by the time its listener runs, and it [carries the txids](https://neon.com/docs/realtime/sdks/typescript.md#client-livequerybatchinfo) of the transactions in it, which the [mutations guide](mutations.md#how-optimistic-mutations-work) uses to confirm optimistic writes.

`utils` omits `unsubscribe()`, because React owns cleanup. The hook unsubscribes when the component unmounts or its sealed query changes. `getState()` and the listeners report the state of the underlying subscription, so they can differ from the hook's `status`.

### Changing query parameters

The hook takes a sealed query, not parameters. Parameters are sealed into the query on the server, where the client can't change them. To show a different channel, fetch a new sealed query for it and pass that in. A new `sealedQuery` is a new logical subscription: the hook cleans up the old one and subscribes to the new one.

Fetch the sealed query in the component that owns the parameter. `useSealedQuery()` fetches one whenever `channelId` changes, and `Channel` renders `Messages` once it has one:

```tsx filename="frontend/Channel.tsx"
import { useCallback, useEffect, useState } from "react"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { fetchSealedMessages } from "./api"
import { Messages } from "./Messages"

interface SealedChannel {
  channelId: number;
  sealedQuery?: SealedLiveQuery<Message>;
  error?: unknown;
}

function useSealedQuery(channelId: number): SealedChannel | undefined {
  const [sealed, setSealed] = useState<SealedChannel>()

  useEffect(() => {
    let cancelled = false

    fetchSealedMessages(channelId).then(
      (sealedQuery) => {
        if (!cancelled) setSealed({ channelId, sealedQuery })
      },
      (error: unknown) => {
        if (!cancelled) setSealed({ channelId, error })
      }
    )

    return () => {
      cancelled = true
    }
  }, [channelId])

  // Never hand the previous channel's sealed query to the new channel
  return sealed?.channelId === channelId ? sealed : undefined
}

export function Channel({ channelId }: { channelId: number }) {
  const sealed = useSealedQuery(channelId)
  // Renewals must be for the same query, so refresh with the same channel
  const refreshQuery = useCallback(() => fetchSealedMessages(channelId), [channelId])

  if (sealed?.sealedQuery) {
    return <Messages sealedQuery={sealed.sealedQuery} refreshQuery={refreshQuery} />
  }

  if (sealed?.error) {
    return <p role="alert">Couldn't open this channel</p>
  }

  return <p>Loading messages</p>
}
```

When the channel changes, `sealed` still holds the previous channel's sealed query until the new one arrives. Without the `channelId` check, `Messages` would get the old channel's sealed query with a `refreshQuery` for the new channel. That `refreshQuery` would then fetch the new channel's query to renew the old subscription, and the renewal would be rejected with status `error`. The `cancelled` flag covers a second race. It drops a response that arrives after the channel has changed, or after Strict Mode's extra effect run in development.

### Handling errors

`useLiveQuery()` doesn't throw for subscription errors, and it doesn't suspend. It reports everything through `status`. Render loading and error states from it, as `Messages` does. The hook throws only when it's used outside a `<RealtimeProvider>`, when the sealed query is malformed or when the client is closed. Those errors reach the nearest error boundary, not `status`.

A failed renewal doesn't end in `error`. If your endpoint can't return a sealed query, for example because the user has signed out and it answers 401, the hook keeps retrying, and once the sealed query expires the status is `stale` until a renewal succeeds. So drive sign-out from your auth state: when the user signs out, unmount the components that subscribe. The status becomes `error` only when a renewal returns a malformed response or a sealed query for a different query. To recover, fetch a new sealed query and pass it in.

As with a parameter change, the new sealed query restarts the subscription. The [subscribing guide](subscribing.md#handle-errors) covers recovery at the subscription level.

## TanStack DB

Each `useLiveQuery()` above subscribes to one sealed query, so each view of the data needs its own sealed query and its own round trip to your server.

With [TanStack DB](https://tanstack.com/db/latest/docs/overview), you can oversync a broader live query into a client-side collection, such as every message in every channel the user belongs to, and then derive as many local live queries from it as you need: [filters, sorts and joins across collections](https://tanstack.com/db/latest/docs/guides/live-queries).

Install the TanStack binding and TanStack DB's React package:

```bash filename="Terminal"
npm install @neon/realtime-tanstack @tanstack/react-db
```

### Seal a broader query

Add `inboxQuery()` to `backend/queries.ts`. It's `messagesQuery()`, with the same membership join but no channel filter. So it returns the newest 500 messages across every channel that the user belongs to:

```ts filename="backend/queries.ts"
export function inboxQuery(userId: string) {
  return db
    .select(getTableColumns(messages))
    .from(messages)
    .innerJoin(channelMembers, eq(channelMembers.channel_id, messages.channel_id))
    .where(eq(channelMembers.user_id, userId))
    .orderBy(desc(messages.created_at))
    .limit(500)
}
```

Then expose it. In `backend/server.ts`, import `inboxQuery` alongside `messagesQuery` and chain one more route onto the app. It takes no body, because its only parameter comes from the verified token:

```ts filename="backend/server.ts"
  .post("/api/v1/inbox/live", async (c) => {
    const user = await requireUser(c)

    const sealedQuery = await realtime.seal({ query: inboxQuery(user.id) })

    return c.json(sealedQuery)
  })
```

### Create the collection

On the frontend, add `fetchSealedInbox()` to `frontend/api.ts`, next to `fetchSealedMessages()`. As before, the `as` restores the row type:

```ts filename="frontend/api.ts"
export async function fetchSealedInbox(): Promise<SealedLiveQuery<Message>> {
  const res = await server.api.v1.inbox.live.$post()

  if (!res.ok) {
    throw new Error(`Could not fetch the inbox query (${res.status})`)
  }

  return (await res.json()) as SealedLiveQuery<Message>
}
```

Then create the collection. `realtimeCollectionOptions()` takes the standard TanStack DB collection options plus the Realtime client and a sealed query, and returns options for `createCollection()`:

```ts filename="frontend/collections.ts"
import { createCollection } from "@tanstack/react-db"
import { realtimeCollectionOptions } from "@neon/realtime-tanstack"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { fetchSealedInbox } from "./api"
import { client } from "./client"

export function createMessagesCollection(sealedQuery: SealedLiveQuery<Message>) {
  return createCollection(
    realtimeCollectionOptions({
      id: "messages",
      client,
      query: sealedQuery,
      refreshQuery: fetchSealedInbox,
      getKey: (message) => message.id
    })
  )
}

export type MessagesCollection = ReturnType<typeof createMessagesCollection>
```

`getKey` returns each message's primary key. TanStack DB keys rows by it, so use a stable, unique column, never a subscription's `rowId`. `refreshQuery` renews the sealed query the same way [the React hook does](#refreshquery). While renewals fail, the collection stays `ready` with its last rows. See the [`realtimeCollectionOptions()` reference](https://neon.com/docs/realtime/sdks/typescript.md#tanstack-db-realtimecollectionoptions) for the rest of the options.

The collection starts syncing as soon as a live query using it is mounted. It's in the `loading` state until the first result arrives, and `ready` from then on, with each committed transaction applied atomically. A `ready` collection stays ready while the subscription is `connecting` or `stale`, so there's no separate recovery flag to render. If the subscription ends in `error`, so does the collection.

### Query the collection

Create the collection once, when the app starts. The collection takes the client directly, but `Channel` still calls `useLiveQuery()`, so keep both views inside `RealtimeProvider`:

```tsx filename="frontend/main.tsx"
import { StrictMode } from "react"
import { createRoot } from "react-dom/client"
import { RealtimeProvider } from "@neon/realtime-react"

import { fetchSealedInbox } from "./api"
import { authClient } from "./auth"
import { Channel } from "./Channel"
import { client } from "./client"
import { createMessagesCollection } from "./collections"
import { Inbox } from "./Inbox"

const root = createRoot(document.getElementById("root")!)
const { data: session } = await authClient.getSession()

if (!session) {
  root.render(<p>Sign in to see the messages.</p>)
} else {
  const messagesCollection = createMessagesCollection(await fetchSealedInbox())

  root.render(
    <StrictMode>
      <RealtimeProvider client={client}>
        <Channel channelId={2} />
        <Inbox messagesCollection={messagesCollection} channelId={2} />
      </RealtimeProvider>
    </StrictMode>
  )
}
```

Sign-in isn't part of these guides: build the page with Neon's [Managed Better Auth UI components](https://neon.com/docs/auth/reference/ui-components.md), as the [running example](https://github.com/neondatabase/examples/tree/main/realtime/guides-chat-app) does by rendering `AuthView` on `/auth/*` before it checks the session.

Then read from the collection with TanStack's own `useLiveQuery()`. It shares a name with the Realtime hook, so import it under an alias that won't clash in an app that uses both. This view shows one channel, newest first:

```tsx filename="frontend/Inbox.tsx"
import { eq, useLiveQuery as useCollectionQuery } from "@tanstack/react-db"

import type { MessagesCollection } from "./collections"

interface Props {
  messagesCollection: MessagesCollection;
  channelId: number;
}

export function Inbox({ messagesCollection, channelId }: Props) {
  const { data, isReady, isError } = useCollectionQuery({
    query: (q) =>
      q
        .from({ message: messagesCollection })
        .where(({ message }) => eq(message.channel_id, channelId))
        .orderBy(({ message }) => message.created_at, "desc")
  })

  if (isError) {
    return <p role="alert">Couldn't sync messages</p>
  }

  if (!isReady) {
    return <p>Loading messages</p>
  }

  return (
    <ul>
      {data.map((message) => (
        <li key={message.id}>{message.body}</li>
      ))}
    </ul>
  )
}
```

Pass a different `channelId` and the query re-runs in the browser, against the rows the collection already holds. Local queries only see what the collection holds, so a channel whose messages fall outside the newest 500 shows fewer rows, or none. There's no new sealed query and no round trip to your server, unlike [changing the parameters](#changing-query-parameters) of a `useLiveQuery()` hook. Every view you add, like an unread count or a search across channels, is another local query over the same synced rows.

For server rendering, the [server-side rendering guide](ssr.md#with-tanstack-db) seeds a collection on the server and hydrates it in the browser, and the [TanStack example](https://neon.com/docs/realtime/examples/tanstack.md) builds a full app with TanStack Start.

### Manage optimistic state

Once you have a TanStack DB collection defined, you can also use it for local optimistic writes. It's often easier in a complex app to normalize data loading into a collection which can have optimistic state applied to it once, than to have to remember to apply optimistic state to a range of different live query subscriptions.

For writes through the collection, with optimistic state, see the [mutations guide](mutations.md#with-tanstack-db-collections).

## Other frameworks

Bindings for other frameworks are coming. Until then, you can use the current implementations as a reference to write your own.

The React binding is a thin layer over the client subscription: `useSyncExternalStore` plus the [`QueryRefreshController`](https://neon.com/docs/realtime/sdks/typescript.md#client-queryrefreshcontroller). Point a coding agent at [its source](https://github.com/neondatabase/neon-pkgs/tree/main/packages/realtime-react) and it can reproduce the binding for other frameworks, such as Solid, Svelte or Vue.

You can also see the [source code for the TanStack DB collection here](https://github.com/neondatabase/neon-pkgs/blob/main/packages/realtime-tanstack/src/collection-options.ts).

<!-- Generated by scripts/sync-skill-docs.mjs from website@thruflo/realtime-2-docs:24a2fd708 (docs/realtime/guides/ssr). Do not edit: change the docs and run npm run sync:docs. sha256:55d466b2446614f5b3ce667589f4dfe9606324c21f578c044ec6fa722f6696ba -->
Canonical page: https://neon.com/docs/realtime/guides/ssr.md

> Summary: Server-render pages that use live queries. Read and seal the same query for each request, send both to the browser, render the initial rows straight away and let the first result bring the page live.

# Server-side rendering

Render sealed queries and initial query results on the server.

## Contents

- [Return the sealed query and initial results](#return-the-sealed-query-and-initial-results)
- [Send both to the browser](#send-both-to-the-browser)
- [Render the initial rows straight away](#render-the-initial-rows-straight-away)
- [Open the connection in the browser](#open-the-connection-in-the-browser)
- [Keep row shapes identical](#keep-row-shapes-identical)
- [Renew from the server](#renew-from-the-server)
- [When the session changes](#when-the-session-changes)
- [With TanStack DB](#with-tanstack-db)
  - [Create a `DbClient` per router](#create-a-dbclient-per-router)
  - [Define the collection](#define-the-collection)
  - [Seed the collection in a loader](#seed-the-collection-in-a-loader)
- [Keep secrets on the server](#keep-secrets-on-the-server)
- [Examples](#examples)

[Server-side rendering (SSR)](https://developer.mozilla.org/en-US/docs/Glossary/SSR) lets a page arrive with its rows already in the HTML, then subscribe to ongoing changes as soon as it hydrates.

Realtime supports SSR, including with [Next.js](https://neon.com/docs/realtime/examples/nextjs.md) and [TanStack Start](https://neon.com/docs/realtime/examples/tanstack.md).

## Return the sealed query and initial results

The pattern is to [build the live query](queries.md#defining-live-queries) once and then use it twice in the same request:

1. run the query to get the initial results
2. [seal the query](queries.md#sealing-queries) so it can be subscribed to safely on the frontend

Running and sealing the query are both `async`, so do them in parallel using `Promise.all`:

```ts filename="backend/messages-page.ts"
import type { SealedLiveQuery } from "@neon/realtime/server"

import { messagesQuery } from "./queries"
import { realtime } from "./realtime"
import type { Message } from "./schema"

export interface MessagesPageData {
  initialData: Message[];
  sealedQuery: SealedLiveQuery<Message>;
}

// Call this per request. The result belongs to one user and is short-lived.
export async function loadMessagesPage(userId: string, channelId: number): Promise<MessagesPageData> {
  const query = messagesQuery(userId, channelId)

  const [initialData, sealedQuery] = await Promise.all([query, realtime.seal({ query })])

  return { initialData, sealedQuery }
}
```

`messagesQuery()` and `realtime` are the ones from [defining queries](queries.md#using-a-query-builder-or-orm). Your Server Component or server function calls `loadMessagesPage()` with the user's id from the [verified auth session](queries.md#authorization).

## Send both to the browser

Both values are serializable, so they go into the hydration payload with the rest of the page's props or loader data. The framework serializes them into the HTML it sends, and the browser picks them up when it hydrates.

For example with [Next.js](https://neon.com/docs/realtime/examples/nextjs.md), call `loadMessagesPage()` from a Server Component and pass the results to a Client Component as props. `auth` is the Managed Better Auth server instance in `lib/auth/server.ts`, set up as in the [Managed Better Auth Next.js quick start](https://neon.com/docs/auth/quick-start/nextjs-api-only.md). The page must render per request, because `auth.getSession()` [reads cookies](https://neon.com/docs/auth/troubleshooting.md#missing-force-dynamic-on-server-components), so it sets `force-dynamic`. If your app enables `cacheComponents`, remove the `dynamic` export and read the session inside a `<Suspense>` boundary instead.

```tsx filename="app/channels/[channelId]/page.tsx"
import { notFound, redirect } from "next/navigation"

import { loadMessagesPage } from "../../../backend/messages-page"
import { auth } from "../../../lib/auth/server"
import { ChannelMessages } from "./ChannelMessages"

export const dynamic = "force-dynamic"

type Props = {
  params: Promise<{ channelId: string }>
}

export default async function MessagesPage({ params }: Props) {
  const { data: session } = await auth.getSession()

  if (!session?.user) {
    redirect("/auth/sign-in")
  }

  const channelId = Number((await params).channelId)

  if (!Number.isInteger(channelId)) {
    notFound()
  }

  const { initialData, sealedQuery } = await loadMessagesPage(session.user.id, channelId)

  return (
    <ChannelMessages
      key={sealedQuery.queryFingerprint}
      channelId={channelId}
      initialData={initialData}
      sealedQuery={sealedQuery}
    />
  )
}
```

`ChannelMessages` is a Client Component. It builds `refreshQuery`, which [renews the sealed query](#renew-from-the-server) through a Route Handler, and renders [`HydratedMessages`](#render-the-initial-rows-straight-away):

```tsx filename="app/channels/[channelId]/ChannelMessages.tsx"
"use client"

import { useCallback } from "react"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../../../backend/schema"
import { HydratedMessages } from "../../../frontend/HydratedMessages"

interface Props {
  channelId: number;
  initialData: Message[];
  sealedQuery: SealedLiveQuery<Message>;
}

export function ChannelMessages({ channelId, initialData, sealedQuery }: Props) {
  // Renewals must be for the same query, so renew with the same channel
  const refreshQuery = useCallback(async () => {
    const res = await fetch(`/api/messages/${channelId}/live`, { method: "POST" })

    if (!res.ok) {
      throw new Error(`Could not fetch the sealed query (${res.status})`)
    }

    return (await res.json()) as SealedLiveQuery<Message>
  }, [channelId])

  return <HydratedMessages sealedQuery={sealedQuery} refreshQuery={refreshQuery} initialData={initialData} />
}
```

The Route Handler takes the user from the session cookie, not from the request, and validates `channelId`, because the browser controls the URL. Then it seals the same `messagesQuery()` as the page:

```ts filename="app/api/messages/[channelId]/live/route.ts"
import { messagesQuery } from "../../../../../backend/queries"
import { realtime } from "../../../../../backend/realtime"
import { auth } from "../../../../../lib/auth/server"

type Context = {
  params: Promise<{ channelId: string }>
}

export async function POST(_request: Request, { params }: Context) {
  const { data: session } = await auth.getSession()

  if (!session?.user) {
    return Response.json({ error: "Unauthorized" }, { status: 401 })
  }

  const channelId = Number((await params).channelId)

  if (!Number.isInteger(channelId)) {
    return Response.json({ error: "channelId must be an integer" }, { status: 400 })
  }

  const sealedQuery = await realtime.seal({ query: messagesQuery(session.user.id, channelId) })

  return Response.json(sealedQuery, { headers: { "Cache-Control": "no-store" } })
}
```

A server re-render, for example from `router.refresh()` or `revalidatePath()`, sends a new object for the same query. `useLiveQuery()` restarts on each new object, which rejects pending `awaitTxId()` calls. So `HydratedMessages` keeps the first sealed query, and renewals go through `refreshQuery`. Callers key it on `queryFingerprint`, so another channel or user remounts it.

Or with [TanStack Start](https://neon.com/docs/realtime/examples/tanstack.md), [create server functions](https://tanstack.com/start/latest/docs/framework/react/guide/server-functions) for the loader and the renewal. During SSR, a route loader needs the session from the request's cookies on the server. The official Managed Better Auth adapter for TanStack Start isn't out yet, so set up the session yourself. `requireUser` reads the cookie during SSR and for calls from the browser:

<details>

<summary>Read the Managed Better Auth session in TanStack Start</summary>

This setup is built on the beta Managed Better Auth server toolkit, `@neondatabase/auth/server`. Swap in the official adapter when it ships.

Create the auth server. It reads and sets the session cookies through TanStack Start's request helpers:

```ts filename="src/backend/auth.ts"
import { createAuthServer, extractNeonAuthCookies } from "@neondatabase/auth/server"
import { getRequest, setCookie } from "@tanstack/react-start/server"

export const authConfig = {
  baseUrl: process.env.NEON_AUTH_BASE_URL!,
  cookieSecret: process.env.NEON_AUTH_COOKIE_SECRET! // 32+ characters
}

export const auth = createAuthServer({
  ...authConfig,
  context: () => {
    const request = getRequest()

    return {
      getCookies: () => extractNeonAuthCookies(request.headers),
      setCookie,
      getHeader: (name) => request.headers.get(name),
      getOrigin: () => request.headers.get("origin") ?? new URL(request.url).origin,
      getFramework: () => "tanstack-start"
    }
  }
})
```

`NEON_AUTH_COOKIE_SECRET` signs the session cookie and must be at least 32 characters. Generate it once:

```bash filename="Terminal"
printf '\nNEON_AUTH_COOKIE_SECRET=%s\n' "$(openssl rand -base64 32)" >> .env.local
```

Mount the Managed Better Auth proxy at `/api/auth`, so the session cookie lives on your app's origin:

```ts filename="src/routes/api/auth/$.ts"
import { handleAuthProxyRequest } from "@neondatabase/auth/server"
import { createFileRoute } from "@tanstack/react-router"

import { authConfig } from "../../../backend/auth"

export const Route = createFileRoute("/api/auth/$")({
  server: {
    handlers: {
      ANY: ({ request, params }) =>
        handleAuthProxyRequest({ request, path: params._splat ?? "", ...authConfig })
    }
  }
})
```

`requireUser` only has a server half. It puts the user's id on `context`, and throws without a session, so it guards your data. `getSessionUser` reads the session without throwing, for route guards:

```ts filename="src/session-middleware.ts"
import { createMiddleware, createServerFn } from "@tanstack/react-start"
import { setResponseHeader } from "@tanstack/react-start/server"

import { auth } from "./backend/auth"

export const requireUser = createMiddleware({ type: "function" }).server(async ({ next }) => {
  const { data } = await auth.getSession()

  if (!data?.user) {
    throw new Error("Unauthorized")
  }

  return next({ context: { user: { id: data.user.id } } })
})

export const getSessionUser = createServerFn({ method: "GET" }).handler(async () => {
  setResponseHeader("Cache-Control", "private, no-store")
  const { data } = await auth.getSession()

  return data?.user ? { id: data.user.id } : null
})
```

Point the auth client at the proxy. An empty URL makes it use `/api/auth` on the page's origin:

```ts filename="src/frontend/auth.ts"
import { createAuthClient } from "@neondatabase/auth"

export const authClient = createAuthClient("") // same origin: /api/auth
```

</details>

The root route puts the user on the router context for every navigation. During SSR, `getSessionUser` runs in the same process, with the request's cookies. The route's `headers` keep every page out of shared caches:

```tsx filename="src/routes/__root.tsx"
import { createRootRoute } from "@tanstack/react-router"

import { getSessionUser } from "../session-middleware"

export const Route = createRootRoute({
  beforeLoad: async () => ({ user: await getSessionUser() }),
  // Pages carry one user's rows and sealed query, so no shared cache may store them
  headers: () => ({ "Cache-Control": "private, no-store", Vary: "Cookie" })
  // … shell unchanged
})
```

The redirects below go to the sign-in route from the [Managed Better Auth TanStack Router quick start](https://neon.com/docs/auth/quick-start/tanstack-router.md), `src/routes/auth.$pathname.tsx`.

Both server functions use `requireUser`, so they take the user from the session and only `channelId` from the request. The route's `beforeLoad` sends signed-out visitors to sign-in before the loader runs. The router reuses the component when only `$channelId` changes, so the route keys `HydratedMessages` on the query's fingerprint:

```tsx filename="src/routes/channels.$channelId.tsx"
import { useCallback } from "react"
import { createFileRoute, redirect } from "@tanstack/react-router"
import { createServerFn } from "@tanstack/react-start"

import { loadMessagesPage } from "../backend/messages-page"
import { messagesQuery } from "../backend/queries"
import { realtime } from "../backend/realtime"
import { HydratedMessages } from "../frontend/HydratedMessages"
import { requireUser } from "../session-middleware"

function validateChannel(data: { channelId: number }) {
  if (!Number.isInteger(data.channelId)) {
    throw new Error("channelId must be an integer")
  }

  return data
}

const loadMessages = createServerFn({ method: "POST" })
  .middleware([requireUser])
  .validator(validateChannel)
  .handler(({ data, context }) => loadMessagesPage(context.user.id, data.channelId))

const sealMessages = createServerFn({ method: "POST" })
  .middleware([requireUser])
  .validator(validateChannel)
  .handler(({ data, context }) =>
    realtime.seal({ query: messagesQuery(context.user.id, data.channelId) })
  )

export const Route = createFileRoute("/channels/$channelId")({
  beforeLoad: ({ context }) => {
    if (!context.user) {
      throw redirect({ to: "/auth/$pathname", params: { pathname: "sign-in" } })
    }
  },
  loader: ({ params }) => loadMessages({ data: { channelId: Number(params.channelId) } }),
  component: MessagesRoute
})

function MessagesRoute() {
  const { channelId } = Route.useParams()
  const { initialData, sealedQuery } = Route.useLoaderData()

  // Renewals must be for the same query, so renew with the same channel
  const refreshQuery = useCallback(
    () => sealMessages({ data: { channelId: Number(channelId) } }),
    [channelId]
  )

  return (
    <HydratedMessages
      key={sealedQuery.queryFingerprint}
      initialData={initialData}
      sealedQuery={sealedQuery}
      refreshQuery={refreshQuery}
    />
  )
}
```

**Important: The sealed query is in your HTML**

Anyone with the page source holds the sealed query until it expires, because [sealed queries are bearer tokens](queries.md#sealing-queries). Serve the page over HTTPS, keep it out of shared caches, and keep the sealed query out of URLs and logs.

## Render the initial rows straight away

Pass the initial rows to [`useLiveQuery()`](reactivity.md) as `initialData`. The hook renders them immediately with the status `connecting`. When the subscription's first result arrives, it replaces them atomically and the status becomes `live`. Any change committed between the server read and the subscription is in that result, so nothing falls into the gap.

The initial rows are real rows. Render them as you would live ones, without a spinner or a loading state, while the status is `connecting`. `HydratedMessages` shows the server's rows straight away and swaps in the live rows when the first result arrives:

```tsx filename="frontend/HydratedMessages.tsx"
"use client"

import { useState } from "react"
import { useLiveQuery } from "@neon/realtime-react"
import type { SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "../backend/schema"
import { LocalTime } from "./LocalTime"

interface HydratedMessagesProps {
  sealedQuery: SealedLiveQuery<Message>;
  refreshQuery: () => Promise<SealedLiveQuery<Message>>;
  initialData: Message[];
}

export function HydratedMessages({ sealedQuery, refreshQuery, initialData }: HydratedMessagesProps) {
  const [query] = useState(sealedQuery) // renewals go through refreshQuery
  const { data, error } = useLiveQuery(query, { refreshQuery, initialData })

  return (
    <>
      {error && <p role="alert">{error.message}</p>}
      <ul>
        {(data ?? initialData).map((message) => (
          <li key={message.id}>
            <LocalTime date={message.created_at} />
            <p>{message.body}</p>
          </li>
        ))}
      </ul>
    </>
  )
}
```

The component takes `refreshQuery` as a prop, so it works with whichever renewal path your framework uses. [Renew from the server](#renew-from-the-server) covers the options. The [reactivity guide](reactivity.md#uselivequery) covers the hook itself and defines `LocalTime`.

## Open the connection in the browser

The server render doesn't open a WebSocket. On the server, the hook renders from its initial rows and never subscribes. In the browser, it renders the same rows while React hydrates, so the HTML matches, and subscribes once hydration is done.

So create the client once, inside the React tree, in a `useState` initializer. It runs once per mounted app, and the client it creates during the server render never connects, because nothing subscribes there:

**Next.js**

```tsx filename="frontend/RealtimeRoot.tsx"
"use client"

import { useState, type ReactNode } from "react"
import { createRealtimeClient } from "@neon/realtime/client"
import { drizzleParsers } from "@neon/realtime-drizzle/client"
import { RealtimeProvider } from "@neon/realtime-react"

export function RealtimeRoot({ children }: { children: ReactNode }) {
  const [client] = useState(() =>
    createRealtimeClient({
      url: process.env.NEXT_PUBLIC_NEON_REALTIME_URL!,
      parsers: { ...drizzleParsers }
    })
  )

  return <RealtimeProvider client={client}>{children}</RealtimeProvider>
}
```

**TanStack Start**

```tsx filename="frontend/RealtimeRoot.tsx"
"use client"

import { useState, type ReactNode } from "react"
import { createRealtimeClient } from "@neon/realtime/client"
import { drizzleParsers } from "@neon/realtime-drizzle/client"
import { RealtimeProvider } from "@neon/realtime-react"

export function RealtimeRoot({ children }: { children: ReactNode }) {
  const [client] = useState(() =>
    createRealtimeClient({
      url: import.meta.env.VITE_NEON_REALTIME_URL,
      parsers: { ...drizzleParsers }
    })
  )

  return <RealtimeProvider client={client}>{children}</RealtimeProvider>
}
```

With Next.js, add the browser's copy of the Realtime URL. Next.js expands `$VAR` in env files, so it follows `neon env pull`:

```bash filename="Terminal"
echo 'NEXT_PUBLIC_NEON_REALTIME_URL=$NEON_REALTIME_URL' >> .env.local
```

With TanStack Start, use the `VITE_NEON_REALTIME_URL` line from the [setup](queries.md#setup).

The [subscribing guide](subscribing.md#subscribe-to-the-sealed-query) creates the client at module level in `frontend/client.ts`, which suits a single-page app. In a server-rendered app, create it inside the React tree instead, which keeps browser-only setup out of module scope. Keep the client stable, too. `useLiveQuery()` subscribes per client and sealed query, so a new client on every render means a new subscription on every render.

Render `RealtimeRoot` around everything that calls `useLiveQuery()`, for example in your root layout. `RealtimeRoot`, `ChannelMessages` and `HydratedMessages` start with `"use client"` so that Next.js renders them as Client Components. TanStack Start ignores the directive.

## Keep row shapes identical

The initial rows and the live rows have to have the same shape. If they don't, the page renders one shape until the first result and another after it.

In the running example, `created_at` is a `Date` on both sides: Drizzle reads `timestamptz` columns as `Date` objects, and so does the Realtime client. The Realtime clients on this page spread `drizzleParsers`, so the live rows match Drizzle's types for other columns too. The [subscribing guide](subscribing.md#subscribe-to-the-sealed-query) explains what the parsers change. The hydration payload has to keep it that way. React Server Component props and TanStack Start's serializer both preserve `Date` values. Plain `JSON.stringify` turns them into strings, so if your payload goes through it, revive the dates in the browser or keep your row types JSON-safe.

Watch numeric types too: `int8` and `numeric` columns [arrive as strings](subscribing.md#subscribe-to-the-sealed-query) on the live side, so read them as strings on the server.

Formatted output has to match too, so format dates and numbers for the user's locale after hydration, as `LocalTime` in the reactivity guide does.

## Renew from the server

The sealed query in the payload expires, so pass `refreshQuery`, and the hook renews the sealed query with the same timing as in the [reactivity guide](reactivity.md#refreshquery). If the page hydrates late, for example in a tab restored after a few minutes, the sealed query may already have expired, and the hook renews it straight away.

The renewal runs in the browser, so it goes through something that authenticates the user, like the Route Handler or server function above. It has to seal the same query with the same parameters, because a different fingerprint is rejected. Building both from `messagesQuery()` keeps them in step.

## When the session changes

A page can outlive its session. The user signs out in another tab, or another tab signs in as someone else, and the page still holds the first user's sealed query and rows.

Stop renewing once a renewal fails as signed out, and leave the page. The hook retries a failed renewal every second without limit, and once the sealed query expires the status stays `stale`.

Check on every renewal and write that the session's user is the one the page was loaded for. Have the page send its user id, and treat a mismatch as signed out, with the same 401 as no session. Otherwise a write from the old page lands in the new user's account.

Leave with a full page load. It drops the previous user's Realtime client, collections and optimistic rows. Then the page's redirect sends a signed-out visitor to sign-in, and a different user gets their own rows. With Next.js, pass `session.user.id` to `ChannelMessages` as `userId`, and have the Route Handler return its 401 when the body's `userId` isn't `session.user.id`:

```tsx filename="app/channels/[channelId]/ChannelMessages.tsx"
// A full page load resets this
let signedOut = false

export function ChannelMessages({ channelId, userId, initialData, sealedQuery }: Props) {
  const refreshQuery = useCallback(async () => {
    if (signedOut) {
      throw new Error("Signed out")
    }

    // The page's user, so the Route Handler can refuse anyone else's session
    const res = await fetch(`/api/messages/${channelId}/live`, {
      method: "POST",
      body: JSON.stringify({ userId })
    })

    if (res.status === 401) {
      signedOut = true // the hook retries every second, so stop calling the server
      window.location.reload()
      throw new Error("Signed out")
    }

    // … check res.ok and return the sealed query, as above
  }, [channelId, userId])

  // … render HydratedMessages, as above
}
```

With TanStack Start, have the server functions also take a `userId`, and make `requireUser` throw when it isn't the session's user. Leave the same way when a write fails as signed out.

Make sign-in and sign-out full page loads too. With the Managed Better Auth UI components, have the provider's `onSessionChange` call `window.location.assign("/")` and return a promise that never settles, so the components don't also navigate in the client.

## With TanStack DB

With [TanStack DB](reactivity.md#tanstack-db), the collection carries the initial rows instead of the hook's `initialData`:

1. a route loader reads and seals the inbox and seeds the collection in the router's `DbClient`
2. `routerWithDbClient()` carries the seeded rows to the browser
3. the collection subscribes in the browser, and the first result takes over from the seeded rows

The code below does this with TanStack Start. It syncs the inbox from the [reactivity guide](reactivity.md#seal-a-broader-query) into one collection and shows one channel from it. As well as the packages from that guide, it uses TanStack's router integration:

```bash filename="Terminal"
npm install @tanstack/react-router-with-db
```

### Create a `DbClient` per router

TanStack Start calls `getRouter()` once per request on the server and once per page load in the browser, so a `DbClient` created with the router belongs to one request or one page load. Create the Realtime client next to it and pass it to the `DbClient` as a dependency, so collections can read it. The server-side Realtime client never connects, because nothing subscribes there. `routerWithDbClient()` wraps the app in `DbProvider`, snapshots the collections that loaders seeded on the server into the HTML, and restores them in the browser before the first render:

```tsx filename="src/router.tsx"
import { DbClient } from "@tanstack/react-db"
import { createRouter } from "@tanstack/react-router"
import { routerWithDbClient } from "@tanstack/react-router-with-db"
import { createRealtimeClient } from "@neon/realtime/client"
import { drizzleParsers } from "@neon/realtime-drizzle/client"

import { routeTree } from "./routeTree.gen"

export function getRouter() {
  // Collections read it with db.requireDependency("realtime"). Nothing subscribes on the server, so it never connects there
  const realtime = createRealtimeClient({
    url: import.meta.env.VITE_NEON_REALTIME_URL,
    parsers: { ...drizzleParsers }
  })
  const dbClient = new DbClient({ realtime })
  const router = createRouter({ routeTree, context: { dbClient } })

  return routerWithDbClient(router, dbClient)
}
```

`routerWithDbClient()` requires `dbClient` in the router's context type, so create the root route from [Send both to the browser](#send-both-to-the-browser) with `createRootRouteWithContext<{ dbClient: DbClient }>()` instead of `createRootRoute()`, and keep its `beforeLoad` and `headers`. If the app also uses `useLiveQuery()` from `@neon/realtime-react`, pass the same Realtime client to `RealtimeProvider`. Get it with `useDbClient().requireDependency<RealtimeClient>("realtime")`.

### Define the collection

Define the collection as a descriptor with the two-argument `collectionOptions(id, factory)` around [`realtimeCollectionOptions()`](https://neon.com/docs/realtime/sdks/typescript.md#tanstack-db-realtimecollectionoptions), and give it a stable id. Each `DbClient` runs the factory once, to create its own instance of the collection. Unlike `createMessagesCollection()` in the reactivity guide, it doesn't take a Realtime client: the factory reads the one from its `DbClient`. The descriptor is cheap to build, so the loader and the channel page each build it, and the id resolves both to the same collection:

```ts filename="src/collections.ts"
import { collectionOptions } from "@tanstack/react-db"
import { realtimeCollectionOptions } from "@neon/realtime-tanstack"
import type { RealtimeClient, SealedLiveQuery } from "@neon/realtime/client"

import type { Message } from "./backend/schema"
import { sealInbox } from "./server-functions"

export function messagesCollection(sealedQuery: SealedLiveQuery<Message>) {
  return collectionOptions("messages", (db) =>
    realtimeCollectionOptions({
      id: "messages",
      client: db.requireDependency<RealtimeClient>("realtime"),
      query: sealedQuery,
      refreshQuery: () => sealInbox(),
      getKey: (message) => message.id,
      gcTime: 0 // never collect it while the page is open: a rebuilt collection would reuse an expired sealed query
    })
  )
}
```

`refreshQuery` calls a server function that seals the same `inboxQuery()` for the user in the session:

```ts filename="src/server-functions.ts"
import { createServerFn } from "@tanstack/react-start"

import { inboxQuery } from "./backend/queries"
import { realtime } from "./backend/realtime"
import { requireUser } from "./session-middleware"

export const sealInbox = createServerFn({ method: "POST" })
  .middleware([requireUser])
  .handler(({ context }) => realtime.seal({ query: inboxQuery(context.user.id) }))
```

### Seed the collection in a loader

A layout route, `src/routes/channels.tsx`, loads the inbox for every channel page. Its loader calls a server function that reads and seals the inbox. Then it seeds the collection in the router's `DbClient`, passing the rows to `dbClient.collection()` as `initialData`, which is TanStack DB's option for seeding a collection in one `DbClient`. It returns only the sealed query. Its `beforeLoad` sends signed-out visitors to sign-in, so the channel pages, which render in its outlet, don't need to:

```tsx filename="src/routes/channels.tsx"
import { createFileRoute, redirect } from "@tanstack/react-router"
import { createServerFn } from "@tanstack/react-start"

import { inboxQuery } from "../backend/queries"
import { realtime } from "../backend/realtime"
import { messagesCollection } from "../collections"
import { requireUser } from "../session-middleware"

const loadInbox = createServerFn({ method: "POST" })
  .middleware([requireUser])
  .handler(async ({ context }) => {
    const query = inboxQuery(context.user.id)
    const [initialData, sealedQuery] = await Promise.all([query, realtime.seal({ query })])

    return { initialData, sealedQuery }
  })

export const Route = createFileRoute("/channels")({
  staleTime: Infinity,
  beforeLoad: ({ context }) => {
    if (!context.user) {
      throw redirect({ to: "/auth/$pathname", params: { pathname: "sign-in" } })
    }
  },
  loader: async ({ context }) => {
    const { initialData, sealedQuery } = await loadInbox()
    context.dbClient.collection(messagesCollection(sealedQuery), { initialData }) // a no-op if it already exists

    return { sealedQuery }
  }
})
```

The layout's loader runs once, because `staleTime: Infinity` keeps its data fresh, so switching channels makes no server round trip. On the first load, the loader and the channel page both run on the server, so the HTML has the rows. If the loader runs again in the browser, the collection already exists, so seeding it again does nothing and the live rows stay.

One collection id is enough, because each `DbClient` belongs to one request or one page load, and the [full page load when the session changes](#when-the-session-changes) drops it.

The channel route builds the same descriptor from the layout's sealed query, so it gets the collection the loader seeded. It queries the collection with TanStack's `useLiveQuery()`, under the same alias as in the reactivity guide:

```tsx filename="src/routes/channels.$channelId.tsx"
import { eq, useLiveQuery as useCollectionQuery } from "@tanstack/react-db"
import { createFileRoute, getRouteApi } from "@tanstack/react-router"

import { messagesCollection } from "../collections"

const channels = getRouteApi("/channels")

export const Route = createFileRoute("/channels/$channelId")({
  component: ChannelRoute
})

function ChannelRoute() {
  const { channelId } = Route.useParams()
  const messages = messagesCollection(channels.useLoaderData().sealedQuery) // same id, so the seeded collection

  const { data, isError } = useCollectionQuery({
    query: (q) =>
      q
        .from({ message: messages })
        .where(({ message }) => eq(message.channel_id, Number(channelId)))
        .orderBy(({ message }) => message.created_at, "desc")
  })

  if (isError) {
    return <p role="alert">Couldn't sync messages</p>
  }

  return (
    <ul>
      {data.map((message) => (
        <li key={message.id}>{message.body}</li>
      ))}
    </ul>
  )
}
```

Render `data` straight away, without waiting for `isReady`. The seeded rows are there for the first render, but the collection only becomes ready when the first result arrives.

## Keep secrets on the server

In a server-rendered app, server and browser code live side by side, so it's easy to import a module on the wrong side. Import `backend/realtime.ts` and `backend/db.ts` only from server code: Server Components, server functions and TanStack Start server function handlers. A TanStack Start route loader also runs in the browser on client-side navigation, so it has to call a server function rather than import `backend/` modules.

Mark those modules as server-only, so a client import fails the build. In Next.js, add `import "server-only"` to `backend/realtime.ts` and `backend/db.ts`. In TanStack Start, add `import "@tanstack/react-start/server-only"` to them, or name the files `*.server.ts`. The queries guide explains why you [keep your Realtime secret on the server](queries.md#sealing-queries).

## Examples

The examples put these principles into framework code. The Next.js example splits the page into a Server Component that reads and seals and a Client Component that subscribes. Instead of `force-dynamic`, it calls `connection()` in that Server Component and renders it inside `<Suspense>`, so the rest of the page renders at once and the list streams in. The TanStack example seeds the router's TanStack DB client in a route loader.

- [Next.js](https://neon.com/docs/realtime/examples/nextjs.md): Read and seal in a Server Component, then subscribe in a Client Component.
- [TanStack](https://neon.com/docs/realtime/examples/tanstack.md): Seed the router's TanStack DB client in a route loader.

<!-- Generated by scripts/sync-skill-docs.mjs from website@thruflo/realtime-2-docs:24a2fd708 (docs/realtime/guides/subscribing). Do not edit: change the docs and run npm run sync:docs. sha256:6d28cb355161f30b463590c44edc4e95dce5a899544807f30530cf4093bc074c -->
Canonical page: https://neon.com/docs/realtime/guides/subscribing.md

> 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.

# Subscribing to queries

Subscribe to live queries using the Realtime client.

## Contents

- [Getting the query to subscribe to](#getting-the-query-to-subscribe-to)
  - [Subscribe in trusted backend code](#subscribe-in-trusted-backend-code)
  - [Fetch from an API endpoint](#fetch-from-an-api-endpoint)
  - [Fetch from a server function](#fetch-from-a-server-function)
- [Subscribe to the sealed query](#subscribe-to-the-sealed-query)
- [Listen for changes](#listen-for-changes)
- [Renew the sealed query](#renew-the-sealed-query)
- [Handle errors](#handle-errors)
- [Debug subscriptions](#debug-subscriptions)
  - [Log client diagnostics](#log-client-diagnostics)
  - [Show full database errors](#show-full-database-errors)

Once you've [defined](queries.md) and [sealed](queries.md#sealing-queries) a query, you can [subscribe to it](#subscribe-to-the-sealed-query) using the [low‑level Realtime client](https://neon.com/docs/realtime/sdks/typescript.md#client-createrealtimeclient).

**Important: Client-side reactivity**

Manual subscriptions are a relatively low-level API. Most apps and agents can use the [higher‑level reactivity primitives](reactivity.md) that manage the subscription for you.

## Getting the query to subscribe to

Workers, trusted agents and other backend processes can [subscribe to a query directly](#subscribe-in-trusted-backend-code).

However, to subscribe to a live query in an untrusted environment, you'll need to [expose a sealed query from your backend](queries.md#exposing-queries), either a [standard API endpoint](#fetch-from-an-api-endpoint) or [using a server function](#fetch-from-a-server-function) (where [SSR](ssr.md) can embed the sealed query and the initial results).

**Tip: Running example**

The guide continues the [running example](https://github.com/neondatabase/examples/tree/main/realtime/guides-chat-app) introduced in the [Defining queries](queries.md#setup) guide.

### Subscribe in trusted backend code

Add a `url` to [`createRealtime()`](https://neon.com/docs/realtime/sdks/typescript.md#backend-createrealtime) and it returns a direct server, which seals queries with your secret and subscribes to them itself.

```ts filename="backend/worker.ts" {11,19,20,22}
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](https://neon.com/docs/realtime/quickstart.md#enable-realtime) 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](#listen-for-changes) 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](queries.md#setup), even on Node.js 22 and later.

Use direct subscriptions for workers, trusted agents and server-side fan-out.

### Fetch from an API endpoint

Most queries depend on who's asking and what they're asking for.

If you've [exposed your sealed query via an API endpoint](queries.md#exposing-an-endpoint), you can provide this context via the request data using your standard auth machinery, such as Neon's [Managed Better Auth](https://neon.com/docs/auth/overview.md):

```ts filename="frontend/auth.ts"
import { createAuthClient } from "@neondatabase/auth"

export const authClient = createAuthClient(import.meta.env.VITE_NEON_AUTH_URL)
```

In our [running example](https://github.com/neondatabase/examples/tree/main/realtime/guides-chat-app), 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:

```ts filename="frontend/api.ts"
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](https://github.com/neondatabase/examples/tree/main/realtime/guides-chat-app/vite.config.ts) does.

### Fetch from a server function

If your framework supports [server functions](https://react.dev/reference/rsc/server-functions), 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](#fetch-from-an-api-endpoint). With [SSR](ssr.md), 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](queries.md#exposing-queries), [Using server functions](queries.md#using-server-functions), the [Server-side rendering](ssr.md) guide and the [TanStack](https://neon.com/docs/realtime/examples/tanstack.md) 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](https://nextjs.org/docs/app/building-your-application/routing/route-handlers) instead, as the [Next.js](https://neon.com/docs/realtime/examples/nextjs.md) example shows.

## Subscribe to the sealed query

Create one client for the whole app, because every subscription shares its connection:

```ts filename="frontend/client.ts"
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:

```ts filename="frontend/app.ts"
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.

## Listen for changes

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](ssr.md), 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](#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:

```ts filename="frontend/raw-messages.ts"
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](mutations.md) uses them to confirm optimistic mutations.

## Renew the sealed query

Sealed queries are [short-lived](queries.md#renewing-sealed-queries), 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:

```ts filename="frontend/renewal.ts"
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:

```ts filename="frontend/app.ts" {6,16,21}
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:

- Before expiry, the subscription stays `live`, with no new `rows` update.
- After expiry, for example when a laptop wakes from sleep, the subscription is `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](reactivity.md) take a `refreshQuery` function and renew with the same timing, 10 seconds before expiry with a 1-second retry. They're built on [`QueryRefreshController`](https://neon.com/docs/realtime/sdks/typescript.md#client-queryrefreshcontroller), which you can use if you're writing a binding for another framework.

## Handle errors

When automatic recovery stops, the subscription moves to `error`, and its `error` is a [`LiveQueryError`](https://neon.com/docs/realtime/sdks/typescript.md#client-livequeryerror) with:

- `code`: a stable, machine-readable error code
- `message`: a description of the failure
- `retryable`: whether reconnecting or a fresh sealed query may recover it
- `sqlState`: the Postgres SQLSTATE code, when [debug mode](#debug-subscriptions) is on and the database supplied one
- `cause`: the underlying error, when there is one

A 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.

## Debug subscriptions

### Log client diagnostics

The client is silent by default. To see what it's doing, pass a `logLevel` to [`createRealtimeClient()`](https://neon.com/docs/realtime/sdks/typescript.md#client-createrealtimeclient): `"error"`, `"warn"`, `"info"` or `"debug"`. Each level includes the ones before it.

```ts filename="frontend/client.ts" {7}
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`](https://neon.com/docs/realtime/sdks/typescript.md#client-realtimelogger), which receives each [entry](https://neon.com/docs/realtime/sdks/typescript.md#client-realtimelogentry) 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`:

```ts
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 })`.

### Show full database errors

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:

```ts filename="backend/realtime.ts" {8}
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.

---
name: neon-realtime
description: >-
  Neon Realtime makes any SQL query live. Build realtime, reactive apps and agents on Lakebase Postgres with live SQL queries and reactivity primitives. Use when app UI or agent state must update live from database rows (dashboards, feeds, collaboration, multi-user or multi-device sync), for optimistic mutations, or to replace polling or other realtime sync systems. Triggers include "live query", "subscribe", "sync engine", "useLiveQuery", "TanStack DB", "optimistic updates", "@neon/realtime". For push that doesn't come from database rows (presence, typing, LLM streams), use neon-functions.
metadata:
  parent: neon
  source: https://github.com/neondatabase/agent-skills/tree/main/skills/neon-realtime
---

**FIRST**: Use the parent `neon` skill for a Neon overview, getting started with Neon, Neon development best practices, and more.

If the `neon` skill is not installed, fetch it from https://neon.com/docs/ai/skills/neon/SKILL.md or install it with:

```bash
neon skills -s neon -y
```

# Neon Realtime

Neon's Realtime makes any SQL query live. Subscribe in realtime to your Lakebase Postgres database. Develop apps and agents with end-to-end reactivity using type-safe live queries and optimistic mutations.

Realtime is new and in beta, so its APIs aren't in your training data. Start from the core pattern below, and read each section's reference file before writing that code.

## How It Works

Realtime's layers compose: a live query engine inside Postgres, the Live SQL API (a WebSocket API), lower-level client SDKs (`seal()`, `subscribe()`), and reactivity primitives such as `useLiveQuery`.

1. In trusted backend code, **define** a live query (any read-only SQL, including joins, aggregates, ordering and limits), **seal** it with `NEON_REALTIME_SECRET` and **expose** it from an authenticated endpoint or server function. The client can't read or change a sealed query's SQL or parameters.
2. The client **subscribes** with the sealed query. It receives the full initial result, then each committed transaction's changes as one batch. Results stay consistent, but delivery of every change event isn't guaranteed.
3. **Writes** go through your own API, which returns each write's `txid` for `awaitTxId()`.

## Core Pattern

The guides' chat app, with `messages` authorized through `channel_members`. The backend seals the query and returns each write's `txid`:

```ts
import { createRealtime, rawSql } from "@neon/realtime/server"
import { drizzleAdapter } from "@neon/realtime-drizzle"
import { and, desc, eq, getTableColumns, sql } from "drizzle-orm"
import { Hono } from "hono"
import { HTTPException } from "hono/http-exception"
import { requireUser } from "./auth"
import { db } from "./db"
import { channelMembers, messages, type Message } from "./schema"

export const realtime = createRealtime({ secret: process.env.NEON_REALTIME_SECRET!, db: "neondb", adapter: drizzleAdapter() })

// Each selected key must match its column name
export function messagesQuery(userId: string, channelId: number) {
  return db
    .select(getTableColumns(messages)).from(messages)
    .innerJoin(channelMembers, eq(channelMembers.channel_id, messages.channel_id))
    .where(and(eq(channelMembers.user_id, userId), eq(messages.channel_id, channelId)))
    .orderBy(desc(messages.created_at))
    .limit(50)
}

// Or raw SQL
const rawMessagesQuery = (userId: string, channelId: number) => rawSql<Message>(
    `SELECT messages.author_id, messages.body, messages.channel_id, messages.created_at, messages.id FROM messages
     INNER JOIN channel_members ON channel_members.channel_id = messages.channel_id WHERE channel_members.user_id = $1 AND messages.channel_id = $2 ORDER BY messages.created_at DESC LIMIT 50`,
    [userId, channelId]
  )

// The id of the transaction that made the change, read in the write's own RETURNING
const txid = sql<string>`pg_current_xact_id()::text`

const app = new Hono()
  .post("/api/v1/messages/live", async (c) => {
    const user = await requireUser(c)
    const { channelId } = await c.req.json<{ channelId: number }>() // validate input
    return c.json(await realtime.seal({ query: messagesQuery(user.id, channelId) }))
  })
  .post("/api/v1/messages", async (c) => {
    const user = await requireUser(c)
    const { id, channelId, body } = await c.req.json<{ id: string; channelId: number; body: string }>() // validate input
    // One statement. The membership row supplies channel_id and author_id, so only a member's message is inserted.
    // List every column in table order, alias each sql field, and cast untyped params (they resolve to text)
    const [sent] = await db.insert(messages).select(
      db.select({
          id: sql`${id}::uuid`.as("id"), channel_id: channelMembers.channel_id, author_id: channelMembers.user_id,
          body: sql`${body}`.as("body"), created_at: sql`now()`.as("created_at"),
        })
        .from(channelMembers)
        .where(and(eq(channelMembers.channel_id, channelId), eq(channelMembers.user_id, user.id)))
    ).returning({ txid })
    if (!sent) throw new HTTPException(403, { message: "Not a member of this channel" })
    return c.json({ kind: "committed", txid: sent.txid })
  })
```

The React frontend subscribes and sends optimistic messages:

```tsx
import { createRealtimeClient, type SealedLiveQuery } from "@neon/realtime/client"
import { drizzleParsers } from "@neon/realtime-drizzle/client"
import { useLiveQuery } from "@neon/realtime-react"
import { startTransition, useCallback, useOptimistic } from "react"
import type { Message } from "../backend/schema"
import { fetchSealedMessages, sendMessage } from "./api"

// One client per app, because subscriptions share its connection. Wrap the app in <RealtimeProvider client={client}>
export const client = createRealtimeClient({ url: import.meta.env.VITE_NEON_REALTIME_URL, parsers: { ...drizzleParsers } })

const NO_MESSAGES: readonly Message[] = []
type Props = { sealedQuery: SealedLiveQuery<Message>; channelId: number; userId: string }

export function Messages({ sealedQuery, channelId, userId }: Props) {
  const refreshQuery = useCallback(() => fetchSealedMessages(channelId), [channelId])
  const { data, status, error, utils } = useLiveQuery(sealedQuery, { refreshQuery })
  // Once the sent message syncs, show only the synced row
  const [messages, addOptimistic] = useOptimistic(data ?? NO_MESSAGES, (current, message: Message) =>
    current.some(({ id }) => id === message.id) ? current : [message, ...current]
  )

  function send(body: string) {
    // The server sets author and timestamp but keeps this id
    const message = { id: crypto.randomUUID(), channel_id: channelId, author_id: userId, body, created_at: new Date() }
    startTransition(async () => {
      addOptimistic(message)
      try {
        const result = await sendMessage(message) // maps channel_id to channelId
        if (result.kind === "committed") await utils.awaitTxId(result.txid, 10_000)
      } catch { /* show "Couldn't confirm the change" */ }
    })
  }

  if (status === "error") return <p role="alert">Live updates stopped: {error.message}</p>
  if (data === undefined) return <p>Loading messages</p>
  return (
    <form action={(form) => send(String(form.get("body")))}>
      <input name="body" required />
      <ul>{messages.map((message) => <li key={message.id}>{message.body}</li>)}</ul>
    </form>
  )
}
```

## Choose the Stack

A live query replaces fetching app data from an API or another sync system. Reuse existing query code, and use the highest layer that fits rather than wiring changes into app state by hand.

| Use | For |
| --- | --- |
| `drizzleAdapter()` from `@neon/realtime-drizzle`, or `kyselyAdapter()` from `@neon/realtime-kysely` | Queries written in Drizzle or Kysely |
| `rawSql` from `@neon/realtime/server` | Any other ORM, or none |
| `useLiveQuery` from `@neon/realtime-react` | React. Binds one sealed query to a component's state |
| `realtimeCollectionOptions` from `@neon/realtime-tanstack` | Many views over the same data, or an app on TanStack DB. Oversync one broader query into a collection, then derive local live queries from it |
| `client.subscribe()` from `@neon/realtime/client` | Other frameworks |
| `realtime.subscribe()` from `@neon/realtime/server` | Agents and workers, in trusted backend code |

Migrating from Electric: Realtime is a drop-in replacement for Electric's role in your stack. Live SQL queries replace shapes, and `realtimeCollectionOptions()` replaces `electricCollectionOptions()`.

## Setup

- **New app:** `neon bootstrap my-app --template realtime-react`, then `neon deploy` in `my-app`.
- **Enable** it on the linked branch with [`neon realtime enable --allowed-origin <app origin>`](https://neon.com/docs/cli/realtime.md), or declare [`preview: { realtime: true }`](https://neon.com/docs/reference/neon-ts.md#realtime) in `neon.ts` (`neon config add realtime`) and run `neon deploy`. Child branches inherit it. It's unavailable on Claimable projects until claimed, or outside the [supported regions](https://neon.com/docs/realtime/overview.md#which-regions-is-it-enabled-in).
- **Tables** need a [stable replica identity](references/queries.md#table-requirements); a primary key is enough.
- **Environment:** `neon env pull --service postgres --service realtime` writes `DATABASE_URL`, `NEON_REALTIME_URL` and `NEON_REALTIME_SECRET`. Expose only the URL to the browser, under your framework's public prefix (like `VITE_`). `url` options take `NEON_REALTIME_URL`; `db` takes the database name from `DATABASE_URL`.
- **Install** `@neon/realtime` and the packages chosen above. Use these SDKs, not the still-evolving WebSocket protocol.

## Define and Seal Queries

Read [references/queries.md](references/queries.md) before writing query or sealing code, including Kysely and server functions.

- **Read-only and deterministic.** A raw query is a single `SELECT` with no `;` and `$1` placeholders. It can't call `now()` or `gen_random_uuid()`, so pass such values as parameters, wrapping every value except strings, numbers, booleans, bigints and null in a `pgParam` helper (`pgParam.timestamptz(date)`).
- **Authorize in the query.** Check the session before sealing, and join through your membership tables so revoked access drops out. Any web framework auth works, including Managed Better Auth, but not [RLS](references/queries.md#why-not-row-level-security) yet. See [Authorization](references/queries.md#authorization).
- **A sealed query is a bearer token.** Keep it out of URLs, logs and persistent storage, and never fill its parameters from unchecked input.
- **Check a query** before wiring it in: [`neon realtime subscribe "SELECT ..."`](https://neon.com/docs/cli/realtime.md#subscribe) prints the initial result, then each transaction's changes, until you stop it. With `--output json`, each change prints a full snapshot. Set `debugMode` on `createRealtime` for full database errors (development only).
- **Sealed queries expire.** Pass `refreshQuery` to the bindings or call `renew()`, sealing the same query and parameters, or the subscription errors. New parameters need a new sealed query and subscription. In Next.js, renew from a Route Handler, because server actions run one at a time.

## Subscribe

Read [references/reactivity.md](references/reactivity.md) before writing React or TanStack DB code. Read [references/subscribing.md](references/subscribing.md) before subscribing directly with `client.subscribe(sealed)` (`getSnapshot`, `onChange`, `onStateChange`), or from agents and workers, which add `url` to `createRealtime` and `await realtime.subscribe(query)` to seal in-process.

- **One database per client.** The first sealed query binds the connection to its `db`.
- **Select a stable, unique key for every row** and key rows by it, including TanStack DB's `getKey`: usually the primary key, the grouped columns for aggregates, or the combined keys for joins that repeat a row. Never use `rowId`, which every `rows` update invalidates.
- **`error` is final**, even after `renew()`. Fetch a fresh sealed query and subscribe again, with a new client if the connection failed. Watch `onStateChange` on direct subscriptions; `useLiveQuery()` reports errors through `status`, not by throwing or suspending.
- **When the session changes, leave with a full page load.** It drops the Realtime client, any TanStack `DbClient` and the previous user's rows. The hook retries a failed renewal every second, so stop renewing once one fails as signed out. Check on every renewal and write that the session's user is the one the page was loaded for, or a write from the old page lands in the new user's account. See [When the session changes](references/ssr.md#when-the-session-changes).

## Optimistic Writes

Realtime has no write path, so writes keep your backend's usual auth. Read [references/mutations.md](references/mutations.md) before writing mutations, for each layer.

- [Return the txid from the write itself](references/mutations.md#returning-the-txid-from-your-backend): `` .returning({ txid: sql<string>`pg_current_xact_id()::text` }) ``, in one round trip on any Postgres driver. Write a batch in one statement, so there's one txid. Never open `db.transaction()` just to `select pg_current_xact_id()`: that's four round trips holding a pooled connection. When a write genuinely needs several statements, put the txid in the `RETURNING` of a write that always changes a row.
- Always pass `awaitTxId()` a timeout: it keeps waiting through reconnects and renewals, and never settles if the txid doesn't reach the query.
- A write that changes no rows has no txid, so skip `awaitTxId()` and let the optimistic state drop.

## Server-Side Rendering

Read [references/ssr.md](references/ssr.md) before rendering live queries on the server, for Next.js and TanStack Start.

- Create the client in a `useState` initializer rather than at module level, render `initialData` as live rows without a loading state, and never share a TanStack `DbClient` between requests.
- Keep row shapes identical on both sides. `JSON.stringify` turns dates into strings, and `int8` and `numeric` arrive as strings.

## More Docs

| Task | Page |
| --- | --- |
| Beta scope, regions, pricing, FAQ, other systems | [Overview](https://neon.com/docs/realtime/overview.md) |
| Architecture and consistency | [How it works](https://neon.com/docs/realtime/how-it-works.md) |
| A first live query | [Quickstart](https://neon.com/docs/realtime/quickstart.md) |
| Every SDK export | [TypeScript SDK reference](https://neon.com/docs/realtime/sdks/typescript.md) |
| Full-stack apps with auth and SSR | [Next.js](https://neon.com/docs/realtime/examples/nextjs.md), [TanStack](https://neon.com/docs/realtime/examples/tanstack.md) |
| Live agent workflows | [Mastra](https://neon.com/docs/realtime/examples/mastra.md) |
| ORM adapters | [Drizzle](https://neon.com/docs/realtime/examples/drizzle.md), [Kysely](https://neon.com/docs/realtime/examples/kysely.md) |