Live queries for SQLite / Durable Objects

package.json

@@ -58,8 +58,8 @@
   ],
   "peerDependencies": {
     "@electric-sql/pglite": "^0.4.4",
-    "pg": "^8.20.0",
-    "better-sqlite3": "^12.11.1"
+    "better-sqlite3": "^12.11.1",
+    "pg": "^8.20.0"
   },
   "peerDependenciesMeta": {
     "@electric-sql/pglite": {
@@ -78,14 +78,16 @@
   "scripts": {
     "build": "tsdown",
     "codegen": "node --experimental-strip-types src/types/postgres/emit.ts && node --experimental-strip-types src/types/sqlite/emit.ts",
-    "codegen:check": "tmp=$(mktemp -d) && node --experimental-strip-types src/types/postgres/emit.ts --out-dir \"$tmp/pg\" && { diff -r \"$tmp/pg\" src/types/postgres/generated || { echo 'src/types/postgres/generated is stale — run `npm run codegen` and commit.' >&2; rm -rf \"$tmp\"; exit 1; }; } && node --experimental-strip-types src/types/sqlite/emit.ts --out-dir \"$tmp/sqlite\" && { diff -r \"$tmp/sqlite/generated\" src/types/sqlite/generated || { echo 'src/types/sqlite/generated is stale — run `npm run codegen` and commit.' >&2; rm -rf \"$tmp\"; exit 1; }; } && rm -rf \"$tmp\"",
+    "codegen:check": "tmp=$(mktemp -d) && node --experimental-strip-types src/types/postgres/emit.ts --out-dir \"$tmp/pg\" && { diff -r \"$tmp/pg\" src/types/postgres/generated || { echo 'src/types/postgres/generated is stale - run `npm run codegen` and commit.' >&2; rm -rf \"$tmp\"; exit 1; }; } && node --experimental-strip-types src/types/sqlite/emit.ts --out-dir \"$tmp/sqlite\" && { diff -r \"$tmp/sqlite/generated\" src/types/sqlite/generated || { echo 'src/types/sqlite/generated is stale - run `npm run codegen` and commit.' >&2; rm -rf \"$tmp\"; exit 1; }; } && rm -rf \"$tmp\"",
     "lint": "eslint src",
     "typecheck": "tsgo --noEmit",
     "format": "prettier --write src",
     "test": "vitest run",
-    "check": "npm run lint && npm run typecheck && npm run test && npm --prefix examples/basic run check"
+    "check": "npm run lint && npm run typecheck && npm run test && npm --prefix examples/basic run check",
+    "test:do": "vitest run --project workerd"
   },
   "devDependencies": {
+    "@cloudflare/vitest-pool-workers": "^0.18.6",
     "@electric-sql/pglite": "^0.5.4",
     "@eslint/js": "^10.0.1",
     "@standard-schema/spec": "^1.1.0",

site/src/demo/runtime.ts

@@ -2,7 +2,7 @@
 // Uses top-level await so schema files can import a ready `db` and
 // define tables at module-eval time.
 
-import { typegres } from "typegres";
+import { ensurePgLiveEventsTable, typegres } from "typegres";
 import { runMigrations, runSeed } from "./seed";
 import type { UserRoot } from "./server/api";
 
@@ -11,9 +11,8 @@ export const { db, conn } = await typegres<UserRoot>({ type: "pglite" });
 await runMigrations(conn);
 await runSeed(conn);
 
-// `installLiveEvents` is a one-time DDL — production callers run it as
+// `ensurePgLiveEventsTable` is a one-time DDL — production callers run it as
 // part of their migrations. The demo's storage doesn't survive page
-// reloads, so we run it on every boot. `startLive` then assumes the
-// events table exists and only spins up the polling bus.
-await conn.installLiveEvents();
-await conn.startLive();
+// reloads, so we run it on every boot. The live engine itself is wired
+// at attach and its poller starts lazily on first .live() use.
+await ensurePgLiveEventsTable(conn);

site/src/demo/schema/customers.ts

@@ -1,9 +1,9 @@
-import { TypegresLiveEvents, expose } from "typegres";
+import { expose } from "typegres";
 import { Int8, Text } from "typegres/postgres";
 import { db } from "../runtime";
 import { Orders } from "./orders";
 import { Organizations } from "./organizations";
-export class Customers extends db.Table("customers", { transformer: TypegresLiveEvents.makeTransformer() }) {
+export class Customers extends db.Table("customers", { live: true }) {
   // @generated-start
   @expose() id = (Int8<1>).column({ nonNull: true, generated: true });
   @expose() organization_id = (Int8<1>).column({ nonNull: true });

site/src/demo/schema/inventory_positions.ts

@@ -1,11 +1,11 @@
-import { Connection, TypegresLiveEvents, sql, expose } from "typegres";
+import { Connection, sql, expose } from "typegres";
 import { Int8, Text } from "typegres/postgres";
 import { z } from "zod";
 import { db } from "../runtime";
 import { Locations } from "./locations";
 import { Organizations } from "./organizations";
 import { OrderLines } from "./order_lines";
-export class InventoryPositions extends db.Table("inventory_positions", { transformer: TypegresLiveEvents.makeTransformer() }) {
+export class InventoryPositions extends db.Table("inventory_positions", { live: true }) {
   // @generated-start
   @expose() id = (Int8<1>).column({ nonNull: true, generated: true });
   @expose() organization_id = (Int8<1>).column({ nonNull: true });

site/src/demo/schema/locations.ts

@@ -1,10 +1,10 @@
-import { TypegresLiveEvents, expose } from "typegres";
+import { expose } from "typegres";
 import { Int8, Text } from "typegres/postgres";
 import { db } from "../runtime";
 import { InventoryPositions } from "./inventory_positions";
 import { Organizations } from "./organizations";
 
-export class Locations extends db.Table("locations", { transformer: TypegresLiveEvents.makeTransformer() }) {
+export class Locations extends db.Table("locations", { live: true }) {
   // @generated-start
   @expose() id = (Int8<1>).column({ nonNull: true, generated: true });
   @expose() organization_id = (Int8<1>).column({ nonNull: true });

site/src/demo/schema/order_lines.ts

@@ -1,10 +1,10 @@
-import { TypegresLiveEvents, expose } from "typegres";
+import { expose } from "typegres";
 import { Int8, Text } from "typegres/postgres";
 import { db } from "../runtime";
 import { InventoryPositions } from "./inventory_positions";
 import { Orders } from "./orders";
 
-export class OrderLines extends db.Table("order_lines", { transformer: TypegresLiveEvents.makeTransformer() }) {
+export class OrderLines extends db.Table("order_lines", { live: true }) {
   // @generated-start
   @expose() id = (Int8<1>).column({ nonNull: true, generated: true });
   @expose() order_id = (Int8<1>).column({ nonNull: true });

site/src/demo/schema/orders.ts

@@ -1,4 +1,4 @@
-import { Connection, TypegresLiveEvents, sql, expose } from "typegres";
+import { Connection, sql, expose } from "typegres";
 import { Int8, Text, Timestamptz } from "typegres/postgres";
 import { z } from "zod";
 import { db } from "../runtime";
@@ -7,7 +7,7 @@ import { OrderLines } from "./order_lines";
 import { Organizations } from "./organizations";
 import { Shipments } from "./shipments";
 
-export class Orders extends db.Table("orders", { transformer: TypegresLiveEvents.makeTransformer() }) {
+export class Orders extends db.Table("orders", { live: true }) {
   // @generated-start
   @expose() id = (Int8<1>).column({ nonNull: true, generated: true });
   @expose() organization_id = (Int8<1>).column({ nonNull: true });
@@ -51,9 +51,9 @@ export class Orders extends db.Table("orders", { transformer: TypegresLiveEvents
       // The WHERE excludes 'delivered' so CASE always matches; the
       // `as Text<1>` asserts the non-null we structurally guarantee.
       // Interpolating `orders.status` into the template emits the
-      // properly-qualified column reference (the live transformer
-      // wraps the UPDATE in a CTE that also has a `status` column;
-      // unqualified would be ambiguous).
+      // properly-qualified column reference (live capture wraps the
+      // UPDATE in a CTE that also has a `status` column; unqualified
+      // would be ambiguous).
       .set(({ orders }) => ({
         status: Text.from(sql`
           CASE ${orders.status}

site/src/demo/schema/organizations.ts

@@ -1,4 +1,4 @@
-import { TypegresLiveEvents, expose } from "typegres";
+import { expose } from "typegres";
 import { Int8, Text } from "typegres/postgres";
 import { db } from "../runtime";
 import { Customers } from "./customers";
@@ -7,7 +7,7 @@ import { Locations } from "./locations";
 import { Users } from "./users";
 import { Orders } from "./orders";
 import { Shipments } from "./shipments";
-export class Organizations extends db.Table("organizations", { transformer: TypegresLiveEvents.makeTransformer() }) {
+export class Organizations extends db.Table("organizations", { live: true }) {
   // @generated-start
   @expose() id = (Int8<1>).column({ nonNull: true, generated: true });
   @expose() name = (Text<1>).column({ nonNull: true });

site/src/demo/schema/shipments.ts

@@ -1,9 +1,9 @@
-import { TypegresLiveEvents, sql, expose } from "typegres";
+import { sql, expose } from "typegres";
 import { Int8, Text, Timestamptz } from "typegres/postgres";
 import { db } from "../runtime";
 import { Orders } from "./orders";
 import { Organizations } from "./organizations";
-export class Shipments extends db.Table("shipments", { transformer: TypegresLiveEvents.makeTransformer() }) {
+export class Shipments extends db.Table("shipments", { live: true }) {
   // @generated-start
   @expose() id = (Int8<1>).column({ nonNull: true, generated: true });
   @expose() organization_id = (Int8<1>).column({ nonNull: true });

site/src/demo/schema/users.ts

@@ -1,8 +1,8 @@
-import { TypegresLiveEvents, expose } from "typegres";
+import { expose } from "typegres";
 import { Int8, Text } from "typegres/postgres";
 import { db } from "../runtime";
 import { Organizations } from "./organizations";
-export class Users extends db.Table("users", { transformer: TypegresLiveEvents.makeTransformer() }) {
+export class Users extends db.Table("users", { live: true }) {
   // @generated-start
   @expose() id = (Int8<1>).column({ nonNull: true, generated: true });
   @expose() organization_id = (Int8<1>).column({ nonNull: true });

site/src/demo/server/api.ts

@@ -148,16 +148,14 @@ export class Api {
     this.#currentUserToken = token;
   }
 
-  // Demo stop-button hook. `db.stopLive()` cancels every active
-  // subscription (parked consumers wake with AbortError and exit
-  // cleanly), then we re-start the bus so the next watch can
-  // subscribe immediately. In a real deployment the wire would have
-  // a per-iter abort channel; here we tear down the whole bus
-  // because the demo only ever has one iter at a time.
+  // Demo stop-button hook: cancels every active subscription (parked
+  // consumers wake with AbortError and exit cleanly); the live engine
+  // stays up and the next watch subscribes immediately. In a real
+  // deployment the wire would have a per-iter abort channel; here we
+  // cancel all subs because the demo only ever has one iter at a time.
   @expose()
   async resetLive(): Promise<void> {
-    await this.conn.stopLive();
-    await this.conn.startLive();
+    this.conn.cancelLiveSubscriptions();
   }
 
   // The principal for this RPC, resolved from the ambient

src/builder/delete.ts

@@ -18,8 +18,8 @@ type DeleteOpts<Name extends string, T extends TableBase, R extends RowType> = {
 };
 
 // Resolved DELETE — alias minted, where/returning evaluated against the
-// bound namespace. Transformers inspect structured fields without
-// re-parsing raw Sql.
+// bound namespace. Consumers (live capture) inspect structured fields
+// without re-parsing raw Sql.
 type FinalizedDeleteOpts<Name extends string, T extends TableBase, R extends RowType> = {
   tableName: Name;
   alias: Alias;
@@ -59,12 +59,10 @@ export class DeleteBuilder<Name extends string, T extends TableBase, R extends R
     this.#opts = opts;
   }
 
-  get tableName(): Name {
-    return this.#opts.instance.constructor.tableName as Name;
-  }
-
-  get database() {
-    return this.#opts.instance.constructor.database;
+  // The target table class — finalize-free access to its statics
+  // (tableName, database, live).
+  get table() {
+    return this.#opts.instance.constructor;
   }
 
   // Multiple where() calls are combined with AND. .where(true) matches all rows —
@@ -96,14 +94,14 @@ export class DeleteBuilder<Name extends string, T extends TableBase, R extends R
   }
 
   rowType(): R | undefined {
-    return this.#opts.returning?.({ [this.tableName]: this.#opts.instance } as Namespace<Name, T>);
+    return this.#opts.returning?.({ [this.table.tableName]: this.#opts.instance } as Namespace<Name, T>);
   }
 
   finalize(): FinalizedDelete<Name, T, R> {
     if (!this.#opts.where && !this.#opts.matchAll) {
       throw new Error("delete() requires .where() — use .where(true) to delete all rows");
     }
-    const tableName = this.tableName;
+    const tableName = this.table.tableName as Name;
     const alias = new Alias(tableName);
     const instance = reAlias(this.#opts.instance as RowType, alias) as T;
     const ns = { [tableName]: instance } as Namespace<Name, T>;
@@ -120,8 +118,10 @@ export class DeleteBuilder<Name extends string, T extends TableBase, R extends R
   }
 
   bind(ctx: CompileContext): BoundSql {
-    const t = this.#opts.instance.constructor.transformer?.delete;
-    return (t ? t(this) : this.finalize()).bind(ctx);
+    // Widen to Sql so ctx forwards: the Finalized* form declares zero-arg
+    // bind() today, but dispatching through the base signature keeps ctx
+    // flowing if it ever starts accepting one.
+    return (this.finalize() as Sql).bind(ctx);
   }
 
   override children() {
@@ -140,7 +140,7 @@ export class DeleteBuilder<Name extends string, T extends TableBase, R extends R
 
   @expose()
   debug(): this {
-    const compiled = compile(this, { database: this.database });
+    const compiled = compile(this, { database: this.table.database });
     console.log("Debugging query:", { sql: compiled.text, parameters: compiled.values });
     return this;
   }

src/builder/insert.ts

@@ -99,12 +99,10 @@ export class InsertBuilder<Name extends string, T extends TableBase, R extends R
     this.#opts = opts;
   }
 
-  get tableName(): Name {
-    return this.#opts.instance.constructor.tableName as Name;
-  }
-
-  get database() {
-    return this.#opts.instance.constructor.database;
+  // The target table class — finalize-free access to its statics
+  // (tableName, database, live).
+  get table() {
+    return this.#opts.instance.constructor;
   }
 
   @expose(fn.returns(z.custom<any>((v) => isRowType(v))))
@@ -127,11 +125,11 @@ export class InsertBuilder<Name extends string, T extends TableBase, R extends R
   // the output carry sql.unbound() as their SQL — harmless since rowType
   // is never compiled, only its [meta].__class is read for deserialize.
   rowType(): R | undefined {
-    return this.#opts.returning?.({ [this.tableName]: this.#opts.instance } as Namespace<Name, T>);
+    return this.#opts.returning?.({ [this.table.tableName]: this.#opts.instance } as Namespace<Name, T>);
   }
 
   finalize(): FinalizedInsert<Name, T, R> {
-    const tableName = this.tableName;
+    const tableName = this.table.tableName as Name;
     const alias = new Alias(tableName);
     const instance = reAlias(this.#opts.instance as RowType, alias) as T;
     const ns = { [tableName]: instance } as Namespace<Name, T>;
@@ -147,8 +145,10 @@ export class InsertBuilder<Name extends string, T extends TableBase, R extends R
   }
 
   bind(ctx: CompileContext): BoundSql {
-    const t = this.#opts.instance.constructor.transformer?.insert;
-    return (t ? t(this) : this.finalize()).bind(ctx);
+    // Widen to Sql so ctx forwards: the Finalized* form declares zero-arg
+    // bind() today, but dispatching through the base signature keeps ctx
+    // flowing if it ever starts accepting one.
+    return (this.finalize() as Sql).bind(ctx);
   }
 
   override children() {
@@ -167,7 +167,7 @@ export class InsertBuilder<Name extends string, T extends TableBase, R extends R
 
   @expose()
   debug(): this {
-    const compiled = compile(this, { database: this.database });
+    const compiled = compile(this, { database: this.table.database });
     console.log("Debugging query:", { sql: compiled.text, parameters: compiled.values });
     return this;
   }

src/builder/update.ts

@@ -86,12 +86,10 @@ export class UpdateBuilder<Name extends string, T extends TableBase, R extends R
     this.#opts = opts;
   }
 
-  get tableName(): Name {
-    return this.#opts.instance.constructor.tableName as Name;
-  }
-
-  get database() {
-    return this.#opts.instance.constructor.database;
+  // The target table class — finalize-free access to its statics
+  // (tableName, database, live).
+  get table() {
+    return this.#opts.instance.constructor;
   }
 
   // Multiple where() calls are combined with AND. .where(true) matches all rows
@@ -119,8 +117,8 @@ export class UpdateBuilder<Name extends string, T extends TableBase, R extends R
   }
 
   // Merge with whatever was already returned. Throws on key conflict.
-  // Used by mutation transformers to add bookkeeping columns without
-  // silently shadowing user columns.
+  // Used by live capture (PgExecutor wrapping, sqlite images) to add
+  // bookkeeping columns without silently shadowing user columns.
   returningMerge<R2 extends RowType>(
     fn: (ns: Namespace<Name, T>) => R2,
   ): UpdateBuilder<Name, T, R & R2> {
@@ -131,7 +129,7 @@ export class UpdateBuilder<Name extends string, T extends TableBase, R extends R
   }
 
   rowType(): R | undefined {
-    return this.#opts.returning?.({ [this.tableName]: this.#opts.instance } as Namespace<Name, T>);
+    return this.#opts.returning?.({ [this.table.tableName]: this.#opts.instance } as Namespace<Name, T>);
   }
 
   finalize(): FinalizedUpdate<Name, T, R> {
@@ -141,7 +139,7 @@ export class UpdateBuilder<Name extends string, T extends TableBase, R extends R
     if (!this.#opts.set) {
       throw new Error("update() requires .set()");
     }
-    const tableName = this.tableName;
+    const tableName = this.table.tableName as Name;
     const alias = new Alias(tableName);
     const instance = reAlias(this.#opts.instance as RowType, alias) as T;
     const ns = { [tableName]: instance } as Namespace<Name, T>;
@@ -160,8 +158,10 @@ export class UpdateBuilder<Name extends string, T extends TableBase, R extends R
   }
 
   bind(ctx: CompileContext): BoundSql {
-    const t = this.#opts.instance.constructor.transformer?.update;
-    return (t ? t(this) : this.finalize()).bind(ctx);
+    // Widen to Sql so ctx forwards: the Finalized* form declares zero-arg
+    // bind() today, but dispatching through the base signature keeps ctx
+    // flowing if it ever starts accepting one.
+    return (this.finalize() as Sql).bind(ctx);
   }
 
   override children() {
@@ -180,7 +180,7 @@ export class UpdateBuilder<Name extends string, T extends TableBase, R extends R
 
   @expose()
   debug(): this {
-    const compiled = compile(this, { database: this.database });
+    const compiled = compile(this, { database: this.table.database });
     console.log("Debugging query:", { sql: compiled.text, parameters: compiled.values });
     return this;
   }