From 0f61310f1904370b26c87bbfdb56a4999a9301a7 Mon Sep 17 00:00:00 2001 From: Ryan Rasti Date: Wed, 22 Jul 2026 16:32:59 -0700 Subject: [PATCH] .live() returns a LiveQuery: iterate locally or .observe() to push MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Collapses the three live() overloads to one signature — live(conn?) returns a LiveQuery, which is an AsyncIterable (iterate `for await` locally) and exposes @expose'd .observe(observer) → LiveSubscription (the push form, which crosses capnweb; the observer's callbacks cross by reference). The body is just asObservable(conn.live(this)). This also removes the Connection|LiveObserver union off live(), which combined with LiveObserver's now-`| undefined` optional fields lets the observer schema be a plain z.object (colocated in observer.ts) instead of a hand-rolled z.custom predicate — the union and exactOptional mismatch were what broke the @expose decorator's inference before. The callback field type is `(...args: unknown[]) => unknown` so the schema stays assignable to every LiveObserver field (incl. the zero-arg onReturn) under the generic method param. Co-Authored-By: Claude Fable 5 --- src/builder/query.ts | 18 +++- src/index.ts | 2 + src/live/exoeval-live.test.ts | 81 +++++++++++++++ src/live/observer.ts | 172 ++++++++++++++++++++++++++++++++ src/live/sqlite/db-live.test.ts | 36 ++++++- 5 files changed, 303 insertions(+), 6 deletions(-) create mode 100644 src/live/exoeval-live.test.ts create mode 100644 src/live/observer.ts diff --git a/src/builder/query.ts b/src/builder/query.ts index a6f5b0de..92a84889 100644 --- a/src/builder/query.ts +++ b/src/builder/query.ts @@ -16,6 +16,7 @@ import { fn, expose, copyToolFields } from "../exoeval/tool"; import { isTableClass, TableBase } from "../table"; import z from "zod"; import { Values } from "./values"; +import { asObservable, type LiveQuery } from "../live/observer"; // Optional Connection — the execute-family default: omitted, terminators // fall back to the builder's Database.defaultConnection. @@ -411,12 +412,19 @@ export class QueryBuilder< return (conn ?? this.opts.database.defaultConnection).execute(this); } - // Streaming terminator. Mirrors `execute` but yields the rowset on - // every committed mutation that touches one of the live-tagged - // tables this query reads from. Caller iterates with `for await`. + // Streaming terminator. Mirrors `execute` but re-yields the rowset on + // every committed mutation that touches one of the live-tagged tables + // this query reads from. Returns a LiveQuery — two consumption forms: + // - iterator: `for await (const rows of q.live())` locally (not RPC); + // - observer: `q.live().observe({ onNext })` pushes each rowset and + // returns a LiveSubscription. Over capnweb the observer's callbacks + // cross by reference (byRef), so a remote client can subscribe to a + // query it authored itself. @expose(zConn) - live(conn?: Connection): AsyncIterable[]> { - return (conn ?? this.opts.database.defaultConnection).live(this) as AsyncIterable[]>; + live(conn?: Connection): LiveQuery[]> { + return asObservable( + (conn ?? this.opts.database.defaultConnection).live(this) as AsyncIterable[]>, + ); } @expose(zConn) diff --git a/src/index.ts b/src/index.ts index cf5253af..8b5a6651 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,6 +11,8 @@ export { Table } from "./table"; export { Relation } from "./relation"; export { sql, Sql } from "./builder/sql"; export { QueryBuilder } from "./builder/query"; +export { LiveSubscription, LiveQuery } from "./live/observer"; +export type { LiveObserver } from "./live/observer"; export { TypegresLiveEvents } from "./live/pg/events"; export { ensurePgLiveEventsTable } from "./live/pg/events-ddl"; export { expose } from "./exoeval/tool"; diff --git a/src/live/exoeval-live.test.ts b/src/live/exoeval-live.test.ts new file mode 100644 index 00000000..e15052b8 --- /dev/null +++ b/src/live/exoeval-live.test.ts @@ -0,0 +1,81 @@ +// `.live()` over the EXOEVAL RPC wire (the path site/ uses) — as opposed +// to the capnweb shim. The client ships the closure as JS source, the +// server re-evaluates it under exoEval, and a streaming (async-iterable) +// return is drained chunk-by-chunk over inMemoryChannel. +// +// The load-bearing detail this guards: `.live()` returns a LiveQuery class +// instance, and inMemoryChannel detects a stream via +// `Symbol.asyncIterator in result`. That walks the prototype chain, so a +// LiveQuery (with `[Symbol.asyncIterator]` on its prototype) is streamed +// exactly like a bare async iterable — each *rowset* is serialized, never +// the LiveQuery itself. If LiveQuery ever lost that prototype method, the +// channel would fall through to `safeStringify(result)` and throw +// "cannot serialize LiveQuery instance". (The equivalent coverage in +// site/src/demo/demo.test.ts is not part of the core suite.) + +import { test, expect, beforeAll, afterAll } from "vitest"; +import { Database, type Connection } from "../database"; +import { SqliteDriver } from "../drivers/sqlite"; +import { sql } from "../builder/sql"; +import { Integer, Text } from "../types/sqlite"; +import { expose } from "../exoeval/tool"; +import { RpcClient, inMemoryChannel } from "../exoeval/rpc"; + +const db = new Database({ dialect: "sqlite" }); + +class Notes extends db.Table("notes", { live: true }) { + @expose() id = Integer.column({ nonNull: true }); + @expose() user_id = Integer.column({ nonNull: true }); + @expose() body = Text.column({ nonNull: true }); +} + +class Api { + @expose() conn: Connection; + constructor(conn: Connection) { + this.conn = conn; + } + @expose() notes() { + return Notes.from(); + } +} + +let conn: Connection; +beforeAll(async () => { + conn = db.attach(await SqliteDriver.create(":memory:")); + await conn.execute( + sql`CREATE TABLE notes (id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL, body TEXT NOT NULL)`, + ); +}); +afterAll(async () => { + await conn.close(); +}); + +test("exoeval: a client-authored .live() streams the rowset and re-yields on commit", async () => { + await Notes.insert({ id: 1, user_id: 1, body: "hello" }).execute(conn); + const rpc = new RpcClient(inMemoryChannel(new Api(conn))); + + // The site's shape: the closure returns `.live(api.conn)` (iterator form), + // shipped as source and re-evaluated server-side; the client iterates the + // streamed rowsets. + const stream = rpc.run((api) => + api + .notes() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ id: notes.id, body: notes.body })) + .live(api.conn), + ); + const it = (stream as AsyncIterable<{ id: number; body: string }[]>)[Symbol.asyncIterator](); + + // Initial snapshot over the wire. + const first = await it.next(); + expect(first.done).toBe(false); + expect(first.value).toEqual([{ id: 1, body: "hello" }]); + + // A matching commit re-yields the full rowset. + await Notes.insert({ id: 2, user_id: 1, body: "world" }).execute(conn); + const second = await it.next(); + expect(second.done).toBe(false); + expect(second.value!.map((r: { body: string }) => r.body).sort()).toEqual(["hello", "world"]); + + await it.return?.(); +}); diff --git a/src/live/observer.ts b/src/live/observer.ts new file mode 100644 index 00000000..ca78c50b --- /dev/null +++ b/src/live/observer.ts @@ -0,0 +1,172 @@ +import { z } from "zod"; +import { expose } from "../exoeval/tool"; + +// `.live()` returns a LiveQuery: iterate it locally (`for await`), or call +// `.observe(observer)` to push. AsyncIterables can't cross an RPC boundary, +// but `.observe` is an @expose'd method and functions cross by reference — a +// remote client does `.live().observe({ onNext: byRef(cb) })`, the callback +// arrives as a stub, and every push is a bidirectional RPC back to it. + +// onThrow/onReturn are `?: fn | undefined` (not just `?: fn`) so the zod +// z.object below — whose `.optional()` yields `fn | undefined` — satisfies +// this type under exactOptionalPropertyTypes. +export type LiveObserver = { + /** + * Called with the full rowset on subscribe and after every committed + * mutation that changes it. The push is awaited before the next + * iteration runs: a slow consumer gets backpressure (intervening + * commits coalesce into the next rowset), and a dead one — e.g. a + * disconnected RPC peer — rejects, tearing the subscription down + * instead of leaking a watcher. + */ + onNext: (rows: Rows) => unknown; + onThrow?: ((error: unknown) => unknown) | undefined; + /** Called if the live stream ends without error (bus stopped). */ + onReturn?: (() => unknown) | undefined; +}; + +// Validates observer shape without wrapping the callbacks — they may be RPC +// stubs whose identity (and .dup()) must survive. z.custom is the field +// check because zod v4's `z.function()` is a function factory, not a field +// schema; z.object only rebuilds the container, so the stub refs pass through. +// Typed `(...args: unknown[]) => unknown` so the schema's type is assignable +// to every LiveObserver field — onNext `(rows: Rows) => …`, onThrow +// `(error) => …`, and the zero-arg onReturn `() => …` — which is what lets the +// @expose decorator reconcile the schema with the generic method param. +const zCallback = z.custom<(...args: unknown[]) => unknown>((v) => typeof v === "function"); +export const zLiveObserver = z.object({ + onNext: zCallback, + onThrow: zCallback.optional(), + onReturn: zCallback.optional(), +}); + +// A callback that arrived over RPC is only guaranteed alive for the +// duration of the call that delivered it (its args payload is disposed on +// return) — but a subscription outlives that call by design. Stubs expose +// dup() to extend their lifetime; retain via dup() when present, and +// release when the subscription ends. Plain local functions pass through. +type Retained = { fn: F; release: () => void }; +const retain = unknown>(fn: F): Retained => { + const dupable = fn as { dup?: () => F }; + if (typeof dupable.dup === "function") { + const duped = dupable.dup(); + return { + fn: duped, + release: () => { + (duped as Partial)[Symbol.dispose]?.(); + }, + }; + } + return { fn, release: () => {} }; +}; + +/** + * A live subscription: drives the underlying iterator and pushes each + * committed rowset into the observer. Construct via `LiveSubscription.start` + * (the constructor is private); iterating stops on `unsubscribe()` / + * disposal, when the stream ends, or when a push throws. Generic in `Rows` + * so the retained callbacks stay typed on the instance — `Rows` appears only + * in the private machinery, not in the public surface. + */ +export class LiveSubscription { + readonly #iter: AsyncIterator; + readonly #onNext: Retained<(rows: Rows) => unknown>; + readonly #onThrow: Retained<(error: unknown) => unknown> | undefined; + readonly #onReturn: Retained<() => unknown> | undefined; + #stopped = false; + + private constructor( + iter: AsyncIterator, + onNext: Retained<(rows: Rows) => unknown>, + onThrow: Retained<(error: unknown) => unknown> | undefined, + onReturn: Retained<() => unknown> | undefined, + ) { + this.#iter = iter; + this.#onNext = onNext; + this.#onThrow = onThrow; + this.#onReturn = onReturn; + } + + static start(iterable: AsyncIterable, observer: LiveObserver): LiveSubscription { + const sub = new LiveSubscription( + iterable[Symbol.asyncIterator](), + retain(observer.onNext), + observer.onThrow ? retain(observer.onThrow) : undefined, + observer.onReturn ? retain(observer.onReturn) : undefined, + ); + void sub.#run(); + return sub; + } + + #stop(): void { + if (!this.#stopped) { + this.#stopped = true; + // return() may reject on an already-closed or transport-broken + // iterator; the subscription is ending regardless, so swallow it + // rather than surface an unhandled rejection during teardown. + void Promise.resolve(this.#iter.return?.()).catch(() => {}); + } + } + + async #run(): Promise { + try { + while (true) { + const r = await this.#iter.next(); + if (this.#stopped || r.done) { + break; + } + await this.#onNext.fn(r.value); + } + if (!this.#stopped) { + await this.#onReturn?.fn(); + } + } catch (err) { + this.#stop(); + try { + await this.#onThrow?.fn(err); + } catch { + // Peer already gone; nothing left to notify. + } + } finally { + this.#onNext.release(); + this.#onThrow?.release(); + this.#onReturn?.release(); + } + } + + @expose() + unsubscribe(): void { + this.#stop(); + } + + [Symbol.dispose](): void { + this.#stop(); + } +} + +/** + * The result of `.live()`: an AsyncIterable (iterate locally) that also + * exposes `.observe(observer)` (push form — crosses RPC). A class, not a + * plain object, so the shim wraps it as a capability with `observe` + * @expose-visible; `[Symbol.asyncIterator]` is symbol-keyed, so it stays + * local and never crosses the wire. + */ +export class LiveQuery { + readonly #iterable: AsyncIterable; + + constructor(iterable: AsyncIterable) { + this.#iterable = iterable; + } + + [Symbol.asyncIterator](): AsyncIterator { + return this.#iterable[Symbol.asyncIterator](); + } + + @expose(zLiveObserver) + observe(observer: LiveObserver): LiveSubscription { + return LiveSubscription.start(this.#iterable, observer); + } +} + +export const asObservable = (iterable: AsyncIterable): LiveQuery => + new LiveQuery(iterable); diff --git a/src/live/sqlite/db-live.test.ts b/src/live/sqlite/db-live.test.ts index 7c9b4d83..220cef53 100644 --- a/src/live/sqlite/db-live.test.ts +++ b/src/live/sqlite/db-live.test.ts @@ -1,4 +1,4 @@ -import { test, expect, beforeAll, afterAll, afterEach } from "vitest"; +import { test, expect, vi, beforeAll, afterAll, afterEach } from "vitest"; import { Database, type Connection } from "../../database"; import { SqliteDriver } from "../../drivers/sqlite"; import { DoSqliteDriver } from "../../drivers/do"; @@ -80,6 +80,40 @@ test("yields current rows then re-yields on a matching commit (insert path)", as await iter.return?.(); }, 10_000); +test("observer form: onNext pushes on each commit; unsubscribe stops it", async () => { + const Notes = await makeNotesTable(); + + const seen: { body: string }[][] = []; + const sub = Notes.from() + .where(({ notes }) => notes.user_id.eq(1)) + .select(({ notes }) => ({ body: notes.body })) + .live() + .observe({ onNext: (rows) => void seen.push(rows) }); + + // Snapshot on subscribe, then a push per matching commit. + await vi.waitFor(() => expect(seen.length).toBeGreaterThanOrEqual(1)); + expect(seen[0]).toEqual([]); + + await Notes.insert({ id: 1, user_id: 1, body: "a" }).execute(conn); + await vi.waitFor(() => expect(seen.at(-1)).toEqual([{ body: "a" }])); + + // A non-matching commit does not push. + const beforeIrrelevant = seen.length; + await Notes.insert({ id: 2, user_id: 99, body: "nope" }).execute(conn); + await Notes.insert({ id: 3, user_id: 1, body: "b" }).execute(conn); + await vi.waitFor(() => + expect(seen.at(-1)).toEqual([{ body: "a" }, { body: "b" }]), + ); + expect(seen.length).toBeGreaterThan(beforeIrrelevant); + + // unsubscribe() stops further pushes. + sub.unsubscribe(); + const afterStop = seen.length; + await Notes.insert({ id: 4, user_id: 1, body: "c" }).execute(conn); + await new Promise((r) => setTimeout(r, 50)); + expect(seen.length).toBe(afterStop); +}, 10_000); + test("UPDATE on a matching row re-yields with new values", async () => { const Notes = await makeNotesTable(); await Notes.insert({ id: 1, user_id: 1, body: "first" }).execute(conn);