neon-realtime

首先：使用父级 neon 技能了解 Neon 概况、Neon 入门、Neon 开发最佳实践等内容。

如果尚未安装 neon 技能，请从 https://neon.com/docs/ai/skills/neon/SKILL.md 获取，或使用以下命令安装：

neon skills -s neon -y

Neon Realtime

Neon Realtime 可让任何 SQL 查询实时更新。实时订阅 Lakebase Postgres 数据库。使用类型安全的实时查询和乐观变更，开发具备端到端响应性的应用和 Agent。

Realtime 是一项新功能，目前仍处于 Beta 阶段，因此它的 API 不在你的训练数据中。请从下面的核心模式入手，并在编写相关代码前阅读每个部分的参考文件。

工作原理

Realtime 的各层相互配合：Postgres 内部的实时查询引擎、Live SQL API（WebSocket API）、较底层的客户端 SDK（seal()、subscribe()），以及 useLiveQuery 等响应式原语。

1. 在受信任的后端代码中，定义一个实时查询（任何只读 SQL 查询都可以，包括联接、聚合、排序和限制），使用 NEON_REALTIME_SECRET 对其进行 密封，并从经过身份验证的端点或服务端函数中 公开。客户端无法读取或更改已密封查询的 SQL 或参数。
2. 客户端使用已密封的查询进行 订阅。客户端会先收到完整的初始结果，之后每个已提交事务的变更都会作为一个批次收到。结果始终保持一致，但不保证传送每个变更事件。
3. 写入通过你自己的 API 完成，该 API 会返回每次写入的 txid，供 awaitTxId() 使用。

核心模式

指南中的聊天应用通过 channel_members 授权访问 messages。后端会密封查询，并返回每次写入的 txid：

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 })
  })

React 前端会进行订阅并发送乐观消息：

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>
  )
}

选择技术栈

实时查询可替代通过 API 或其他同步系统获取应用数据的方式。复用现有查询代码，并选用满足需求的最高层级，而不是手动将变更接入应用状态。

用途	适用场景
@neon/realtime-drizzle 中的 drizzleAdapter()，或 @neon/realtime-kysely 中的 kyselyAdapter()	使用 Drizzle 或 Kysely 编写的查询
@neon/realtime/server 中的 rawSql	使用其他 ORM，或不使用 ORM
@neon/realtime-react 中的 useLiveQuery	React。将一个已密封的查询绑定到组件状态
@neon/realtime-tanstack 中的 realtimeCollectionOptions	需要基于相同数据构建多个视图，或使用 TanStack DB 的应用。将更宽泛的查询过度同步到集合中，再从中派生本地实时查询
@neon/realtime/client 中的 client.subscribe()	其他框架
@neon/realtime/server 中的 realtime.subscribe()	受信任后端代码中的 Agent 和工作进程

从 Electric 迁移：Realtime 可直接替代 Electric 在技术栈中的作用。实时 SQL 查询取代 shapes，realtimeCollectionOptions() 取代 electricCollectionOptions()。

设置

• 新建应用： neon bootstrap my-app --template realtime-react，然后在 my-app 中运行 neon deploy。
• 在关联分支上使用 neon realtime enable --allowed-origin <app origin> 启用它，或者在 neon.ts 中声明 preview: { realtime: true }（neon config add realtime），然后运行 neon deploy。子分支会继承此设置。在 Claimable 项目认领之前，或在支持的区域之外，此功能不可用。
• 表需要稳定的副本标识；有主键即可。
• 环境： neon env pull --service postgres --service realtime 会写入 DATABASE_URL、NEON_REALTIME_URL 和 NEON_REALTIME_SECRET。只向浏览器公开 URL，并使用你所用框架的公开前缀（例如 VITE_）。url 选项使用 NEON_REALTIME_URL；db 使用 DATABASE_URL 中的数据库名称。
• 安装 @neon/realtime 和上文选用的程序包。请使用这些 SDK，而不是仍在不断演进的 WebSocket 协议。

定义并封存查询

编写查询或封存代码（包括 Kysely 和服务端函数）之前，请先阅读 references/queries.md。

• 只读且确定性。 原始查询必须是单条 SELECT，不能包含 ;，并使用 $1 占位符。查询不能调用 now() 或 gen_random_uuid()，因此请将这类值作为参数传入，并使用 pgParam 辅助函数包装除字符串、数字、布尔值、大整数和 null 以外的每个值（pgParam.timestamptz(date)）。
• 在查询中进行授权。 封存前检查会话，并通过成员关系表进行联接，以便撤销的访问权限不再生效。任何 Web 框架的身份验证方式均可，包括 Managed Better Auth，但目前还不支持 RLS。请参阅授权。
• 封存查询是持有者令牌。 不要将其放入 URL、日志或持久化存储中，也绝不要使用未经校验的输入填充其参数。
• 接入查询前先进行检查： neon realtime subscribe "SELECT ..." 会先打印初始结果，然后打印每笔事务的变更，直到你停止它。使用 --output json 时，每项变更都会打印完整快照。为 createRealtime 设置 debugMode，即可查看完整的数据库错误（仅限开发环境）。
• 封存查询会过期。 将 refreshQuery 传给绑定，或调用 renew()，以封存相同的查询和参数，否则订阅会报错。使用新参数时，需要创建新的封存查询和订阅。在 Next.js 中，请从 Route Handler 进行续期，因为服务端操作会依次运行。

订阅

编写 React 或 TanStack DB 代码前，请阅读 references/reactivity.md。直接使用 client.subscribe(sealed)（getSnapshot、onChange、onStateChange）订阅，或从 Agent 和 worker 订阅前，请阅读 references/subscribing.md。这两种方式都需要将 url 添加到 createRealtime，并通过 await realtime.subscribe(query) 在进程内封存查询。

• 每个客户端对应一个数据库。 第一个封存的查询会将连接绑定到其 db。
• 为每一行选择稳定且唯一的键，并用它作为行的键，包括 TanStack DB 的 getKey：通常是主键、聚合的分组列，或连接中重复行的组合键。切勿使用 rowId，因为每次 rows 更新都会使其失效。
• error 是最终状态，即使调用了 renew() 也不例外。请获取新的封存查询并重新订阅；如果连接失败，请使用新客户端。对于直接订阅，请监听 onStateChange；useLiveQuery() 会通过 status 报告错误，不会抛出错误或暂停。
• 会话变更时，请通过完整页面加载退出。 这样会丢弃 Realtime 客户端、任何 TanStack DbClient 以及前一位用户的行。hook 每秒重试一次失败的续订，因此如果续订失败是因为用户已退出登录，请停止续订。每次续订和写入时都要检查会话用户是否为加载此页面时的用户，否则旧页面上的写入可能会落入新用户的账户。请参阅会话变更时。

乐观写入

Realtime 不提供写入路径，因此写入仍使用后端常规的身份验证。编写变更前，请阅读 references/mutations.md，了解各层的相关说明。

• 在写入操作本身返回 txid：.returning({ txid: sql<string>`pg_current_xact_id()::text` })，适用于任何 Postgres 驱动程序，且仅需一次往返。将一批写入合并到一条语句中，这样就只有一个 txid。切勿仅为了执行 select pg_current_xact_id() 而开启 db.transaction()：这需要四次往返，并会占用池化连接。如果写入确实需要多条语句，请将 txid 放入必定会更改行的写入操作的 RETURNING 中。
• 始终为 awaitTxId() 传入超时时间：它会在重新连接和续订期间持续等待；如果 txid 未到达查询，则永远不会结束。
• 不更改任何行的写入没有 txid，因此请跳过 awaitTxId()，让乐观状态消失。

服务端渲染

在服务器上渲染实时查询前，请阅读 references/ssr.md，其中涵盖 Next.js 和 TanStack Start。

• 在 useState 初始化器中创建客户端，而不是在模块级别创建；将 initialData 渲染为实时行，不显示加载状态；绝不要在请求之间共享 TanStack DbClient。
• 两端的行形状必须相同。JSON.stringify 会将日期转换为字符串，而 int8 和 numeric 会以字符串形式传入。

更多文档

任务	页面
Beta 范围、地区、定价、常见问题、其他系统	概览
架构与一致性	工作原理
第一个实时查询	快速入门
所有 SDK 导出项	TypeScript SDK 参考
带身份验证和 SSR 的全栈应用	Next.js、TanStack
实时 Agent 工作流	Mastra
ORM 适配器	Drizzle、Kysely