Setting the file. One moment.
Subchapter 6.1
references/mutations.mdMarkdown32 KBView on GitHub
Canonical page: https://neon.com/docs/realtime/guides/mutations.md (opens in a new tab)
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.
Make apps and agents super fast, with instant local writes.
Apply instant local writes using optimistic mutations that automatically discard their optimistic state when the changes sync in through your live query subscriptions.
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.
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:
awaitTxId() (opens in a new tab)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.
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.
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 and the /api/v1/inbox/live route from the reactivity guide:
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
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() (opens in a new tab). 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:
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.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 … 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 (opens in a new tab) 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.
Add a helper for each route to frontend/api.ts, next to fetchSealedMessages() from the subscribing guide. Add MutationResponse to the file’s type import from ../backend/server, so each helper returns the route’s own response type:
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.
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:
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.
Pick without a framework to see the mechanism or for apps that don’t use React, React’s useOptimistic hook for React apps that keep optimistic state locally in a component, or TanStack DB collections for apps that sync into a normalized client store.
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:
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 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 (opens in a new tab) session, as frontend/main.tsx does in 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.
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. 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:
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
}
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() (opens in a new tab). 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, give Channel a userId prop and pass it to Messages with channelId. Read userId from the Managed Better Auth (opens in a new tab) session where you render Channel, as frontend/main.tsx does in render the app:
// ...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>
}For more control, keep optimistic state in a TanStack DB (opens in a new tab) 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. 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:
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() (opens in a new tab) 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. It runs the same local query, and its insert(), update() and delete() calls apply each change locally at once and call the matching handler:
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 (opens in a new tab) session where the app creates the collection, as frontend/main.tsx does in render the app.
The TanStack example (opens in a new tab) puts these handlers into a full app with server-side rendering, and the Next.js example (opens in a new tab) shows the React binding in a server-rendered app.
frontend/main.tsx from the reactivity guide renders both views inside RealtimeProvider. Read the signed-in user’s id from the session and pass it to Channel and Inbox:
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 (opens in a new tab) shows one way to add it.