neon-realtime

Files (6)

• SKILL.md
• mutations.md · references/
• queries.md · references/
• reactivity.md · references/
• ssr.md · references/
• subscribing.md · references/

SKILL.md (first lines)

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

From neondatabase/agent-skills, Apache-2.0.

More from neondatabase/agent-skills