Live queries for SQLite / Durable Objects

Extend .live to sqlite (better-sqlite3, Durable Objects' SqlStorage) with
a minimal delta from the pg design: the Bus (reverse index, subscribe +
backfill, lifecycle) is dialect-shared; only the event source differs.

- Event source: pg keeps the shadow-events table + polling loop. sqlite
  has NO shadow table and no polling — the runtime is a synchronous
  single writer, so capture pushes events into the new `Bus.ingest()` in
  the same tick as the mutation, entirely in memory. Cursors are integer
  seqs embedded in the snapshot Cursor shape ({seq+1, seq+1, ∅}), so the
  existing MVCC `visible()` test doubles as the seq comparison and the
  subscribe-time backfill check precisely covers the register race.
- Capture (live/sqlite/capture.ts): opted-in tables (`transformer:
  SqliteLiveCapture.makeTransformer()`) get json_object image columns
  appended to RETURNING — insert→after, delete→before, update→after —
  plus a pre-SELECT for update before-images (sqlite RETURNING can't see
  OLD). Runs via a new optional `Driver.executeSync` in one synchronous
  block with the mutation: no awaits between statements, so single-writer
  atomicity replaces the transaction.
- Extractor portability: accept bare `Param` literals as anchors (sqlite
  binds primitives as `?`, not pg's `CAST($n AS T)`); dialect-aware text
  casts; decoy aliases so extractor CTEs never shadow their own table
  (sqlite resolves a CTE name inside its body — "circular reference").
- live/canonical.ts: shared value-rendering contract; collapses integral
  REALs so `'1.0'`-bound anchors match `'1'` stored INTEGERs (drivers
  bind every JS number as REAL). A parity test hangs on divergence.
- Transport intentionally deferred: `.live()` stays a local AsyncIterable
  on both dialects (already streams over exoeval). Cap'n Web prototype
  learnings (by-ref callbacks vs map recorder, dup()/dispose lifecycle,
  parked-await teardown) are recorded in ISSUES.md #13.

Tested end-to-end on better-sqlite3 (zero timers — dispatch is sync) and
via a fake SqlStorage through DoSqliteDriver; pg suite unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: ryanrasti

src/database.ts

@@ -1,4 +1,4 @@
-import type { ExecuteFn, Driver, QueryResult } from "./drivers/types";
+import { type ExecuteFn, type Driver, isSyncDriver, type QueryResult } from "./drivers/types";
 import type { Fromable, RowType, RowTypeToTsType } from "./builder/query";
 import { QueryBuilder, hydrateRows } from "./builder/query";
 import { deserializeRows } from "./util";
@@ -10,9 +10,10 @@ import { InsertBuilder } from "./builder/insert";
 import { UpdateBuilder } from "./builder/update";
 import { DeleteBuilder } from "./builder/delete";
 import { Bus, type Subscription, type BusOptions } from "./live/bus";
-import { eventsTableSqlStatements } from "./live/events-ddl";
+import { eventsTableSqlStatements } from "./live/pg/events-ddl";
 import { runLiveIteration } from "./live/extractor";
 import { parseSnapshot } from "./live/snapshot";
+import { executeMutationWithCapture, type MutationBuilder, type MutationOp } from "./live/sqlite/capture";
 import type { DialectName } from "./builder/sql";
 
 export type TransactionIsolation = "read committed" | "repeatable read" | "serializable";
@@ -131,10 +132,34 @@ export class Connection<C = undefined> {
   get dialect() { return this.driver.dialect; }
 
   #exec(query: Sql): Promise<QueryResult> {
+    const captured = this.#tryCaptureMutation(query);
+    if (captured) {
+      return Promise.resolve(captured);
+    }
     const compiled = compile(query, { database: this.database });
     return (this.#boundExecute ?? this.driver.execute.bind(this.driver))(compiled);
   }
 
+  // Route mutation builders through the sqlite live capture path when the
+  // bus is running. Pool-backed connections only: inside a transaction the
+  // driver-level sync path would escape the txn's bound execute (and
+  // sqlite live doesn't use transactions anyway).
+  #tryCaptureMutation(query: Sql): QueryResult | undefined {
+    const bus = this.#bus;
+    const driver = this.driver;
+    // startLive() guarantees a SyncDriver whenever a sqlite bus is running;
+    // the guard here is for the type narrowing.
+    if (!bus || this.database.dialect === "postgres" || this.#boundExecute || !isSyncDriver(driver)) {
+      return undefined;
+    }
+    const op: MutationOp | undefined =
+      query instanceof InsertBuilder ? "insert"
+      : query instanceof UpdateBuilder ? "update"
+      : query instanceof DeleteBuilder ? "delete"
+      : undefined;
+    return op && executeMutationWithCapture(op, query as MutationBuilder, this.database, driver, bus);
+  }
+
   // Overload resolution matches top-to-bottom and stops on the first
   // match — list specific builders before the general `Sql` fallback, or
   // every QueryBuilder/Insert/... call resolves as `Sql` → QueryResult.
@@ -254,17 +279,31 @@ export class Connection<C = undefined> {
   // --- Live queries ---
   #bus: Bus | undefined;
 
+  // Pg-only schema setup (the `_typegres_live_events` shadow table). On
+  // sqlite there is nothing to install — capture rides on RETURNING and
+  // events stay in memory (see Bus.ingest).
   async installLiveEvents(): Promise<void> {
+    if (this.database.dialect !== "postgres") {
+      return;
+    }
     for (const stmt of eventsTableSqlStatements(this.database)) {
       await this.driver.execute(compile(stmt, { database: this.database }));
     }
   }
 
+  // `intervalMs` only applies on pg (poll cadence); sqlite dispatch is
+  // synchronous with each mutation.
   async startLive(opts: BusOptions = {}): Promise<void> {
     if (this.#boundExecute) {
       throw new Error("startLive() must be called on a pool-backed Connection, not inside a transaction");
     }
     if (this.#bus) { throw new Error("Live bus already started"); }
+    if (this.database.dialect !== "postgres" && !isSyncDriver(this.driver)) {
+      throw new Error(
+        "sqlite live requires a SyncDriver (SqliteDriver, DoSqliteDriver) — " +
+          "capture must run pre-image read + mutation + dispatch without awaits between them",
+      );
+    }
     this.#bus = new Bus(this, opts);
     await this.#bus.start();
   }
@@ -291,8 +330,13 @@ export class Connection<C = undefined> {
     let currentSub: Subscription | undefined;
     try {
       while (true) {
+        // sqlite: read the seq cursor BEFORE the iteration's queries, so
+        // any event ingested during them is "not yet seen" and trips the
+        // subscribe-time backfill check. pg captures its cursor inside
+        // the iteration's REPEATABLE READ txn instead.
+        const preCursor = this.database.dialect === "postgres" ? undefined : bus.sqliteCursor();
         const { rows, cursor, predicateSet } = await runLiveIteration(this, query);
-        currentSub = bus.subscribe(parseSnapshot(cursor), predicateSet);
+        currentSub = bus.subscribe(cursor !== undefined ? parseSnapshot(cursor) : preCursor!, predicateSet);
         yield rows as any;
         if (currentSub) {
           try {

src/drivers/do.ts

@@ -1,5 +1,5 @@
 import type { CompiledSql } from "../builder/sql";
-import type { Driver, ExecuteFn, QueryResult } from "./types";
+import type { ExecuteFn, QueryResult, SyncDriver } from "./types";
 import { normalizeRow, stripMatchedOuterParens } from "./shared-sqlite";
 
 // Duck-typed Cloudflare SqlStorage — no @cloudflare/workers-types dependency.
@@ -16,15 +16,18 @@ export interface SqlStorageLike {
 //     callers / the dialect must not emit bigint bindings for this backend;
 //   - results are native JS values, normalized to strings for deserialize;
 //   - blobs come back as ArrayBuffer (better-sqlite3 gives Uint8Array/Buffer).
-export class DoSqliteDriver implements Driver {
+export class DoSqliteDriver implements SyncDriver {
   readonly dialect = "sqlite" as const;
 
   constructor(private readonly sql: SqlStorageLike) {}
 
-  execute: ExecuteFn = ({ text, values }: CompiledSql): Promise<QueryResult> => {
+  execute: ExecuteFn = (compiled: CompiledSql): Promise<QueryResult> =>
+    Promise.resolve(this.executeSync(compiled));
+
+  executeSync = ({ text, values }: CompiledSql): QueryResult => {
     const query = stripMatchedOuterParens(text);
     const rows = this.sql.exec(query, ...values).toArray().map(normalizeRow);
-    return Promise.resolve({ rows });
+    return { rows };
   };
 
   runInSingleConnection = <T>(cb: (execute: ExecuteFn) => Promise<T>): Promise<T> =>

src/drivers/sqlite.ts

@@ -1,13 +1,13 @@
 import type { CompiledSql } from "../builder/sql";
 import type { DialectName } from "../builder/sql";
 import type BetterSqlite3 from "better-sqlite3";
-import type { Driver, ExecuteFn, QueryResult } from "./types";
+import type { ExecuteFn, QueryResult, SyncDriver } from "./types";
 import { normalizeRow, stripMatchedOuterParens } from "./shared-sqlite";
 
 // better-sqlite3 adapter. Synchronous under the hood; wrapped in
 // Promise.resolve for the async Driver contract. `better-sqlite3` is an
 // optional peer (see package.json).
-export class SqliteDriver implements Driver {
+export class SqliteDriver implements SyncDriver {
   readonly dialect: DialectName = "sqlite";
 
   static async create(
@@ -22,11 +22,15 @@ export class SqliteDriver implements Driver {
 
   private constructor(private db: BetterSqlite3.Database) {}
 
-  async execute({ text, values }: CompiledSql): Promise<QueryResult> {
+  async execute(compiled: CompiledSql): Promise<QueryResult> {
+    return this.executeSync(compiled);
+  }
+
+  executeSync = ({ text, values }: CompiledSql): QueryResult => {
     // QueryBuilder.bind() wraps statements in `(...)` for subquery splicing;
     // SQLite refuses top-level parenthesized statements — unwrap one matched pair.
     return this.runOne(stripMatchedOuterParens(text), values);
-  }
+  };
 
   private runOne(text: string, values: readonly unknown[]): QueryResult {
     const stmt = this.db.prepare(text);

src/drivers/types.ts

@@ -14,9 +14,26 @@ export type QueryResult = { rows: { [key: string]: string | null }[] };
 // hand the query to its underlying pool/wasm and normalize the result rows.
 export type ExecuteFn = (sql: CompiledSql) => Promise<QueryResult>;
 
+export type ExecuteSyncFn = (sql: CompiledSql) => QueryResult;
+
 export interface Driver {
   readonly dialect: DialectName;
   execute: ExecuteFn;
+  // Synchronous execute for drivers whose engine is sync under the hood
+  // (better-sqlite3, Durable Object SqlStorage). The sqlite live backend
+  // requires it: mutation capture runs pre-image read + write + dispatch
+  // in one synchronous block so no other task can interleave between the
+  // statements — the single-writer substitute for a transaction.
+  executeSync?: ExecuteSyncFn;
   runInSingleConnection<T>(cb: (execute: ExecuteFn) => Promise<T>): Promise<T>;
   close(): Promise<void>;
 }
+
+// A driver that guarantees the synchronous path. Sqlite drivers implement
+// this; code that needs the sync invariant (live capture) takes SyncDriver
+// so the requirement lives in the signature, not a runtime check.
+export interface SyncDriver extends Driver {
+  executeSync: ExecuteSyncFn;
+}
+
+export const isSyncDriver = (d: Driver): d is SyncDriver => d.executeSync !== undefined;

src/index.ts

@@ -11,13 +11,14 @@ export { Table } from "./table";
 export { Relation } from "./relation";
 export { sql, Sql } from "./builder/sql";
 export { QueryBuilder } from "./builder/query";
-export { TypegresLiveEvents } from "./live/events";
+export { TypegresLiveEvents } from "./live/pg/events";
+export { SqliteLiveCapture } from "./live/sqlite/capture";
 export { expose } from "./exoeval/tool";
 export type { ToolFunction } from "./exoeval/tool";
 export { RpcClient, inMemoryChannel, safeStringify } from "./exoeval/rpc";
 export type { RawChannel } from "./exoeval/rpc";
 export type { Config } from "./config";
-export type { Driver, ExecuteFn, QueryResult } from "./drivers/types";
+export type { Driver, SyncDriver, ExecuteFn, ExecuteSyncFn, QueryResult } from "./drivers/types";
 
 import type { Connection } from "./database";
 import { Database } from "./database";

src/live/ISSUES.md

@@ -11,32 +11,30 @@ A query is live-able iff **every table reference has at least one top-level
 Otherwise arbitrary: `JOIN`, `LEFT JOIN`, `GROUP BY`, `HAVING`,
 subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`.
 
-## Open
-
-1. **Public API gap: live mode isn't actually consumable from the package yet.**
-   The npm entrypoint exports `Database`, `db.startLive()`, and `db.live()`,
-   but not the setup pieces users need to make live mode work:
-   - `TypegresLiveEvents.makeTransformer()` to opt tables into event capture
-   - `TypegresLiveEvents.createTableSql()` to create `_typegres_live_events`
-
-   `package.json` also uses explicit export maps, so consumers can't rely on
-   a deep import escape hatch. Result: the live subsystem exists, but from a
-   package user's perspective it's effectively private / unusable.
+The Bus (index, subscribe/backfill, lifecycle) is dialect-shared; only
+the **event source** differs. pg (`pg/`): shadow events table + polling +
+MVCC snapshot cursors. sqlite (`sqlite/`): no shadow table and no polling
+— the runtime is a synchronous single writer (better-sqlite3, a Durable
+Object's SqlStorage), so RETURNING-image capture pushes events into
+`Bus.ingest()` in the same tick as the mutation, entirely in memory;
+cursors are integer seqs embedded in the snapshot `Cursor` shape so the
+MVCC `visible()` test doubles as the seq comparison. The scope rule above
+applies to both dialects.
 
-   Fix options: export `TypegresLiveEvents`; expose a higher-level
-   `enableLive()` helper; or intentionally keep the whole feature internal
-   until the setup surface is ready.
+## Open
 
-2. **`stopLive()` leaves active subscriptions wedged.**
-   Shutting down the bus stops polling, but today it does not drain existing
-   subscriptions or settle their `wait` promises. Any async iterator parked in
-   `await currentSub.wait` can hang forever if `db.stopLive()` is called while
-   it's waiting. The old Bus instance also keeps stale subscription/index state
-   alive until GC.
+1. **Public API gap, narrowed but open.** `TypegresLiveEvents` and
+   `SqliteLiveCapture` are exported from the root entrypoint now, so tables
+   can opt into capture on both backends. Still missing: a friendlier setup
+   surface — `conn.installLiveEvents()` covers the pg DDL but a higher-level
+   `enableLive()` (install + startLive + per-table opt-in in one place)
+   would remove the three-step ceremony.
 
-   `Bus.stop()` should explicitly resolve/reject/unsubscribe outstanding subs,
-   and live iterators should observe shutdown as completion or error rather
-   than hanging.
+2. ~~**`stopLive()` leaves active subscriptions wedged.**~~ Resolved:
+   `Bus.stop()` cancels outstanding subscriptions (rejects parked `wait`s
+   with AbortError; the live generator observes shutdown as clean
+   completion). Covered by the sqlite test "stopLive releases a parked
+   consumer cleanly".
 
 3. **More `.live()` tests.** Five today (insert/update/delete/join/not-started).
    Missing: backfill-from-buffer (subscribe returns `undefined`, caller
@@ -82,21 +80,59 @@ subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`.
    perf footnote: an atomic-swap that diffs old vs. new and only mutates
    leaves that changed would be cheaper for stable predicate sets.
 
-9. **Text canonicalization landmines** — both extractor and events
-   transformer canonicalize values via `col::text`; bus matches by string
-   equality. Symmetric *as long as both sides agree on text form*. Three
-   pg types drift:
-   - **`timestamptz`** — `::text` honors session `TimeZone`; same instant,
-     different text in different sessions.
-   - **`citext`** — case-insensitive at compare time but `::text` preserves
-     case; equal values can produce non-equal text.
-   - **`numeric`** — trailing zeros / scale survive `::text`;
-     `1.0::numeric` and `1::numeric` compare equal but differ as text.
-
-   No live test exercises any of these today; a user query against one of
-   these column types may silently miss matches. Fix range: canonicalizing
-   casts (`lower()` / `extract(epoch from …)`) up to a binary comparator
-   in the bus.
+9. **Text canonicalization landmines** — the matcher replays SQL `=`
+   *outside* SQL: it lifts predicate values out of the query and compares
+   canonical text in JS. That loses every normalization `=` would have
+   applied (operator resolution, pg implicit casts, sqlite affinity /
+   numeric cross-class comparison), so the invariant the whole scheme
+   rests on is: **the canonical rendering must be constant on each SQL
+   equality class** — two `=`-equal values that render differently are a
+   silently *missed wakeup* (always a false negative; conflating unequal
+   values merely causes a harmless spurious rerun).
+
+   `::text` / `CAST(x AS text)` satisfies this for the common types
+   (int, text, uuid, bool, float8) when both sides render through the
+   same type's output function. Known drift:
+
+   **pg** (pre-existing, unfixed):
+   - **`numeric`** — scale survives `::text`: `1 = 1.0 = 1.00` but
+     `'1'` / `'1.0'` / `'1.00'` (verified). `trim_scale()` (pg13+)
+     normalizes. This is the most likely real-world miss.
+   - **`timestamptz`** — `::text` honors session `TimeZone`/`DateStyle`;
+     event images render in the *writer's* session, extractor values in
+     the live connection's session — differing GUCs break matching.
+     Same-pool same-settings holds in practice but is undocumented.
+   - **`citext`** / non-binary collations — equal at compare time,
+     case-preserving as text.
+
+   **sqlite** (as of the sqlite backend):
+   - **Fixed**: drivers bind every JS number as REAL while
+     integer-affinity columns store INTEGER (`'1.0'` vs `'1'`);
+     `canonical.ts` collapses integral REALs before the text cast.
+     Root cause: sqlite operators live on root `Any`, which binds bare
+     `?` params (`sql-value.ts` — `CAST(? AS any)` would be wrong), so
+     unlike pg's `TypedParam` nothing re-normalizes the binding class.
+     A parity test hangs-on-timeout if the render sites ever diverge.
+   - **Open (contrived)**: TEXT-affinity column vs number param —
+     SQL applies TEXT affinity to REAL `1.0` → matches stored `'1.0'`,
+     but the collapsed anchor renders `'1'` → miss. Reachable because
+     root-`Any.eq` accepts any primitive.
+   - **Harmless**: no-affinity columns distinguish integer `1` from
+     text `'1'`; canonical text conflates them → spurious rerun only.
+
+   Fix directions (complementary, roughly in order of leverage):
+   1. **Typed casts in the sqlite operator codegen** — per-type arg slots
+      so `eq` emits `CAST(? AS INTEGER)` etc., mimicking compare-time
+      affinity at the binding layer. Fixes the anchor at the source
+      (including the TEXT-affinity edge) and would let `canonical.ts`'s
+      CASE collapse be deleted. Correct independent of live.
+   2. **Per-type canonical-render hook** on the type classes (all three
+      render sites hold typed values): `Numeric → trim_scale(x)::text`,
+      `timestamptz → UTC-normalized`, sqlite classes → their affinity
+      semantics (or JS-side `String()` canonicalization, which collapses
+      `1.0`/`1` for free). Subsumes `canonical.ts` and fixes pg numeric.
+   3. Status quo: `canonical.ts`'s dialect-level rendering — works for
+      the common types, locked by tests, ad-hoc.
 
 10. **Unsupported predicate forms produce a generic "unrooted" error**
    instead of a clear "this predicate form isn't supported" message —
@@ -135,7 +171,10 @@ subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`.
    correctly qualified.
 
 12. **WAL / logical-replication ingestion as an alternative to the shadow
-   table** — today every mutation gains a `_typegres_live_events` insert
+   table** (pg; the sqlite backend already has no shadow table — its
+   analog of this gap is that RETURNING-image capture also only sees
+   builder-routed writes, where per-table triggers would catch raw SQL
+   too) — today every pg mutation gains a `_typegres_live_events` insert
    in the same statement. That captures only writes that go through
    typegres builders, costs an extra row per mutation, and grows a table
    the user has to manage. A WAL-backed mode (logical decoding via a
@@ -144,3 +183,28 @@ subqueries — fine. The extractor only inspects top-level `WHERE`/`ON`.
    replication permissions, per-table `REPLICA IDENTITY` config, slot
    monitoring (a stalled consumer holds WAL forever). Worth offering as
    a swappable backend for clients who prefer it.
+
+13. **Cap'n Web streaming surface (deferred; learnings recorded).**
+   `.live()` is a local AsyncIterable on both dialects (already
+   wire-consumable via exoeval's streaming); a capnweb push surface —
+   `subscribe(conn, cb)` driving the generator with `await cb(rows)`
+   backpressure — was prototyped and works, deferred to keep the live
+   core minimal. What the prototype established, so the next attempt
+   starts from knowledge:
+   - capnweb's `map` recorder serializes any captured plain function as a
+     nested record-replay closure (async ones are rejected outright:
+     "RPC closures cannot be async functions"). Passing a callback **by
+     reference** requires a plain RPC call — build the refined query
+     capability via `doRpc`, then call `subscribe` directly on the stub.
+   - An argument stub is auto-released when the call frame returns
+     ("RpcImportHook was already disposed" on the first push). The callee
+     must `dup()` the callback on entry and dispose the retained handle
+     on unsubscribe — which is exactly the pin/unpin-a-Durable-Object
+     lifecycle, and what the harness leak guard verifies.
+   - A driver loop that always has a `next()` in flight parks the live
+     generator at `await sub.wait` — a non-yield suspension `.return()`
+     cannot interrupt. `Connection.live()` needs an AbortSignal (or
+     equivalent) that `cancel()`s the in-flight Subscription so
+     unsubscribe can wake a parked consumer.
+   - Client-side, capnweb invokes an exported callback with args as
+     RpcPromises — the client cb must resolve them.