.live() observer form: push each rowset into a callback

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<any>): AsyncIterable<RowTypeToTsType<O>[]> {
-    return (conn ?? this.opts.database.defaultConnection).live(this) as AsyncIterable<RowTypeToTsType<O>[]>;
+  live(conn?: Connection<any>): LiveQuery<RowTypeToTsType<O>[]> {
+    return asObservable(
+      (conn ?? this.opts.database.defaultConnection).live(this) as AsyncIterable<RowTypeToTsType<O>[]>,
+    );
   }
 
   @expose(zConn)

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";

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<Api>(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?.();
+});

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<Rows> = {
+  /**
+   * 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<F> = { fn: F; release: () => void };
+const retain = <F extends (...args: never[]) => unknown>(fn: F): Retained<F> => {
+  const dupable = fn as { dup?: () => F };
+  if (typeof dupable.dup === "function") {
+    const duped = dupable.dup();
+    return {
+      fn: duped,
+      release: () => {
+        (duped as Partial<Disposable>)[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<Rows = unknown> {
+  readonly #iter: AsyncIterator<Rows>;
+  readonly #onNext: Retained<(rows: Rows) => unknown>;
+  readonly #onThrow: Retained<(error: unknown) => unknown> | undefined;
+  readonly #onReturn: Retained<() => unknown> | undefined;
+  #stopped = false;
+
+  private constructor(
+    iter: AsyncIterator<Rows>,
+    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<Rows>(iterable: AsyncIterable<Rows>, observer: LiveObserver<Rows>): LiveSubscription<Rows> {
+    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<void> {
+    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<Rows> {
+  readonly #iterable: AsyncIterable<Rows>;
+
+  constructor(iterable: AsyncIterable<Rows>) {
+    this.#iterable = iterable;
+  }
+
+  [Symbol.asyncIterator](): AsyncIterator<Rows> {
+    return this.#iterable[Symbol.asyncIterator]();
+  }
+
+  @expose(zLiveObserver)
+  observe(observer: LiveObserver<Rows>): LiveSubscription<Rows> {
+    return LiveSubscription.start(this.#iterable, observer);
+  }
+}
+
+export const asObservable = <Rows>(iterable: AsyncIterable<Rows>): LiveQuery<Rows> =>
+  new LiveQuery(iterable);

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