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 データベースにリアルタイムでサブスクライブできます。型安全なライブクエリと楽観的ミューテーションを使用して、エンドツーエンドでリアクティブなアプリやエージェントを開発できます。

Realtime は新しく、ベータ版のため、その API は学習データに含まれていません。まず以下の基本パターンを確認し、コードを書く前に各セクションのリファレンスファイルを読んでください。

仕組み

Realtime の各レイヤーは連携します。Postgres 内のライブクエリエンジン、Live SQL API（WebSocket API）、低レベルのクライアント SDK（seal()、subscribe()）、そして useLiveQuery などのリアクティビティプリミティブがあります。

1. 信頼されたバックエンドコードで、ライブクエリ（結合、集計、並べ替え、件数制限を含む、あらゆる読み取り専用 SQL）を定義し、NEON_REALTIME_SECRET でシールして、認証済みのエンドポイントまたはサーバー関数から公開します。クライアントは、シールされたクエリの SQL やパラメーターを読み取ったり変更したりできません。
2. クライアントはシールされたクエリを使ってサブスクライブします。最初に完全な結果を受け取り、その後はコミットされた各トランザクションの変更を1つのバッチとして受け取ります。結果の一貫性は保たれますが、すべての変更イベントが配信されるとは限りません。
3. 書き込みは独自の API を経由し、awaitTxId() 用に各書き込みの txid を返します。

基本パターン

ガイドのチャットアプリでは、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。シールされた1つのクエリをコンポーネントの状態にバインドします
@neon/realtime-tanstack の realtimeCollectionOptions	同じデータに対する複数のビュー、または TanStack DB を使用するアプリ。より広範なクエリをコレクションにオーバーシンクし、そこからローカルのライブクエリを導出します
@neon/realtime/client の client.subscribe()	その他のフレームワーク
@neon/realtime/server の realtime.subscribe()	信頼されたバックエンドコードで実行するエージェントやワーカー

Electric からの移行：Realtime は、スタック内で Electric が担っていた役割をそのまま置き換えられます。ライブ SQL クエリがシェイプを置き換え、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 と、上記で選んだパッケージをインストールします。進化の途上にある WebSocket プロトコルではなく、これらの SDK を使ってください。

クエリの定義とシール

Kysely やサーバー関数を含め、クエリやシール処理のコードを書く前に references/queries.md をお読みください。

• 読み取り専用かつ決定的であること。 生のクエリは ; を含まない単一の SELECT で、$1 プレースホルダーを使う必要があります。now() や gen_random_uuid() は呼び出せないため、これらの値はパラメーターとして渡してください。文字列、数値、ブール値、bigint、null 以外の値はすべて pgParam ヘルパー（pgParam.timestamptz(date) など）でラップしてください。
• クエリ内で認可してください。 シールする前にセッションを確認し、メンバーシップテーブルを介して結合して、取り消されたアクセス権が反映されるようにしてください。Managed Better Auth を含め、どのウェブフレームワークの認証も利用できますが、RLS にはまだ対応していません。認可を参照してください。
• シール済みクエリはベアラートークンです。 URL、ログ、永続ストレージには保存せず、未検証の入力からパラメーターを設定しないでください。
• クエリを組み込む前に確認してください： neon realtime subscribe "SELECT ..." を実行すると、停止するまで初期結果と各トランザクションの変更が出力されます。--output json を指定すると、変更ごとに完全なスナップショットが出力されます。データベースエラーの詳細をすべて表示するには、createRealtime に debugMode を設定します（開発時のみ）。
• シール済みクエリには有効期限があります。 バインディングに refreshQuery を渡すか、renew() を呼び出して、同じクエリとパラメーターをシールしてください。そうしないとサブスクリプションでエラーが発生します。新しいパラメーターを使う場合は、新しいシール済みクエリとサブスクリプションが必要です。Next.js では、サーバーアクションは一度に 1 つずつしか実行されないため、Route Handler から更新してください。

Subscribe

React または TanStack DB のコードを書く前に、references/reactivity.md をお読みください。client.subscribe(sealed)（getSnapshot、onChange、onStateChange）を使って直接サブスクライブする場合、またはエージェントやワーカーからサブスクライブする場合は、references/subscribing.md をお読みください。エージェントやワーカーでは、createRealtime に url を追加し、await realtime.subscribe(query) を呼び出してインプロセスでクエリを確定します。

• クライアントごとにデータベースは 1 つです。 最初に確定したクエリによって、接続先の db が決まります。
• すべての行で、安定した一意のキーを選択してください。 TanStack DB の getKey を含め、行のキーにはその値を使います。通常は主キー、集計の場合はグループ化する列、同じ行が繰り返される結合の場合は複合キーを使います。すべての rows 更新で無効になる rowId は使わないでください。
• error は最終状態です。 renew() の後も変わりません。新しい確定済みクエリを取得して再度サブスクライブしてください。接続に失敗した場合は、新しいクライアントを使ってください。直接サブスクライブする場合は onStateChange を監視してください。useLiveQuery() はエラーをスローしたりサスペンドしたりせず、status を通じて報告します。
• セッションが変わったら、ページ全体を再読み込みして移動してください。 これにより、Realtime クライアント、TanStack の DbClient、前のユーザーの行が破棄されます。このフックは失敗した更新を 1 秒ごとに再試行するため、サインアウト後に失敗したら更新を停止してください。更新と書き込みのたびに、セッションのユーザーがページを読み込んだユーザーと同じであることを確認してください。そうしないと、前のページからの書き込みが新しいユーザーのアカウントに反映されます。セッションが変わったときを参照してください。

Optimistic Writes

Realtime には書き込みパスがないため、書き込みにはバックエンドの通常の認証が適用されます。各レイヤーでミューテーションを書く前に、references/mutations.md をお読みください。

• 書き込み自体から txid を返してください：.returning({ txid: sql<string>`pg_current_xact_id()::text` }) を使うと、どの Postgres ドライバーでも 1 回の往復で取得できます。バッチ書き込みは 1 つのステートメントで行い、txid が 1 つになるようにしてください。pg_current_xact_id() を select するだけのために db.transaction() を開かないでください。プールされた接続を保持したまま、4 回の往復が発生します。複数のステートメントが本当に必要な書き込みでは、必ず行を変更する書き込みの RETURNING に txid を含めてください。
• awaitTxId() には必ずタイムアウトを渡してください。再接続や更新の間も待機を続け、txid がクエリに到達しない場合は決して完了しません。
• 行を変更しない書き込みには txid がないため、awaitTxId() をスキップし、楽観的な状態を破棄してください。

Server-Side Rendering

Next.js と TanStack Start でサーバー上にライブクエリをレンダリングする前に、references/ssr.md をお読みください。

• クライアントはモジュールレベルではなく useState の初期化子で作成し、initialData をローディング状態なしでライブ行としてレンダリングしてください。また、リクエスト間で TanStack の DbClient を共有しないでください。
• 両側で行の形状を同一にしてください。JSON.stringify は日付を文字列に変換し、int8 と numeric は文字列として届きます。

More Docs

タスク	ページ
ベータ版の対象範囲、リージョン、料金、FAQ、その他のシステム	概要
アーキテクチャと一貫性	仕組み
最初のライブクエリ	クイックスタート
SDK のすべてのエクスポート	TypeScript SDK リファレンス
認証と SSR を備えたフルスタックアプリ	Next.js、TanStack
ライブエージェントワークフロー	Mastra
ORM アダプター	Drizzle、Kysely