Live Queries

live query は、plasma が状態変化を UI に送り込む方法です。一度きりの select を書くのと同じ方法でクエリを書き、.live()（React では useLiveQuery）を追加すると、基盤となる行が変化するたびに再実行 / 再 diff するサブスクリプションになります。

基本の形

const live = db.select().from(todos).where(eq(todos.done, 0)).live()
const unsub = live.subscribe((rows) => {
  console.log("open todos:", rows.length)
})
// 後で
unsub()

• .live() は LiveQuery<T> ハンドルを返します。
• .subscribe(cb) はスナップショットのサブスクライバーを登録します。ウィンドウが変化するたびに cb(rows) が発火します。
• 返される関数はサブスクリプションを解除します。

React では:

import { useLiveQuery } from "@sh1n4ps/plasma-react"

function OpenTodos() {
  const rows = useLiveQuery(
    () => plasma.db.select().from(todos).where(eq(todos.done, 0)),
    [],
  )
  return <ul>{rows.map((t) => <li key={t.id}>{t.title}</li>)}</ul>
}

hook は現在のスナップショットを返し、それが変化するたびに React が再レンダリングします。

subscribeDelta — O(delta) の diff

.subscribe は何かが変化するたびに完全な行リストを渡します。10,000 行の table で 1 件の挿入があると、React のリデューサーが diff しなければならない 10,001 要素の配列になります。

.subscribeDelta は変化した分だけを正確に渡します:

const live = db.select().from(todos).live()
const unsub = live.subscribeDelta?.((delta) => {
  console.log("added:", delta.added)
  console.log("removed:", delta.removed)
  console.log("changed:", delta.changed)
})

形状:

interface RowDelta<T> {
  readonly added: readonly T[]
  readonly removed: readonly T[]
  readonly changed: readonly T[]
}

• added — ウィンドウに入った行
• removed — ウィンドウから出た行
• changed — 残ったが投影された値が変化した行

?. ガードは、subscribeDelta がインターフェイス上でオプショナルな LiveQuery<T>['subscribeDelta']? だからです — 再計算なしに diff を生成できないバックエンド（例: サーバー側のフォールバック）はこれを省略できます。クライアントの IVM 対象クエリは常にこれを公開します。

ヒント

subscribeDelta は初回配信時、現在のウィンドウ全体を added として報告します — ただしウィンドウが空（lastSnapshot.length === 0）の場合は初回コールバックがスキップされるため、「delta があれば作業がある」という consumer が null チェックをする必要はありません。

どちらを使うか

状況	使うもの
1000 行未満のリストのレンダリング	subscribe（React が十分安価に調停する）
行ごとに重いレンダリングを伴う大きなリストのレンダリング	subscribeDelta（手動で DOM をパッチ / virtualiser に供給）
RxJS / Solid signal / Preact signal への供給	subscribeDelta（形状がこれらのプリミティブに合う）
状態変化の監査（「ユーザーが今何をしたか?」）	subscribeDelta（形状がタッチされた行を正確にリストする）

whenReady — 決定的なセットアップ待機

IVM エンジンは非同期にセットアップします（IDB からシード行を読む）。subscribe(cb) は cb が初回スナップショットを発火する前に返ります。初回スナップショットを決定的に観測したいコード — テスト、特定のビューが準備できたことで起動する遷移 — には、初回スナップショットが配信されたときに解決する Promise を whenReady() が返します:

const live = db.select().from(todos).live()
live.subscribe((rows) => setRows(rows))
await live.whenReady?.()
// setRows が少なくとも 1 回発火している

whenReady はインターフェイス上でオプショナル（whenReady?()）です — subscribeDelta と同じ理由です。

IVM 適格性ゲート

すべてのクエリが IVM の扱いを受けるわけではありません。classifyIvm(ast) は 3 つの種類のいずれかを返します:

• "select" — 単一または inner/left join されたソースにわたる WHERE / ORDER BY / LIMIT / OFFSET。IvmLiveQuery が処理します。mutation ごとのコスト: O(delta)。
• "aggregate" — GROUP BY / HAVING / count/sum/avg/max/min。IvmAggregateLiveQuery が処理します。join なしのバリアントは対象を絞った per-group 再フォールドを行い、groupBy 付きの join は完全な再構築にフォールバックします。
• "none" — RIGHT / FULL join、fromSubquery、またはまだサポートされていない AST 形状。クエリ全体を再実行するパス（engine.ts の runAndDeliver）にフォールバックします — 正しいが O(total)。

クエリがどの分岐を取ったかを検査できます:

import { classifyIvm } from "@sh1n4ps/plasma-client"

const kind = classifyIvm(db.select().from(todos).where(eq(todos.done, 0)).$ast)
// "select"

subscribe のセマンティクス

• コールバックは subscribe 時に現在のスナップショットで 同期的に 発火します。初回配信を決定的に待ちたい場合は、代わりに whenReady() を使ってください。
• ウィンドウが変化するたびに再発火します。同一の重複スナップショットは subscribe パスでは抑制されません — リアクティブハブは mutation バッチごとに 1 回通知し、クエリが再発火します。
• 同じ LiveQuery 上の複数のサブスクライバーは、refcount を介して 1 つの内部リアクティブ計算を共有します — 同じクエリ上のスナップショットサブスクライバーと delta サブスクライバーは、それぞれ別々のインデックスを立ち上げることはありません。

空 delta の抑制

エンジンは空の delta（{added:[], removed:[], changed:[]}）を subscribeDelta サブスクライバーに配信する前に抑制するため、サブスクライバーはすべてのサブスクリプションで push-then-pull のリフレッシュサイクルから来る { added:[], removed:[], changed:[] } のノイズをフィルタする必要がありません。

集約クエリの詳細

IvmAggregateLiveQuery を介した集約クエリは、join なしの .groupBy(col) 形状を per-group の対象を絞った再フォールドで処理します:

• 1 行の変化 → 影響を受けるグループ（最大 2 つ — 行の変更前グループと変更後グループ）のみが再フォールドされます。
• delta サブスクライバーは、グループタプルでキーが付けられた {added, removed, changed} を受け取ります。
• スナップショット配信では、完全な行配列を再構築するために依然としてすべてのグループを走査します（それが subscribe cb の契約です）。

スナップショットパス自体の mutation ごとの完全な O(1) は v1.1 に予定されています（Phase 2.1 SnapshotPatch — Roadmap を参照）。

よくあるパターン

「最近の」フィード向けの sort + limit

const recent = useLiveQuery(
  () => plasma.db.select().from(todos).orderBy(desc(todos.updatedAt)).limit(20),
  [],
)

IVM エンジンは top-20 のウィンドウのみを維持します。そのウィンドウの外への新しい挿入は再レンダリングのコストになりません。

subscribeDelta を使った join ビュー

const messagesWithAuthor = useLiveQuery(
  () => plasma.db
    .select()
    .from(messages)
    .innerJoin(users, eq(messages.userId, users.id))
    .where(eq(messages.roomId, "general")),
  [],
)

返される行は { messages: { ... }, users: { ... } }（join されたタプル）の形状です。delta の識別子は完全な (messages.id, users.id) タプルにキーが付くため、著者名の変更はメッセージごとに changed: 1 を発火し、完全な再マウントにはなりません。

集約 count

const openCount = useLiveQuery(
  () => plasma.db
    .select({ n: count() })
    .from(todos)
    .where(eq(todos.done, 0)),
  [],
)
// openCount → [{ n: 42 }]

次に読むべきもの

• Concepts / Push, Pull, Rebase — live query の通知を引き起こす sync ループ
• CRDT columns — 自動マージされる状態に対する live query