diff --git a/CHANGELOG.md b/CHANGELOG.md index e67c9cb..9f284d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,16 +8,20 @@ While pre-1.0, the public API may change between 0.x releases. ## [Unreleased] -### Changed - -- **Rejection reasons are surfaced uniformly for mutations and commands (revises - ADR-0012 D3).** An `authorize` throw or a schema validation failure now reaches - the client with its reason (validation failures carry a `VALIDATION` code); - only `execute` errors stay sanitized. Previously a command's `authorize` error - was sanitized like its `execute`, unlike a mutation's. - ### Added +- **`Syncable(Base)` mixin — cohost sync on any Durable Object base (ADR-0015).** + The sync machinery is now a curried mixin factory, + `Syncable()(Base)`, so one DO can be both its framework's host (the + Agents SDK `Agent`, `@cloudflare/think`'s `Think`, a bare `DurableObject`) and a + tddc sync source — no dedicated sync DO, no mirror write. Exposed from the root + and a `./server/mixin` subpath. Sync sockets carry a reserved tag and a plain + attachment and claim only the `/_sync` path; all other traffic delegates to the + host base, so the two protocols never cross (proof: partyserver's `__pk` + filtering). `Actor` (`@cloudflare/actors`) is documented as unsupported because + its `Sockets` helper adopts foreign sockets on wake. See the README "Cohosting" + section. + - **Optional Standard Schema validation (ADR-0014).** A collection's `insert.schema` (the row schema, which also infers the collection's Row) and `update.schema` (a partial patch schema), and a command's schema, are checked @@ -27,6 +31,24 @@ While pre-1.0, the public API may change between 0.x releases. transforms, defaults, or coercion. See `recipes/zod-standard-schema-collections.md`. +### Changed + +- **`SyncDurableObject` is now `Syncable()(DurableObject)`** — zero API change. + All existing `extends SyncDurableObject` code, including + `this.sql`, `this.registerSync`, `this.runSyncedWrite`, and an overridable + `parseAttachment`, keeps compiling and behaving identically to 0.4.0 (the two + DO-global side effects — `ping/pong` auto-response and `PRAGMA + case_sensitive_like = ON` — stay ON for this base; they default OFF over any + other base, opt in with `this.sync.configure`). The internal `sql` getter was + removed from the mixin because it shadowed the host's `sql` tagged-template + method; reach `this.ctx.storage.sql` directly on a non-`DurableObject` base. + +- **Rejection reasons are surfaced uniformly for mutations and commands (revises + ADR-0012 D3).** An `authorize` throw or a schema validation failure now reaches + the client with its reason (validation failures carry a `VALIDATION` code); + only `execute` errors stay sanitized. Previously a command's `authorize` error + was sanitized like its `execute`, unlike a mutation's. + ## [0.4.0] — 2026-07-01 ### Changed diff --git a/README.md b/README.md index 1229520..06f68de 100644 --- a/README.md +++ b/README.md @@ -298,6 +298,78 @@ Task-oriented guides in [`recipes/`](./recipes): --- +## Cohosting: `Syncable` over a framework base + +`SyncDurableObject` is the trivial application of a mixin, `Syncable(Base)`, that +adds the sync machinery to **any** Durable Object subclass. Use it when a DO +already extends a framework base — the Cloudflare Agents SDK `Agent`, +`@cloudflare/think`'s `Think` — and you want that same DO to also be a sync +source, instead of standing up a second DO and mirror-writing to it (ADR-0015). + +```ts +import { Syncable } from "tanstack-do-db-collection" // or ".../server/mixin" + +// Curried: pin Env and your claims type, then apply over the runtime base. +class FeedAgent extends Syncable()(Agent) { + constructor(ctx: DurableObjectState, env: Env) { + super(ctx, env) // host constructor first + // Auth hook for the sync upgrade (same contract as parseAttachment): + this.sync.configure({ parseAttachment: (req) => readClaims(req) }) + ctx.blockConcurrencyWhile(async () => { + migrate(ctx.storage.sql) // you create the tables… + this.sync.registerSync(feedSchema) // …then register (ADR-0007) + }) + } +} +``` + +The sync API lives behind one facade, `this.sync` (`registerSync`, +`runSyncedWrite`, `parseAttachment`, `configure`), so the only names the mixin +adds to your class are `sync` and the four WebSocket/`fetch` handlers. tddc's +sockets carry a reserved tag and a plain attachment, and it claims only the +`/_sync` path (configurable) — everything else is delegated to your host base, so +the two protocols never cross. No framework is added to tddc's dependency graph; +you supply `Base`. + +**Reach `this.ctx.storage.sql`, not `this.sql`, on a mixed base.** The mixin does +**not** define a `sql` member, because both partyserver and agents define `sql` +as a tagged-template method and a getter would shadow it. (`SyncDurableObject` +still has `this.sql` — a bare `DurableObject` has no `sql` to shadow.) + +**Your `parseAttachment` claims must not use the key `__pk`.** partyserver marks +its own sockets with a `__pk` attachment key; a sync claim object carrying `__pk` +would make a partyserver-like host mis-claim the sync socket. The reserved tag +keeps tddc's own side correct regardless, and the mixin logs an error if it sees +`__pk` in a sync attachment — but keep it out of your claims. + +> [!IMPORTANT] +> **Two DO-global side effects default OFF over a non-`DurableObject` base**, and +> ON for plain `SyncDurableObject` (0.4.0 behavior). Opt in with `configure`: +> - `autoResponse` — `setWebSocketAutoResponse("ping","pong")` is DO-wide and +> would answer a literal `"ping"` frame from *your host's* client before the +> host sees it. +> - `caseSensitiveLike` — `PRAGMA case_sensitive_like = ON` is connection-wide and +> changes your host's own `LIKE` queries. (tddc needs it for filtered-sub +> parity — ADR-0013.) +> +> ```ts +> this.sync.configure({ autoResponse: true, caseSensitiveLike: true }) +> ``` + +> [!WARNING] +> **Never register a host-owned table as a synced collection.** tddc installs CDC +> triggers only on the tables you register, so a host's own tables +> (`cf_agents_*`, Think's `assistant_*`) stay untouched by default. But nothing +> stops you from *registering* one by mistake — do not, or every host write to it +> emits change rows to your clients. + +**`@cloudflare/actors`' `Actor` is not supported.** Its `Sockets` helper adopts +every hibernated socket on wake (including tddc's), broadcasts to them, and closes +sockets it does not own — a host defect the mixin cannot work around without +changing the Actors package. Use `Agent`, `Think`, or a plain `DurableObject`. + +--- + ## Non-goals - **Multi-DO transactions.** A transaction touches collections in one DO. diff --git a/docs/adr/0001-sync-architecture.md b/docs/adr/0001-sync-architecture.md index 7ea4ca7..9f96c89 100644 --- a/docs/adr/0001-sync-architecture.md +++ b/docs/adr/0001-sync-architecture.md @@ -2,7 +2,11 @@ **Status:** Accepted, **amended by [ADR-0002](./0002-adversarial-review-corrections.md)** (ordering barrier, on-demand shaping, retention/liveness, before-image dropped). -Read 0002 alongside this document. +Read 0002 alongside this document. **D13's "base class" is reframed as a mixin by +[ADR-0015](./0015-syncable-mixin.md)**: the sync core is now `Syncable(Base)` and +`SyncDurableObject` is its trivial application over `DurableObject` — same +hibernation/`acceptWebSocket`/`serializeAttachment` mechanics, now composable over +any DO base. ## Context diff --git a/docs/adr/0006-server-originated-writes.md b/docs/adr/0006-server-originated-writes.md index 95d2bfa..6d582de 100644 --- a/docs/adr/0006-server-originated-writes.md +++ b/docs/adr/0006-server-originated-writes.md @@ -4,6 +4,10 @@ model with a third write origin. The "caller ensures init" caveat below is retired by [ADR-0007](./0007-author-owned-schema-register-sync.md) (schema + triggers now exist at construction via `registerSync`). +[ADR-0015](./0015-syncable-mixin.md) leaves `runSyncedWrite` unchanged and makes +it also the write path for a host tool body on a mixed base +(`this.sync.runSyncedWrite`), so a committed insert drains the CDC log and +broadcasts in the same step regardless of the DO's framework base. ## Context diff --git a/docs/adr/0007-author-owned-schema-register-sync.md b/docs/adr/0007-author-owned-schema-register-sync.md index cb8c6c9..f5811ec 100644 --- a/docs/adr/0007-author-owned-schema-register-sync.md +++ b/docs/adr/0007-author-owned-schema-register-sync.md @@ -4,6 +4,11 @@ moves the [ADR-0001](./0001-sync-architecture.md) D9 enforcement point, and retires [ADR-0006](./0006-server-originated-writes.md)'s init caveat. Hard breaking change to the DO authoring API (pre-1.0). +[ADR-0015](./0015-syncable-mixin.md) leans on this decision: because +`registerSync` is author-driven (called in the constructor's +`blockConcurrencyWhile` after the tables exist), the mixin needs no +base-constructor magic and composes cleanly with a host base's own constructor — +the author creates tables, then calls `this.sync.registerSync`, in that order. ## Context diff --git a/docs/adr/0008-orphaned-cdc-triggers.md b/docs/adr/0008-orphaned-cdc-triggers.md index 0d18cb4..a9e2ad5 100644 --- a/docs/adr/0008-orphaned-cdc-triggers.md +++ b/docs/adr/0008-orphaned-cdc-triggers.md @@ -3,6 +3,10 @@ **Status:** Accepted — implemented. Records a limitation of [ADR-0007](./0007-author-owned-schema-register-sync.md)'s `registerSync` and the fix now shipped: Option 1, reaping in `registerSync`. +[ADR-0015](./0015-syncable-mixin.md) cites this GLOB `_sync_changes_*` namespace +as its trigger-collision-safety proof: on a host that owns unregistered tables +(`cf_agents_*`, `assistant_*`), the reaper's literal-`_` GLOB can never drop a +host trigger, and unregistered tables get no capture triggers at all. ## Context diff --git a/docs/adr/0015-syncable-mixin.md b/docs/adr/0015-syncable-mixin.md new file mode 100644 index 0000000..3ac74e7 --- /dev/null +++ b/docs/adr/0015-syncable-mixin.md @@ -0,0 +1,208 @@ +# 0015 — `Syncable` mixin: the sync core as a mixin over any DO base + +**Status:** Accepted — implemented. Reframes [ADR-0001](./0001-sync-architecture.md) +D13's "base class" as a mixin factory; `SyncDurableObject` is now its trivial +application over `DurableObject`. Composes with, does not change, +[ADR-0006](./0006-server-originated-writes.md) (`runSyncedWrite`), +[ADR-0007](./0007-author-owned-schema-register-sync.md) (author-driven +`registerSync`), and [ADR-0008](./0008-orphaned-cdc-triggers.md) (the GLOB trigger +namespace — the collision-safety proof cited below). + +## Context + +`SyncDurableObject` was an abstract base class that `extends DurableObject` +directly (`sync-do.ts:38`, 0.4.0). A JavaScript class has one base, so a DO that +already extends a framework base — the Cloudflare Agents SDK `Agent`, +`@cloudflare/think`'s `Think`, or any other DO subclass — could not *also* be a +sync source. That forced a dedicated sync DO per scope plus a mirror write from +the DO that owns the data, the exact pattern this library exists to avoid. + +An audit of the class found no structural blocker to hosting both a framework's +WebSocket surface and tddc's sync WebSocket protocol on one DO. Every coupling is +host-agnostic already or a mechanical namespacing fix, and the one hard +name collision (`sql`) has a clean resolution. The direction of every required +change is tddc-side: today tddc is the bad citizen (restores all sockets on wake +untagged, claims every upgrade in `fetch`, never delegates unknown WS events to +`super`). + +## The cohosting proof + +The mixin is only useful if one DO can host both protocols with **zero** change +to partyserver, agents, or Think. Verified against partyserver 0.5.8, agents +0.17.3, `@cloudflare/think` 0.12.1: + +- **partyserver ignores sockets it did not open.** `isPartyServerWebSocket(ws)` + is true only if the socket attachment carries a `__pk` key; every hibernation + handler short-circuits a non-`__pk` socket (`webSocketMessage`, `webSocketClose`, + `webSocketError`) and connection enumeration filters the same way, so + `broadcast()` never touches a foreign socket. tddc accepts its socket with a + plain attachment and **no** `__pk`, so partyserver is already blind to sync + sockets with no change on its side. +- **Agent and Think add no socket entry points.** `Agent extends Server` and + defines no `webSocketMessage/Close/Error`; `Agent.fetch` delegates upward via + `super.fetch`. Think's dist contains zero `acceptWebSocket`/`WebSocketPair`/ + `getWebSockets` and only wraps `onConnect`. +- **tddc defines no `alarm()`.** Compaction rides `ctx.waitUntil` + (`mixin.ts`, `#maybeCompact`), so the DO's single alarm slot stays wholly owned + by the host. No chaining needed. + +## Decision + +Ship the server as a curried mixin factory `Syncable()(Base)` +(`src/server/mixin.ts`), with `SyncDurableObject = Syncable()(DurableObject)` +re-exposing its legacy protected surface so every `extends SyncDurableObject` keeps compiling and behaving identically to 0.4.0. The 187 tests passing +without a single assertion edit is the back-compat proof gate. + +Three independent discriminators keep the two protocols apart. Any one suffices +for host-side safety; together they are belt and suspenders: + +1. **Reserved hibernation tag** `SYNC_TAG = "_tddc"` on `acceptWebSocket`. Tags + are the only server-side filter `getWebSockets` offers, so the wake-time + restore and every handler's ownership check key off it. Without it, the + broadcaster fans sync frames onto host sockets after a wake. **Exception for a + bare `DurableObject`:** there is no host to share with, so tddc owns *every* + socket — the restore uses `getWebSockets()` (all) and the ownership check + returns true unconditionally. This is what keeps a **legacy untagged socket**, + accepted by a pre-mixin 0.4.0 build and surviving a hibernation wake across the + upgrade, working instead of being silently ignored. Over any other base the + restore is tag-filtered (`getWebSockets(SYNC_TAG)`). +2. **Plain attachment, no `__pk`.** The independent discriminator that keeps a + `__pk`-filtering host blind to sync sockets, needing no cooperation from tddc. + This is the one discriminator an author could accidentally violate — a claims + object that itself carries a `__pk` key would make the host mis-claim the sync + socket. The tag (discriminator 1) keeps tddc's own side correct regardless, and + `#acceptSyncSocket` logs a loud error if a sync attachment carries `__pk` over a + non-DO base; the README documents the reserved key. +3. **Dedicated fetch path** (default `/_sync`). Upgrades are partitioned before + either protocol sees them; a non-matching upgrade returns `super.fetch(request)` + (safe because `Agent.fetch` itself delegates upward). WS events on an untagged + socket delegate via `super.webSocketMessage?.()` — a no-op on a bare DO, and on + Agent/Think it resolves to partyserver's handler, which re-guards on `__pk`. + +### The `sql` getter is deleted (mandatory, not stylistic) + +0.4.0 exposed `protected get sql(): SqlStorage` (`sync-do.ts:103`). partyserver +defines `sql` as a **tagged-template method** (`dist/index.js:557`) and agents +redefines it (`agent-tool-types` d.ts, `sql(strings, …values)`). A property +getter named `sql` on the mixed class shadows that method and breaks every +`cf_agents_*` query in the host. The mixin therefore uses a private `get #sql()` +over `this.ctx.storage.sql` internally and defines **no** public/protected `sql`. +`SyncDurableObject` re-adds `protected get sql()` because a bare `DurableObject` +defines no `sql` member to shadow; over any other host, authors reach +`this.ctx.storage.sql` directly (as the host-matrix fixture does). + +### The two DO-global side effects are base-dependent opt-ins + +`setWebSocketAutoResponse("ping","pong")` and `PRAGMA case_sensitive_like = ON` +affect the whole DO, not just tddc's sockets or queries: the auto-response would +swallow a literal `"ping"` frame from a host's client before the host sees it, and +the pragma changes the host's own `LIKE` semantics. Both default **ON** when +`Base === DurableObject` (bit-identical 0.4.0) and **OFF** over any other base, +with `this.sync.configure({ autoResponse, caseSensitiveLike })` to opt in. + +**Open questions resolved before merge (greps against the installed hosts):** + +- *Does any host client send a literal `"ping"` keepalive?* No. No `"ping"` + string frame in agents 0.17.3, partyserver 0.5.8, or think 0.12.1 dists + (client or server). The auto-response slot is free. Default-off over a non-DO + base is defensive belt-and-suspenders and is what the cohosting smoke confirms. +- *Do any host `LIKE` queries rely on case-insensitivity?* No. No `LIKE` + SELECT/WHERE in the agents/partyserver/think server dists; Think's search uses + FTS5 `MATCH`. Default-off over a non-DO base is defensive. +- *Is the facade name `sync` taken on any host?* No. Neither partyserver's + `Server`, agents' `Agent`, nor Think expose a `sync` member (confirmed against + their `.d.ts` type surfaces). The fallback name `tddcSync` was not needed. + +### One facade to shrink the collision surface + +The loose protected members (`codec`, `subs`, `registry`, `broadcaster`, and +the internal handlers) are now `#private` or live behind a single +`this.sync: SyncApi` facade (`registerSync`, `runSyncedWrite`, `parseAttachment`, +`configure`, `registry`, `drainAndBroadcast`). So the names a mixed class puts on +its prototype that could collide with an arbitrary host shrink to the four +runtime-dispatched methods (`fetch`, `webSocketMessage/Close/Error`) plus `sync`. + +Two deliberate narrowings, each documented rather than hidden: + +- **Numeric tuning knobs stay `protected` overridable fields** (`tickMs`, + `compactionEvery`, `maxOpsPerMutation`, `maxSubsPerSocket`, `maxFrameBytes`, + `changelogRetentionMs`, `dedupRetentionMs`). They are the documented + subclass-tuning contract, they provably do not collide with any supported host + (grep-verified), and collapsing them would break existing subclasses with no + security benefit. The mixin's *return type* hides them (so they never widen the + mixed-class public surface); `SyncDurableObject` re-declares them (ambient) so + legacy `protected override readonly tickMs = …` subclasses keep compiling. Over + a non-DO host, tune via `configure`/private fields. +- **`registry` and `drainAndBroadcast` are exposed under the facade** + (`this.sync.registry`, `this.sync.drainAndBroadcast`) — behind `sync`, so never + on the bare collision surface — and re-aliased as protected on + `SyncDurableObject` for the two white-box tests and the documented manual-drain + API. + +## Trigger safety on a host with pre-existing tables + +A host base (Agent, Think) owns tables the author never registers +(`cf_agents_state`, `assistant_*`, `cf_think_*`). Safety holds by construction, +needing only tests and one documented rule: + +- Triggers install per **registered** collection only (`registerSync` → + `ensureTriggers` over the declared set). An unregistered host table gets no + trigger, so host writes emit no CDC rows. +- The reaper drops only triggers matching `GLOB '_sync_changes_*'`, and GLOB + treats `_` literally ([ADR-0008](./0008-orphaned-cdc-triggers.md)), so it can + never drop a host trigger. +- tddc's own tables are `_sync_`-prefixed and cannot collide with host names; + `assertValidCollection` rejects the `_sync_` prefix and `assertSyncCompatible` + requires a sole TEXT client-supplied pk, which `cf_agents_*` fail. + +The one real constraint, documented in the README: **never register a +host-owned table as a synced collection** (the prefix guard does not catch +`cf_`/`assistant_` names). The host-matrix test pins all of the above. + +## What the mixin cannot support: `@cloudflare/actors`' `Actor` + +The Actors `Sockets` helper adopts **every** hibernated socket in its constructor +with an unfiltered `ctx.getWebSockets()` (`packages/sockets/src/index.ts:22–39`), +broadcasts to all of them on `message('*')` (`:43–55`), and closes foreign +sockets in `webSocketClose` (`:61–77`); `Actor.webSocketMessage` hands frames from +all sockets to the app handler. A subclass mixin can intercept the dispatch +methods, but it cannot prevent the constructor-time adoption or the fan-out +without changing the Actors package. Until Actors filters foreign sockets the way +partyserver does, `Actor` is **unsupported**. This is a host limitation, not a +tddc design gap; a PR against Actors is possible in principle but out of scope. + +## Consequences + +- **`Syncable()` must be applied exactly once, over the outermost DO base — + never stacked over another `Syncable()` application.** Verified: + `class Outer extends Syncable()(Syncable()(DurableObject)) {}` is a real + `tsc` error (the inner application's `ctx`/`Env` typing doesn't satisfy + `DOCtor`'s construct-signature constraint without a cast). Forcing it past + the type checker with a cast still doesn't work at runtime: `fetch`, + `webSocketMessage/Close/Error`, and the `sync` getter are per-class + overrides, so the outer layer's definitions always shadow the inner + layer's — the inner layer's `registerSync` is never reachable and it never + dispatches. Cohost by putting a **host framework** under `Syncable()` + (`Syncable()(Agent)`), never another `Syncable()` application. +- **Cold-snapshot row order is now deterministic.** The `#handleSub` snapshot + path (no `since` cursor) and the paginated `#handleFetch` path both lower + through `compileSubsetQuery` (`sql-compiler.ts`), which defaults to + `ORDER BY rowid` when the client sends no `orderBy` — previously a bare + `SELECT * FROM tbl` left row order as an accident of SQLite's query plan + (field-verified: a `WHERE` touching the pk can make the planner prefer the + pk's autoindex over a rowid scan, returning pk-sorted rows instead of + insertion order). `rowid` matches insertion order among currently-live rows + and needs no schema change, since `assertSyncCompatible` (ADR-0007, D9) + already forbids the `INTEGER PRIMARY KEY` pk that would alias it. +- One DO class can be both a framework host and a sync source + (`class FeedAgent extends Syncable()(Agent)`), with no + framework added to tddc's dependency graph — the app supplies `Base`. +- `SyncDurableObject` is unchanged for its users; only the internal home of the + code moved. +- The entire cohosting guarantee rests on partyserver's `__pk` filtering, an + internal behavior of a pre-1.0 package that accepts no external PRs. Every bump + of `agents`/`partyserver` must be gated by a real-`agents` cohosting smoke (out + of CI, to keep the ~13 MB dep out of the package) that opens both socket types + on one DO, forces a hibernation wake, and asserts no cross-talk. The in-CI + host-matrix suite pins the same contract against a fake host. diff --git a/docs/adr/README.md b/docs/adr/README.md index c683d94..58cf345 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -8,7 +8,7 @@ explains the displacement. | # | Title | Status | |---|---|---| | [0000](./0000-record-architecture-decisions.md) | Record architecture decisions | Accepted | -| [0001](./0001-sync-architecture.md) | Sync architecture: single-ordered-stream over a Durable Object | Accepted (amended by 0002; D11 builder superseded by 0014) | +| [0001](./0001-sync-architecture.md) | Sync architecture: single-ordered-stream over a Durable Object | Accepted (amended by 0002; D11 builder superseded by 0014; D13 base class reframed as a mixin by 0015) | | [0002](./0002-adversarial-review-corrections.md) | Corrections from adversarial review: ordering, shaping, retention | Accepted (C5 retention refined by 0009) | | [0003](./0003-atomic-cursor-fetch.md) | Cursor load-more is one atomic fetch, not two | Accepted (naming amended by 0005) | | [0004](./0004-after-commit-hook.md) | Side effects go in a fire-and-forget `afterCommit`, not the transaction | Accepted | @@ -21,3 +21,4 @@ explains the displacement. | [0012](./0012-wire-input-hardening.md) | Wire-input hardening: frame-shape guards, inbound limits, sanitized execute errors | Accepted | | [0013](./0013-predicate-floor-one-evaluator.md) | Filtered-subscription membership: one evaluator is the source of truth; the floor is the verified-agreeing set | Accepted | | [0014](./0014-object-sync-schema.md) | `defineSync`: one schema value, mutations on the collection, commands on the connection | Accepted (supersedes 0001 D11 builder; closes 0010 manifest) | +| [0015](./0015-syncable-mixin.md) | `Syncable` mixin: the sync core as a mixin over any DO base | Accepted (reframes 0001 D13; Actor unsupported) | diff --git a/package-lock.json b/package-lock.json index 49a9a1e..ab96388 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "tanstack-do-db-collection", - "version": "0.3.3", + "version": "0.4.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "tanstack-do-db-collection", - "version": "0.3.3", + "version": "0.4.0", "license": "MIT", "dependencies": { "@msgpack/msgpack": "^3.0.0" diff --git a/package.json b/package.json index 75296cc..8c2fe3e 100644 --- a/package.json +++ b/package.json @@ -33,6 +33,10 @@ "./client": { "types": "./dist/client/index.d.ts", "import": "./dist/client/index.js" + }, + "./server/mixin": { + "types": "./dist/server/mixin.d.ts", + "import": "./dist/server/mixin.js" } }, "files": [ diff --git a/src/server/index.ts b/src/server/index.ts index 317ebe8..2104f81 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -4,6 +4,7 @@ // diffs. Imports the workerd runtime (`cloudflare:workers`); not for browsers. // // - SyncDurableObject: hibernating-WebSocket base class. +// - Syncable: the mixin factory; apply the sync machinery over any DO base. // - defineSync: the object-schema authoring API (collection/command/schema). export { assertValidCollection, compileSchema, defineSync } from "./registry.ts" @@ -29,3 +30,5 @@ export type { UpdateOp, } from "./registry.ts" export { SyncDurableObject } from "./sync-do.ts" +export { Syncable, SYNC_TAG } from "./mixin.ts" +export type { DOCtor, SyncApi, SyncableOptions, SyncMixin } from "./mixin.ts" diff --git a/src/server/mixin.ts b/src/server/mixin.ts new file mode 100644 index 0000000..e222947 --- /dev/null +++ b/src/server/mixin.ts @@ -0,0 +1,893 @@ +// Syncable(Base) — the sync machinery as a mixin factory (ADR-0015). +// +// The body of `SyncDurableObject` extracted into a factory that applies over any +// Durable Object subclass, so one DO class can be both its framework's host (the +// Agents SDK `Agent`, `@cloudflare/think`'s `Think`, or a bare `DurableObject`) +// AND a tddc sync source. `SyncDurableObject` (sync-do.ts) is the trivial +// application of this factory over `DurableObject`, preserving 0.4.0 exactly. +// +// Cohosting safety rests on three independent discriminators (ADR-0015): +// - a reserved hibernation tag (SYNC_TAG) on every sync socket, so wake-time +// restore and the broadcaster only ever touch tddc's own sockets; +// - a plain socket attachment with no partyserver `__pk` key, so a host that +// filters on `__pk` (partyserver, and therefore Agent/Think) is blind to +// sync sockets with no cooperation from tddc; +// - a dedicated fetch path (default "/_sync") so upgrades are partitioned +// before either protocol sees them, with non-matching traffic delegated to +// `super.fetch`. +// The `sql` getter is deliberately absent: a property `sql` would shadow the +// `sql` tagged-template method partyserver/agents define (ADR-0015). Internals +// reach SQLite through `this.ctx.storage.sql`. + +import { DurableObject } from "cloudflare:workers" +import type { SqlStorage, SqlStorageValue } from "@cloudflare/workers-types" +import { createFrameCodec, type FrameCodec } from "../wire/frame-codec.ts" +import type { ClientFrame, ServerFrame } from "../wire/frames.ts" +import { + compactChanges, + currentSeq, + ensureTriggers, + getDrainCursor, + hydrateRows, + initSchema, + minChangeSeq, + pruneChanges, + readChangesSince, + readChangesSinceFor, + setDrainCursor, +} from "./changes.ts" +import { Broadcaster } from "./broadcast.ts" +import { decodeResult, encodeResult, lookupTx, recordTx, type SeenTx, sweepDedup } from "./dedup.ts" +import { compileSchema, type CompiledSync, type SyncSchema, ValidationError } from "./registry.ts" +import { andPredicates, compileSubsetQuery, UnsupportedPredicateError } from "./sql-compiler.ts" +import { SubscriptionRegistry, type Sub } from "./subscriptions.ts" + +/** Reserved hibernation tag stamped on every sync socket. The wake-time restore + * (`getWebSockets(SYNC_TAG)`) and every handler's socket-ownership check key off + * this tag, so tddc never touches a host's sockets and vice versa (ADR-0015). */ +export const SYNC_TAG = "_tddc" + +/** Runtime options for the mixin. `configure()` in your constructor. + * Numeric tuning knobs stay protected overridable fields (see ADR-0015). */ +export interface SyncableOptions { + /** URL pathname the mixin claims in `fetch`. Default "/_sync". Ignored on a + * bare `DurableObject` base, which has no host `fetch` to delegate to and so + * owns every upgrade (0.4.0 parity). */ + path?: string + /** `setWebSocketAutoResponse("ping","pong")` — DO-global. Default: true when + * Base === DurableObject (preserves SyncDurableObject), false otherwise. */ + autoResponse?: boolean + /** `PRAGMA case_sensitive_like = ON` — connection-global. Same defaulting rule + * as `autoResponse`. */ + caseSensitiveLike?: boolean + /** Auth hook, same contract as the legacy `parseAttachment`: validate the + * upgrade and produce the attachment, or throw a `Response` to reject. */ + parseAttachment?: (req: Request) => TUser | Promise +} + +/** The single facade the mixin adds. Everything that used to be a loose + * protected member lives behind this one name to shrink the collision surface + * with an arbitrary host to the four runtime-dispatched methods plus `sync`. */ +export interface SyncApi { + /** Unchanged semantics (ADR-0007): call in blockConcurrencyWhile after your + * tables exist. */ + registerSync(schema: SyncSchema): void + /** Unchanged semantics (ADR-0006): server write + drain + broadcast. */ + runSyncedWrite(fn: (sql: SqlStorage) => T): T + /** The resolved auth hook (as configured). */ + parseAttachment(req: Request): TUser | Promise + /** Set options; safe to call from the host constructor. */ + configure(opts: SyncableOptions): void + /** The compiled schema; throws (ADR-0007) if `registerSync` hasn't run yet. + * Behind the facade so the name never shadows a host member (ADR-0015). */ + readonly registry: CompiledSync + /** Drain the CDC log and broadcast pending deltas (ADR-0006). The manual + * broadcast trigger for a raw server-side write done outside `runSyncedWrite`. */ + drainAndBroadcast(): void +} + +/** The surface the mixin adds to `Base`. The four methods are real, runtime- + * dispatched overrides so workerd finds them on the prototype. */ +export interface SyncMixin { + readonly sync: SyncApi + fetch(request: Request): Promise + webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise + webSocketClose(ws: WebSocket, code: number, reason: string, wasClean: boolean): Promise + webSocketError(ws: WebSocket, error: unknown): Promise +} + +/** The structural constructor shape the mixin needs from a host — nothing more + * than a `DurableObject` subclass. No framework is imported or depended on. */ +export type DOCtor = abstract new (...args: any[]) => DurableObject + +/** + * `Syncable` — curried so `Env` and `TUser` are pinned by the caller while + * `Base` stays a runtime value: + * + * class FeedAgent extends Syncable()(Agent) { … } + * + * The outer call pins the generics; the inner call takes the runtime `Base` and + * returns a class extending it with the `sync` facade re-exposed at `Env`/`TUser`. + */ +export function Syncable() { + return function (Base: TBase) { + abstract class SyncableMixin extends Base { + // ---- configuration ------------------------------------------------------ + /** Fetch path this instance claims (see SyncableOptions.path). */ + #path = "/_sync" + #autoResponse = false + #caseSensitiveLike = false + #parseAttachmentHook: (req: Request) => TUser | Promise = () => undefined as TUser + /** True when `Base` defines a `fetch` we can delegate non-sync traffic to. + * On a bare `DurableObject` this is false and the mixin owns all upgrades. */ + readonly #hasSuperFetch: boolean + /** True when `Base === DurableObject`. A bare DO shares its sockets with no + * host, so tddc owns every socket (incl. legacy untagged 0.4.0 sockets). */ + readonly #isBareDO: boolean + + // ---- tuning knobs (protected, overridable — ADR-0015) ------------------ + /** Egress coalescer tick (ms) — the single user-perceived-latency knob. */ + protected readonly tickMs: number = 50 + /** Compact the change log every this-many drained mutations (not on a timer — + * an alarm would wake idle DOs; this rides recent work). */ + protected readonly compactionEvery: number = 200 + /** Age bound for `_sync_changes` (ADR-0009). Changes older than this are + * pruned during compaction; a reconnect older than the surviving floor gets a + * full re-snapshot instead of a delta. `null` disables retention. Default 2 days. */ + protected readonly changelogRetentionMs: number | null = 172_800_000 + /** Dedup retention window (ms), independent of changelog retention (C5). */ + protected readonly dedupRetentionMs: number = 3_600_000 + /** Maximum ops in a single `mut` frame (ADR-0012). Reject-don't-truncate. */ + protected readonly maxOpsPerMutation: number = 128 + /** Maximum concurrent subscriptions per socket (ADR-0012). */ + protected readonly maxSubsPerSocket: number = 256 + /** Maximum inbound frame size in bytes (ADR-0012). */ + protected readonly maxFrameBytes: number = 1_048_576 + + // ---- internal machinery (private — off the collision surface) ---------- + #compiled: CompiledSync | undefined + readonly #codec: FrameCodec = createFrameCodec() + readonly #subs = new SubscriptionRegistry() + #writesSinceCompaction = 0 + readonly #broadcaster: Broadcaster + readonly #liveWs = new Set() + readonly #api: SyncApi + + constructor(...args: any[]) { + super(...args) + // Base-dependent defaults: reproduce 0.4.0 exactly on a bare DO, default + // the two DO-global side effects OFF over any other host (ADR-0015). + this.#isBareDO = (Base as unknown) === (DurableObject as unknown) + this.#hasSuperFetch = typeof (Base.prototype as { fetch?: unknown }).fetch === "function" + this.#autoResponse = this.#isBareDO + this.#caseSensitiveLike = this.#isBareDO + if (this.#autoResponse) this.#applyAutoResponse(true) + if (this.#caseSensitiveLike) this.#applyCaseSensitiveLike(true) + // Restore live sockets after a hibernation wake. On a bare DO tddc owns + // EVERY socket (there is no host to share with), so restore all of them — + // including legacy untagged sockets accepted by a pre-mixin 0.4.0 build + // that survive the wake across an upgrade. Over any other base, restore + // ONLY our tagged sockets so the broadcaster never touches a host socket. + const restore = this.#isBareDO ? this.ctx.getWebSockets() : this.ctx.getWebSockets(SYNC_TAG) + for (const ws of restore) this.#liveWs.add(ws) + this.#broadcaster = new Broadcaster((ws, frame) => this.#send(ws, frame), this.tickMs) + this.#broadcaster.start(() => this.#liveWs) + const self = this + this.#api = { + registerSync: (schema) => self.#registerSync(schema), + runSyncedWrite: (fn) => self.#runSyncedWrite(fn), + parseAttachment: (req) => self.#parseAttachmentHook(req), + configure: (opts) => self.#configure(opts), + drainAndBroadcast: () => self.#drainAndBroadcast(), + get registry() { + return self.#registry + }, + } + } + + get sync(): SyncApi { + return this.#api + } + + #applyAutoResponse(on: boolean): void { + // Auto-pong via the runtime: survives hibernation, no per-message billing. + // `off` clears the pair so `configure({ autoResponse: false })` is a real + // toggle (undoing the bare-DO default), not a dead option. + this.ctx.setWebSocketAutoResponse(on ? new WebSocketRequestResponsePair("ping", "pong") : undefined) + } + + #applyCaseSensitiveLike(on: boolean): void { + // Make SQLite LIKE case-sensitive so the SQL snapshot path matches + // @tanstack/db's case-sensitive `like` evaluator on the delta path — the + // single source of truth for filtered-subscription membership (ADR-0013). + // Connection-scoped; re-applied on every instantiation (incl. a wake). The + // `off` branch makes `configure({ caseSensitiveLike: false })` a real toggle. + this.ctx.storage.sql.exec(`PRAGMA case_sensitive_like = ${on ? "ON" : "OFF"}`) + } + + #configure(opts: SyncableOptions): void { + if (opts.path !== undefined) this.#path = opts.path + if (opts.parseAttachment !== undefined) this.#parseAttachmentHook = opts.parseAttachment + if (opts.autoResponse !== undefined) { + this.#autoResponse = opts.autoResponse + this.#applyAutoResponse(opts.autoResponse) + } + if (opts.caseSensitiveLike !== undefined) { + this.#caseSensitiveLike = opts.caseSensitiveLike + this.#applyCaseSensitiveLike(opts.caseSensitiveLike) + } + } + + /** SQLite handle. Private accessor (NOT a public `sql` getter — that would + * shadow the host's `sql` tagged-template method, ADR-0015). */ + get #sql(): SqlStorage { + return this.ctx.storage.sql + } + + /** The compiled schema. Throws if `registerSync` hasn't run yet (ADR-0007). */ + get #registry(): CompiledSync { + if (!this.#compiled) { + throw new Error( + "sync not registered — call this.sync.registerSync(schema) in your constructor's " + + "blockConcurrencyWhile, after creating your tables", + ) + } + return this.#compiled + } + + /** + * Wire collections for sync: validate each table is sync-compatible (ADR-0007) + * and reconcile its CDC triggers — install the registered set, drop triggers + * for any collection no longer registered (ADR-0008). The author owns table + * creation; call this AFTER the tables exist. Idempotent. + */ + #registerSync(schema: SyncSchema): void { + const compiled = compileSchema(schema) + initSchema(this.#sql) + ensureTriggers(this.#sql, compiled.collections.values()) + this.#compiled = compiled + } + + // ---- runtime-dispatched overrides -------------------------------------- + + override async fetch(req: Request): Promise { + if (req.headers.get("Upgrade") === "websocket" && this.#claimsUpgrade(req)) { + return this.#acceptSyncSocket(req) + } + // Non-sync traffic: delegate to the host if it has a fetch (Agent.fetch + // itself delegates upward; partyserver's fetch handles the rest). On a + // bare DurableObject there is nothing to delegate to. + if (this.#hasSuperFetch) { + return (super.fetch as (r: Request) => Promise).call(this, req) + } + return new Response("expected websocket upgrade", { status: 426 }) + } + + /** True iff this upgrade is ours. On a bare DO (no host fetch) we own every + * upgrade regardless of path — exactly 0.4.0. With a host present we claim + * only the configured path and let everything else fall to `super.fetch`. */ + #claimsUpgrade(req: Request): boolean { + if (!this.#hasSuperFetch) return true + const pathname = new URL(req.url).pathname + return pathname === this.#path || pathname.endsWith(this.#path) + } + + async #acceptSyncSocket(req: Request): Promise { + let attachment: TUser + try { + attachment = await this.#parseAttachmentHook(req) + } catch (e) { + if (e instanceof Response) return e + return new Response("unauthorized", { status: 401 }) + } + + // The socket attachment must not carry partyserver's reserved `__pk` key, + // or a partyserver-like host would mis-claim this sync socket as its own + // (the second, host-side discriminator). The SYNC_TAG below always keeps + // tddc's own side correct; this guard fails loud so a claim object that + // happens to use `__pk` cannot silently break host isolation. + if (!this.#isBareDO && attachment != null && typeof attachment === "object" && "__pk" in attachment) { + console.error( + "sync attachment carries a reserved `__pk` key — a partyserver-like host will mis-claim " + + "this socket. Remove `__pk` from your parseAttachment claims (ADR-0015).", + ) + } + + const pair = new WebSocketPair() + const client = pair[0] + const server = pair[1] + server.serializeAttachment(attachment) + // Tagged accept (SYNC_TAG) + plain attachment (no `__pk`): the two + // independent discriminators that keep host and sync sockets apart. + this.ctx.acceptWebSocket(server, [SYNC_TAG]) + this.#liveWs.add(server) + + return new Response(null, { status: 101, webSocket: client }) + } + + /** True iff `ws` is a tddc sync socket. On a bare DO tddc owns every socket + * (no host to share with), so all sockets are sync — which also keeps a + * legacy untagged 0.4.0 socket working after the mixin upgrade. Over any + * other base, only SYNC_TAG sockets are ours; the rest delegate to `super`. */ + #isSyncSocket(ws: WebSocket): boolean { + return this.#isBareDO || this.ctx.getTags(ws).includes(SYNC_TAG) + } + + override async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise { + if (!this.#isSyncSocket(ws)) { + const base = super.webSocketMessage as + | ((ws: WebSocket, m: string | ArrayBuffer) => void | Promise) + | undefined + if (base) await base.call(this, ws, message) + return + } + // "ping"/"pong" are handled by the auto-response and never arrive here. + + // Reject oversize frames before decode (ADR-0012): mirrors the + // undecodable-frame stance — drop + log, no reply, no crash. + const byteLen = typeof message === "string" ? message.length : message.byteLength + if (byteLen > this.maxFrameBytes) { + console.error(`oversize frame dropped (${byteLen} bytes > maxFrameBytes ${this.maxFrameBytes})`) + return + } + + let decoded: unknown + try { + decoded = this.#codec.decode(message) + } catch { + return // ignore undecodable frames + } + + // Shape-guard after decode (ADR-0012): a frame that decodes but has the + // wrong structure is dropped + logged. The guard runs BEFORE any SQL + // binding so no arbitrary decoded value reaches lookupTx or sql.exec. + if (!this.#wellFormed(decoded)) { + // Safe stringify: decoded may contain bigints (MessagePack useBigInt64); + // JSON.stringify throws on bigint — use a replacer to avoid crashing the + // logging itself. + let summary: string + try { + summary = JSON.stringify(decoded, (_k, v) => (typeof v === "bigint" ? String(v) : v)) + } catch { + summary = String(decoded) + } + console.error("malformed frame dropped", summary) + return + } + + await this.#dispatch(ws, decoded) + } + + override webSocketClose(ws: WebSocket, code?: number, reason?: string, wasClean?: boolean): void { + if (!this.#isSyncSocket(ws)) { + const base = super.webSocketClose as + | ((ws: WebSocket, code: number, reason: string, wasClean: boolean) => void) + | undefined + if (base) base.call(this, ws, code ?? 1000, reason ?? "", wasClean ?? false) + return + } + this.#subs.removeAll(ws) + this.#liveWs.delete(ws) + } + + override webSocketError(ws: WebSocket, error?: unknown): void { + if (!this.#isSyncSocket(ws)) { + const base = super.webSocketError as ((ws: WebSocket, error: unknown) => void) | undefined + if (base) base.call(this, ws, error) + return + } + this.#subs.removeAll(ws) + this.#liveWs.delete(ws) + } + + /** Shape-guard: returns true iff `v` is a structurally valid ClientFrame. + * (ADR-0012) Runs after decode, before any SQL binding. + * + * Optional fields treat null == absent (the client transport serialises + * absent fields as null in MessagePack rather than omitting them). */ + #wellFormed(v: unknown): v is ClientFrame { + if (v === null || typeof v !== "object") return false + const f = v as Record + const t = f["t"] + if (typeof t !== "string") return false + + const isNonEmptyString = (x: unknown): x is string => typeof x === "string" && x.length > 0 + /** null is treated as absent for optional fields */ + const absent = (x: unknown): boolean => x === undefined || x === null + + switch (t) { + case "sub": + return ( + isNonEmptyString(f["subId"]) && + isNonEmptyString(f["collection"]) && + (absent(f["since"]) || typeof f["since"] === "string") && + (absent(f["limit"]) || typeof f["limit"] === "number") && + (absent(f["offset"]) || typeof f["offset"] === "number") + ) + case "unsub": + return typeof f["subId"] === "string" + case "mut": { + if (!isNonEmptyString(f["txId"]) || !isNonEmptyString(f["collection"])) return false + if (!Array.isArray(f["ops"]) || f["ops"].length === 0) return false + const validOpTypes = new Set(["insert", "update", "delete"]) + for (const op of f["ops"] as Array) { + if (op === null || typeof op !== "object") return false + const o = op as Record + if (!validOpTypes.has(o["type"] as string)) return false + if (typeof o["key"] !== "string") return false + if (!absent(o["cols"]) && (typeof o["cols"] !== "object" || Array.isArray(o["cols"]))) return false + } + return true + } + case "call": + return isNonEmptyString(f["txId"]) && isNonEmptyString(f["name"]) + case "fetch": + return ( + isNonEmptyString(f["fetchId"]) && + isNonEmptyString(f["collection"]) && + (absent(f["cursor"]) || typeof f["cursor"] === "object") + ) + default: + return false + } + } + + async #dispatch(ws: WebSocket, frame: ClientFrame): Promise { + switch (frame.t) { + case "sub": + return this.#handleSub(ws, frame) + case "unsub": + this.#subs.remove(ws, frame.subId) + return + case "mut": + return this.#handleMut(ws, frame) + case "call": + return this.#handleCall(ws, frame) + case "fetch": + return this.#handleFetch(ws, frame) + } + } + + /** One-shot paginated page fetch — a subset snapshot, NO live registration. + * Used by the client for cursor load-more; the window's live deltas already + * flow via the `sub` on the query's `where`. + * + * The frame mirrors @tanstack/db's `LoadSubsetOptions` (ADR-0005): a base + * `where` plus a raw `cursor` (whereFrom/whereCurrent, which exclude the base). We compose + * `base AND whereCurrent` (ties, unbounded) and `base AND whereFrom` (next + * page, bounded by `limit`) as TWO SELECTs in ONE handler turn: synchronous + * SQLite, no `await` between them, so both observe the same database at one + * `seq`. The page therefore slots into the delta stream at a single position + * — a concurrent mutation is either reflected in it or arrives as a delta + * AFTER it, never split across the two reads (ADR-0003). */ + #handleFetch(ws: WebSocket, frame: Extract): void { + const coll = this.#registry.collections.get(frame.collection) + if (!coll) { + this.#send(ws, { t: "page", fetchId: frame.fetchId, rows: [], seq: "0" }) + return + } + // A cursor must carry BOTH halves (TanStack's CursorExpressions always + // does). A missing `whereCurrent` would otherwise compose to an empty + // predicate and run the ties SELECT unbounded — a silent full-table scan, + // which the operator floor exists to forbid. Reject loudly instead. + if (frame.cursor != null && (frame.cursor.whereCurrent == null || frame.cursor.whereFrom == null)) { + console.error(`fetch '${frame.fetchId}' on '${frame.collection}' rejected: malformed cursor`) + this.#send(ws, { t: "page", fetchId: frame.fetchId, rows: [], seq: String(currentSeq(this.#sql)) }) + return + } + const seq = String(currentSeq(this.#sql)) + try { + const rows: Array = [] + // Cursor present: ties first (base AND whereCurrent, unbounded boundary + // set), then the bounded next page (base AND whereFrom). The cursor + // expressions arrive raw — excluding the base — so we compose them here. + // No cursor: a plain bounded `where` read. + if (frame.cursor != null) { + const tq = compileSubsetQuery(frame.collection, { + where: andPredicates(frame.where, frame.cursor.whereCurrent), + orderBy: frame.orderBy, + }) + rows.push(...Array.from(this.#sql.exec(tq.sql, ...tq.params))) + } + const nextWhere = frame.cursor != null ? andPredicates(frame.where, frame.cursor.whereFrom) : frame.where + const nq = compileSubsetQuery(frame.collection, { + where: nextWhere, + orderBy: frame.orderBy, + limit: frame.limit, + }) + rows.push(...Array.from(this.#sql.exec(nq.sql, ...nq.params))) + this.#send(ws, { t: "page", fetchId: frame.fetchId, rows, seq }) + } catch (e) { + if (e instanceof UnsupportedPredicateError) { + console.error(`fetch '${frame.fetchId}' on '${frame.collection}' rejected: ${e.message}`) + this.#send(ws, { t: "page", fetchId: frame.fetchId, rows: [], seq }) + return + } + throw e + } + } + + /** + * Apply a mutation atomically and confirm on the single ordered stream. + * + * Order is the load-bearing invariant (ADR-0002 C1): this connection's + * matched deltas are flushed BEFORE its `committed` frame, so the client's + * single cursor only ever advances over a contiguous applied prefix and the + * optimistic overlay is never dropped before the authoritative row lands. + */ + async #handleMut(ws: WebSocket, f: Extract): Promise { + // Inbound limit: reject over-length batches without applying anything + // (ADR-0012). Reject-don't-truncate: a partial apply silently drops writes. + if (f.ops.length > this.maxOpsPerMutation) { + return this.#rejectTx(ws, f.txId, `mutation exceeds maxOpsPerMutation (${this.maxOpsPerMutation})`, "LIMIT_EXCEEDED") + } + + const seen = lookupTx(this.#sql, f.txId) + if (seen) return this.#replayReceipt(ws, f.txId, seen) + + const user = this.#userFor(ws) + + // Authorize every op BEFORE the transaction (may be async). + try { + for (const op of f.ops) { + const def = this.#registry.mutations.get(`${f.collection}:${op.type}`) + if (!def) throw new Error(`no mutation handler for '${f.collection}:${op.type}'`) + if (def.authorize) await def.authorize({ user, op, sql: this.#sql, env: this.env }) + } + } catch (e) { + // authorize and validation errors surface to the client: a schema failure + // carries a VALIDATION code, an authz "throw to deny" keeps its message. The + // execute catch below stays sanitized. + if (e instanceof ValidationError) return this.#rejectTx(ws, f.txId, e.message, "VALIDATION") + return this.#rejectTx(ws, f.txId, errorMessage(e)) + } + + // Apply all ops in one synchronous transaction (atomic with the trigger + // rows). A handler that returns a Promise is a programming error. + let commitSeq: string + try { + this.ctx.storage.transactionSync(() => { + for (const op of f.ops) { + const def = this.#registry.mutations.get(`${f.collection}:${op.type}`)! + const result = def.execute({ user, op, sql: this.#sql, env: this.env }) as unknown + if (result !== undefined && typeof (result as PromiseLike).then === "function") { + // `execute` runs inside transactionSync, which cannot await; an async + // execute also can't be atomic with its CDC rows. Do async work in + // `authorize` (pre-tx), `afterCommit` (post-commit), or a command. + throw new Error( + `mutation '${f.collection}:${op.type}' execute must be synchronous — do async work in authorize, afterCommit, or a command`, + ) + } + } + }) + commitSeq = String(currentSeq(this.#sql)) + } catch (e) { + // Log full detail server-side; send only a generic message to the client + // (ADR-0012). SQLite constraint strings, column names, and programming- + // error text are internal detail — not client API surface. The authorize + // catch above is intentionally kept user-facing (README: "throw to deny"). + console.error(`mutation '${f.collection}' execute failed: ${errorMessage(e)}`) + return this.#rejectTx(ws, f.txId, "mutation failed", "EXECUTE_FAILED") + } + + recordTx(this.#sql, f.txId, true, commitSeq, null, null) + // Enqueue deltas for all subscribers, then flush THIS socket before its + // receipt (C1) so its deltas land first. Other subscribers flush on the + // coalescer tick. + this.#drainAndBroadcast() + this.#broadcaster.flushOne(ws) + this.#send(ws, { t: "committed", txId: f.txId, seq: commitSeq }) + + // Fire-and-forget post-commit hooks AFTER the receipt — never on the + // client's critical path. Each runs under `waitUntil` (keeps the DO alive + // until it settles) and is isolated: a throw is logged and dropped, leaving + // the committed mutation untouched. The hook owns its own idempotency + // (ADR-0004); the library guarantees only "runs once per commit, off-path". + for (const op of f.ops) { + const after = this.#registry.mutations.get(`${f.collection}:${op.type}`)?.afterCommit + if (!after) continue + this.ctx.waitUntil( + (async () => { + try { + await after({ user, op, sql: this.#sql, env: this.env }) + } catch (e) { + console.error(`afterCommit '${f.collection}:${op.type}' failed: ${errorMessage(e)}`) + } + })(), + ) + } + } + + /** Run a named command (outside any transaction) and confirm with its result. */ + async #handleCall(ws: WebSocket, f: Extract): Promise { + const seen = lookupTx(this.#sql, f.txId) + if (seen) return this.#replayReceipt(ws, f.txId, seen) + + const def = this.#registry.commands.get(f.name) + if (!def) return this.#rejectTx(ws, f.txId, `unknown command '${f.name}'`, "UNKNOWN_COMMAND") + + const user = this.#userFor(ws) + // authorize and validation errors surface like a mutation's: a schema failure + // carries a VALIDATION code, an authz "throw to deny" keeps its message. This + // matches mutation surfacing, revising ADR-0012 D3 (which sanitized a + // command's authorize too). + try { + if (def.authorize) await def.authorize({ user, args: f.args, sql: this.#sql, env: this.env }) + } catch (e) { + if (e instanceof ValidationError) return this.#rejectTx(ws, f.txId, e.message, "VALIDATION") + return this.#rejectTx(ws, f.txId, errorMessage(e)) + } + // execute runs arbitrary, often async code, so its errors are sanitized like a + // mutation's execute — internal detail never leaks. + let result: unknown + try { + result = await def.execute({ user, args: f.args, sql: this.#sql, env: this.env }) + } catch (e) { + console.error(`command '${f.name}' execute failed: ${errorMessage(e)}`) + return this.#rejectTx(ws, f.txId, "command failed", "EXECUTE_FAILED") + } + + // Serialize the result for dedup replay BEFORE recording success. A + // non-serializable result can't be replayed, so record an error rather + // than risk re-running the command's side effects on retry. + let stored: string | null + try { + stored = encodeResult(result) + } catch (e) { + return this.#rejectTx(ws, f.txId, `non-serializable command result: ${errorMessage(e)}`, "NON_SERIALIZABLE") + } + + const commitSeq = String(currentSeq(this.#sql)) + recordTx(this.#sql, f.txId, true, commitSeq, null, stored) + this.#drainAndBroadcast() + this.#broadcaster.flushOne(ws) + this.#send(ws, { t: "committed", txId: f.txId, seq: commitSeq, result }) + } + + #rejectTx(ws: WebSocket, txId: string, message: string, code?: string): void { + recordTx(this.#sql, txId, false, null, message, null) + this.#send(ws, { t: "rejected", txId, error: code ? { code, message } : { message } }) + } + + #replayReceipt(ws: WebSocket, txId: string, seen: SeenTx): void { + if (seen.ok) { + this.#send(ws, { t: "committed", txId, seq: seen.cursor ?? "0", result: decodeResult(seen.result) }) + } else { + this.#send(ws, { t: "rejected", txId, error: { message: seen.error ?? "unknown" } }) + } + } + + /** + * Apply a SERVER-ORIGINATED write and broadcast it to connected clients + * (ADR-0006). The home for writes outside the client mutation flow — an agent + * inserting a row, a webhook, a cron/`alarm` job, an admin edit, a bulk seed. + * + * `fn` runs inside `transactionSync` (atomic, and the same synchronous + * constraint mutations live under) and may return a value; its CDC is then + * drained and broadcast on the next coalescer tick. A thenable return is + * rejected (and rolls back): any async work belongs BEFORE the call. + */ + #runSyncedWrite(fn: (sql: SqlStorage) => T): T { + let result: T + this.ctx.storage.transactionSync(() => { + result = fn(this.#sql) + if (result != null && typeof (result as unknown as PromiseLike).then === "function") { + throw new Error("runSyncedWrite fn must be synchronous (it returned a thenable)") + } + }) + this.#drainAndBroadcast() + return result! + } + + /** + * Drain `_sync_changes` from the last broadcast watermark, fan out one `d` + * per changed key to each subscriber of the affected collection, then a + * single `uptodate` boundary per touched socket. Multiple changes to a key + * within the drain collapse to the latest op. + */ + #drainAndBroadcast(): void { + const sql = this.#sql + const last = getDrainCursor(sql) + const changes = readChangesSince(sql, last) + if (changes.length === 0) return + const cursor = String(changes[changes.length - 1]!.seq) + + const byTable = new Map>() + for (const c of changes) { + let arr = byTable.get(c.tbl) + if (!arr) { + arr = [] + byTable.set(c.tbl, arr) + } + arr.push(c) + } + + for (const [tbl, tableChanges] of byTable) { + const coll = this.#registry.collections.get(tbl) + if (!coll) continue + const latest = new Map() + for (const c of tableChanges) latest.set(c.key, c) + const liveKeys = [...latest.values()].filter((c) => c.op !== "delete").map((c) => c.key) + const hydrated = hydrateRows(sql, tbl, coll.pk, liveKeys) + + // Enqueue into the coalescer; it flushes one `d` per surviving key plus a + // single `uptodate` boundary per socket (on the tick, or via flushOne). + for (const { ws, sub } of this.#subs.forCollection(tbl)) { + for (const [key, change] of latest) { + const row = hydrated.get(key) + // Always-emit rule (no before-image, ADR-0002 C4): a key that is + // deleted, gone, or no longer matches this sub's predicate -> a + // synthetic delete (idempotent; move-out). A matching live row -> + // its current state with the actual op (move-in via update upserts + // on the client — verified). Predicate is always-true when unfiltered. + if (change.op === "delete" || !row || !sub.predicate(row)) { + this.#broadcaster.enqueue(ws, { subId: sub.subId, key, op: "delete" }, cursor) + } else { + // Full row as the partial patch; column-level diffs arrive later. + this.#broadcaster.enqueue(ws, { subId: sub.subId, key, op: change.op, cols: row }, cursor) + } + } + } + } + + setDrainCursor(sql, changes[changes.length - 1]!.seq) + this.#maybeCompact() + } + + /** + * Opportunistic GC: every `compactionEvery` drained mutations, collapse the + * change log to latest-op-per-key and sweep expired dedup entries. Deferred + * via `ctx.waitUntil` so it rides just after a burst of work — it never + * blocks a mutation's response, and (unlike an alarm) never wakes an idle DO. + */ + #maybeCompact(): void { + if (++this.#writesSinceCompaction < this.compactionEvery) return + this.#writesSinceCompaction = 0 + this.ctx.waitUntil( + (async (): Promise => { + compactChanges(this.#sql) + pruneChanges(this.#sql, this.changelogRetentionMs, Date.now()) + sweepDedup(this.#sql, this.dedupRetentionMs, Date.now()) + })(), + ) + } + + /** Full-collection subscribe: emit every current row as a snapshot, then a + * boundary. */ + #handleSub(ws: WebSocket, frame: Extract): void { + const coll = this.#registry.collections.get(frame.collection) + if (!coll) { + // Unknown collection: drop the subscriber's view. Richer sub-error + // signalling is deferred; for now reset is the honest minimum. + this.#send(ws, { t: "reset", sub: frame.subId }) + return + } + + // Per-socket subscription cap (ADR-0012). A re-sub on an existing subId + // replaces the old entry (SubscriptionRegistry.add semantics) — count + // only new subIds against the cap. + const existingCount = this.#subs.countFor(ws) + const existingSub = this.#subs.forWs(ws).find((s) => s.subId === frame.subId) + if (!existingSub && existingCount >= this.maxSubsPerSocket) { + console.error(`sub '${frame.subId}' refused: maxSubsPerSocket (${this.maxSubsPerSocket}) reached`) + this.#send(ws, { t: "reset", sub: frame.subId }) + return + } + // Lower where/orderBy/limit/offset into SQLite. An un-lowerable predicate + // (outside the supported floor) is rejected, not silently full-scanned. + let query: { sql: string; params: Array } + try { + query = compileSubsetQuery(frame.collection, { + where: frame.where, + orderBy: frame.orderBy, + limit: frame.limit, + offset: frame.offset, + }) + } catch (e) { + if (e instanceof UnsupportedPredicateError) { + console.error(`sub '${frame.subId}' on '${frame.collection}' rejected: ${e.message}`) + this.#send(ws, { t: "reset", sub: frame.subId }) + return + } + throw e + } + + // C1′ (ADR-0011, generalizing ADR-0002 C1): what follows is a synchronous + // cursor-advancing emission — a snapshot's `snap-end` or a catch-up's + // `uptodate` carries the CURRENT seq, which may include changes whose + // deltas are still buffered in the coalescer for this socket. Flush them + // first, or the client's cursor claims a seq it never applied and a drop + // before the tick loses the write (reconnect resumes past it). + this.#broadcaster.flushOne(ws) + + // Registering compiles the predicate in @tanstack/db's evaluator. If the + // predicate is outside the JS floor (e.g. an operator the SQL floor somehow + // let through), that throws UnsupportedPredicateError — reject with `reset` + // rather than letting it escape uncaught and hang the client (ADR-0013). + let sub: Sub + try { + sub = this.#subs.add(ws, frame.subId, frame.collection, frame.where) + } catch (e) { + if (e instanceof UnsupportedPredicateError) { + console.error(`sub '${frame.subId}' on '${frame.collection}' rejected: ${e.message}`) + this.#send(ws, { t: "reset", sub: frame.subId }) + return + } + throw e + } + const seq = String(currentSeq(this.#sql)) + + // Reconnect catch-up: a `since` cursor asks for changes after that point + // rather than a fresh snapshot. Serve a windowed delta while the change log + // still reaches back that far; otherwise fall back to reset + snapshot. + // + // The floor is `minChangeSeq` — no persisted watermark needed (ADR-0009). + const since = frame.since != null ? Number(frame.since) : 0 + if (since > 0) { + const floor = minChangeSeq(this.#sql) + if (floor !== 0 && since >= floor - 1) { + this.#emitCatchUp(ws, sub, coll, since, seq) + return + } + this.#send(ws, { t: "reset", sub: frame.subId }) + } + + const rows = Array.from(this.#sql.exec(query.sql, ...query.params)) as Array> + for (const row of rows) { + this.#send(ws, { t: "snap", sub: frame.subId, key: row[coll.pk], row, seq }) + } + this.#send(ws, { t: "snap-end", sub: frame.subId, seq }) + } + + /** Windowed catch-up: the latest op per changed key since `since`, resolved + * through the sub's predicate (move-in/out), then an `uptodate` boundary. */ + #emitCatchUp( + ws: WebSocket, + sub: { subId: string; predicate: (row: Record) => boolean }, + coll: { table: string; pk: string }, + since: number, + seq: string, + ): void { + const changes = readChangesSinceFor(this.#sql, coll.table, since) + const latest = new Map() + for (const c of changes) latest.set(c.key, c) + const liveKeys = [...latest.values()].filter((c) => c.op !== "delete").map((c) => c.key) + const hydrated = hydrateRows(this.#sql, coll.table, coll.pk, liveKeys) + for (const [key, change] of latest) { + const row = hydrated.get(key) + if (change.op === "delete" || !row || !sub.predicate(row)) { + this.#send(ws, { t: "d", sub: sub.subId, key, op: "delete", seq }) + } else { + this.#send(ws, { t: "d", sub: sub.subId, key, op: change.op, cols: row, seq }) + } + } + this.#send(ws, { t: "uptodate", seq }) + } + + /** Encode and send a server frame on one socket. */ + #send(ws: WebSocket, frame: ServerFrame): void { + ws.send(this.#codec.encode(frame)) + } + + /** The attachment bound at upgrade, surviving hibernation. */ + #userFor(ws: WebSocket): TUser { + return ws.deserializeAttachment() as TUser + } + } + + // Explicit, NAMEABLE return type — required for declaration emit (a raw + // `class extends Base` would emit an anonymous class type whose inherited + // protected `ctx`/`env` trip TS4094). A single `(...args: any[])` construct + // signature (so a mixed base's own constructor params don't leak into a + // generic subclass's `super(...)`) yielding the host instance intersected + // with the sync surface. The protected tuning knobs stay protected fields at + // runtime; `SyncDurableObject` re-declares them for override typing. + return SyncableMixin as unknown as abstract new ( + ...args: Array + ) => InstanceType & SyncMixin + } +} + +function errorMessage(e: unknown): string { + return e instanceof Error ? e.message : String(e) +} diff --git a/src/server/sql-compiler.ts b/src/server/sql-compiler.ts index 42f216e..0537d8f 100644 --- a/src/server/sql-compiler.ts +++ b/src/server/sql-compiler.ts @@ -181,7 +181,20 @@ export function compileSubsetQuery(tbl: string, opts: SubsetQuery): { sql: strin } const orderBy = compileOrderBy(opts.orderBy) - if (orderBy) sql += ` ORDER BY ${orderBy}` + if (orderBy) { + sql += ` ORDER BY ${orderBy}` + } else { + // No client-specified order: default to `rowid` so a subset read is + // deterministic instead of an accident of SQLite's query plan (a bare scan + // happens to walk rowid order, but a WHERE clause touching the pk can make + // the planner pick the pk's autoindex instead, returning pk-sorted rows). + // `rowid` also matches insertion order among currently-live rows: a rowid + // table assigns each new row 1 + the current max, which is monotonic + // across inserts regardless of intervening deletes. Every synced table + // qualifies — `assertSyncCompatible` (ADR-0007, D9) forbids an `INTEGER + // PRIMARY KEY` pk, so the real rowid is always intact underneath. + sql += ` ORDER BY rowid` + } if (opts.limit != null) { sql += ` LIMIT ?` diff --git a/src/server/sync-do.ts b/src/server/sync-do.ts index 6bb3d1a..353c1de 100644 --- a/src/server/sync-do.ts +++ b/src/server/sync-do.ts @@ -1,724 +1,98 @@ -// SyncDurableObject — hibernating-WebSocket base class (ADR-0001 D13). +// SyncDurableObject — the trivial application of the `Syncable` mixin over a +// bare `DurableObject` (ADR-0001 D13, ADR-0015). // -// Provides the lifecycle every sync-enabled DO shares: -// - WebSocket upgrade with a subclass-typed attachment (parseAttachment), -// bound via serializeAttachment so identity survives hibernation. -// - ctx.acceptWebSocket (NOT addEventListener) for hibernation support. -// - "ping"/"pong" auto-response registered once in the constructor — does not -// wake or bill the DO. -// - inbound frame decode (binary/JSON) dispatched to an onFrame hook. -// - lazy schema + trigger init from the collection registry. +// The sync machinery now lives in `Syncable(Base)` (mixin.ts). This module keeps +// `SyncDurableObject` as the zero-config base class it has always been, so every +// existing `extends SyncDurableObject` keeps compiling and behaving +// identically to 0.4.0: sockets are the mixin's sync sockets, and the two +// DO-global side effects (`ping/pong` auto-response and `PRAGMA +// case_sensitive_like = ON`) default ON because the base IS `DurableObject`. // -// Frame handling (sub/mut/call -> snap/d/committed/...) arrives in M3; this -// milestone establishes the lifecycle and the wire decode/encode path. +// The legacy protected surface (`this.sql`, `this.registerSync`, +// `this.runSyncedWrite`, an overridable `parseAttachment`) is re-exposed here as +// thin aliases over the `this.sync` facade. `this.sql` is safe on this base +// because a bare `DurableObject` defines no `sql` member to shadow — on a +// non-trivial host reach `this.ctx.storage.sql` directly (ADR-0015). import { DurableObject } from "cloudflare:workers" -import type { SqlStorage, SqlStorageValue } from "@cloudflare/workers-types" -import { createFrameCodec, type FrameCodec } from "../wire/frame-codec.ts" -import type { ClientFrame, ServerFrame } from "../wire/frames.ts" -import { - compactChanges, - currentSeq, - ensureTriggers, - getDrainCursor, - hydrateRows, - initSchema, - minChangeSeq, - pruneChanges, - readChangesSince, - readChangesSinceFor, - setDrainCursor, -} from "./changes.ts" -import { Broadcaster } from "./broadcast.ts" -import { decodeResult, encodeResult, lookupTx, recordTx, type SeenTx, sweepDedup } from "./dedup.ts" -import { compileSchema, type CompiledSync, type SyncSchema, ValidationError } from "./registry.ts" -import { andPredicates, compileSubsetQuery, UnsupportedPredicateError } from "./sql-compiler.ts" -import { SubscriptionRegistry, type Sub } from "./subscriptions.ts" - -export abstract class SyncDurableObject extends DurableObject { - /** Set by `registerSync` — the compiled dispatch tables this DO serves. */ - #registry: CompiledSync | undefined - - /** The compiled schema. Throws if `registerSync` hasn't run yet (ADR-0007). */ - protected get registry(): CompiledSync { - if (!this.#registry) { - throw new Error( - "sync not registered — call this.registerSync(registry) in your constructor's " + - "blockConcurrencyWhile, after creating your tables", - ) - } - return this.#registry - } - - /** Wire codec. Binary MessagePack by default; override for a JSON transport. */ - protected readonly codec: FrameCodec = createFrameCodec() - - protected readonly subs = new SubscriptionRegistry() - /** Egress coalescer tick (ms) — the single user-perceived-latency knob. */ - protected readonly tickMs: number = 50 - /** Compact the change log every this-many drained mutations (not on a timer — - * an alarm would wake idle DOs; this rides recent work). */ - protected readonly compactionEvery: number = 200 - /** Age bound for `_sync_changes` (ADR-0009). Changes older than this are - * pruned during compaction; a reconnect older than the surviving floor gets a - * full re-snapshot instead of a delta. `null` disables retention (the log - * reverts to compaction-only, unbounded by age). Sibling to - * `dedupRetentionMs`. Default 2 days. */ - protected readonly changelogRetentionMs: number | null = 172_800_000 - /** Dedup retention window (ms), independent of changelog retention (C5). */ - protected readonly dedupRetentionMs: number = 3_600_000 - /** Maximum ops in a single `mut` frame (ADR-0012). Reject-don't-truncate: - * a partial apply would silently drop client writes. Override in subclasses - * to tune for your workload. */ - protected readonly maxOpsPerMutation: number = 128 - /** Maximum concurrent subscriptions per socket (ADR-0012). Over-limit subs - * are refused with a `reset` frame so legitimate earlier subs keep flowing. - * Override to tune for your data model. */ - protected readonly maxSubsPerSocket: number = 256 - /** Maximum inbound frame size in bytes (ADR-0012). Cloudflare's own cap is - * ~1 MiB; this makes the bound explicit, testable, and overrideable. - * Oversize frames are dropped + logged without closing the socket (mirrors - * the undecodable-frame stance). */ - protected readonly maxFrameBytes: number = 1_048_576 - private writesSinceCompaction = 0 - protected readonly broadcaster: Broadcaster - private readonly liveWs = new Set() +import type { SqlStorage } from "@cloudflare/workers-types" +import { Syncable } from "./mixin.ts" +import type { CompiledSync, SyncSchema } from "./registry.ts" + +// Generics are erased at runtime, so the base VALUE is the plain application of +// the factory over DurableObject; Env/TUser are re-exposed through this class's +// own typed shims below (a base-class expression cannot reference a class's own +// type parameters — TS2562). The factory already exposes a `(...args: any[])` +// construct signature, so this generic subclass can forward `super(ctx, env)` +// for any `Env`. +const SyncableBase = Syncable()(DurableObject) + +export abstract class SyncDurableObject extends SyncableBase { + // The mixin's return type hides the tuning knobs (they must not widen the + // host-collision surface). They ARE protected fields on the mixin at runtime; + // re-declare them here (ambient — no runtime field, no shadow) so existing + // `protected override readonly tickMs = …` subclasses keep compiling. Behind a + // non-DO host, tune with `this.sync.configure` instead. + declare protected readonly tickMs: number + declare protected readonly compactionEvery: number + declare protected readonly changelogRetentionMs: number | null + declare protected readonly dedupRetentionMs: number + declare protected readonly maxOpsPerMutation: number + declare protected readonly maxSubsPerSocket: number + declare protected readonly maxFrameBytes: number constructor(ctx: ConstructorParameters[0], env: Env) { super(ctx, env) - // Auto-pong via the runtime: survives hibernation, no per-message billing. - this.ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong")) - // Make SQLite LIKE case-sensitive so the SQL snapshot path matches - // @tanstack/db's case-sensitive `like` evaluator on the delta path — the - // single source of truth for filtered-subscription membership (ADR-0013). - // Connection-scoped pragma; re-applied on every instantiation, including a - // hibernation wake (same lifecycle as the auto-response registration above). - this.sql.exec("PRAGMA case_sensitive_like = ON") - // Restore the live-socket set after a hibernation wake. - for (const ws of this.ctx.getWebSockets()) this.liveWs.add(ws) - this.broadcaster = new Broadcaster((ws, frame) => this.send(ws, frame), this.tickMs) - this.broadcaster.start(() => this.liveWs) + // Bridge the overridable protected `parseAttachment` into the facade so an + // override on a subclass is honoured at upgrade time (resolved dynamically). + this.sync.configure({ parseAttachment: (req) => this.parseAttachment(req) }) } + /** SQLite handle. Safe on this base (a bare `DurableObject` defines no `sql` + * member); on a non-trivial host reach `this.ctx.storage.sql` directly to + * avoid shadowing the host's `sql` tagged template (ADR-0015). */ protected get sql(): SqlStorage { return this.ctx.storage.sql } /** - * Wire collections for sync: validate each table is sync-compatible (ADR-0007) - * and reconcile its CDC triggers — install the registered set, drop triggers - * for any collection no longer registered (ADR-0008). The author owns table - * creation; call this AFTER the tables exist — typically in your constructor's - * `blockConcurrencyWhile`, after migrating. Idempotent; re-callable to update - * the whole trigger state when the registry changes. + * Wire collections for sync (ADR-0007). Legacy alias for + * `this.sync.registerSync`. Call in your constructor's `blockConcurrencyWhile`, + * after your tables exist. */ protected registerSync(schema: SyncSchema): void { - const compiled = compileSchema(schema) - initSchema(this.sql) - ensureTriggers(this.sql, compiled.collections.values()) - this.#registry = compiled - } - - /** - * Validate the upgrade and produce the attachment bound to the WebSocket - * (available as `userFor(ws)` in handlers). Override to read a Worker-forged - * claims header and/or reject by throwing a `Response`. Default: no identity. - */ - protected parseAttachment(_req: Request): TUser | Promise { - return undefined as TUser - } - - override async fetch(req: Request): Promise { - if (req.headers.get("Upgrade") !== "websocket") { - return new Response("expected websocket upgrade", { status: 426 }) - } - - let attachment: TUser - try { - attachment = await this.parseAttachment(req) - } catch (e) { - if (e instanceof Response) return e - return new Response("unauthorized", { status: 401 }) - } - - const pair = new WebSocketPair() - const client = pair[0] - const server = pair[1] - server.serializeAttachment(attachment) - this.ctx.acceptWebSocket(server) - this.liveWs.add(server) - - return new Response(null, { status: 101, webSocket: client }) - } - - override async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): Promise { - // "ping"/"pong" are handled by the auto-response and never arrive here. - - // Reject oversize frames before decode (ADR-0012): mirrors the - // undecodable-frame stance — drop + log, no reply, no crash. - const byteLen = typeof message === "string" ? message.length : message.byteLength - if (byteLen > this.maxFrameBytes) { - console.error(`oversize frame dropped (${byteLen} bytes > maxFrameBytes ${this.maxFrameBytes})`) - return - } - - let decoded: unknown - try { - decoded = this.codec.decode(message) - } catch { - return // ignore undecodable frames - } - - // Shape-guard after decode (ADR-0012): a frame that decodes but has the - // wrong structure is dropped + logged. The guard runs BEFORE any SQL - // binding so no arbitrary decoded value reaches lookupTx or sql.exec. - if (!this.wellFormed(decoded)) { - // Safe stringify: decoded may contain bigints (MessagePack useBigInt64); - // JSON.stringify throws on bigint — use a replacer to avoid crashing the - // logging itself. - let summary: string - try { - summary = JSON.stringify(decoded, (_k, v) => (typeof v === "bigint" ? String(v) : v)) - } catch { - summary = String(decoded) - } - console.error("malformed frame dropped", summary) - return - } - - await this.dispatch(ws, decoded) - } - - override webSocketClose(ws: WebSocket): void { - this.subs.removeAll(ws) - this.liveWs.delete(ws) - } - - override webSocketError(ws: WebSocket): void { - this.subs.removeAll(ws) - this.liveWs.delete(ws) - } - - /** Shape-guard: returns true iff `v` is a structurally valid ClientFrame. - * (ADR-0012) Runs after decode, before any SQL binding — ensures no - * arbitrary decoded value reaches lookupTx or sql.exec. - * - * Optional fields treat null == absent (the client transport serialises - * absent fields as null in MessagePack rather than omitting them). */ - private wellFormed(v: unknown): v is ClientFrame { - if (v === null || typeof v !== "object") return false - const f = v as Record - const t = f["t"] - if (typeof t !== "string") return false - - const isNonEmptyString = (x: unknown): x is string => typeof x === "string" && x.length > 0 - /** null is treated as absent for optional fields */ - const absent = (x: unknown): boolean => x === undefined || x === null - - switch (t) { - case "sub": - return ( - isNonEmptyString(f["subId"]) && - isNonEmptyString(f["collection"]) && - (absent(f["since"]) || typeof f["since"] === "string") && - (absent(f["limit"]) || typeof f["limit"] === "number") && - (absent(f["offset"]) || typeof f["offset"] === "number") - ) - case "unsub": - return typeof f["subId"] === "string" - case "mut": { - if (!isNonEmptyString(f["txId"]) || !isNonEmptyString(f["collection"])) return false - if (!Array.isArray(f["ops"]) || f["ops"].length === 0) return false - const validOpTypes = new Set(["insert", "update", "delete"]) - for (const op of f["ops"] as Array) { - if (op === null || typeof op !== "object") return false - const o = op as Record - if (!validOpTypes.has(o["type"] as string)) return false - if (typeof o["key"] !== "string") return false - if (!absent(o["cols"]) && (typeof o["cols"] !== "object" || Array.isArray(o["cols"]))) return false - } - return true - } - case "call": - return isNonEmptyString(f["txId"]) && isNonEmptyString(f["name"]) - case "fetch": - return ( - isNonEmptyString(f["fetchId"]) && - isNonEmptyString(f["collection"]) && - (absent(f["cursor"]) || typeof f["cursor"] === "object") - ) - default: - return false - } - } - - private async dispatch(ws: WebSocket, frame: ClientFrame): Promise { - switch (frame.t) { - case "sub": - return this.handleSub(ws, frame) - case "unsub": - this.subs.remove(ws, frame.subId) - return - case "mut": - return this.handleMut(ws, frame) - case "call": - return this.handleCall(ws, frame) - case "fetch": - return this.handleFetch(ws, frame) - } - } - - /** One-shot paginated page fetch — a subset snapshot, NO live registration. - * Used by the client for cursor load-more; the window's live deltas already - * flow via the `sub` on the query's `where`. - * - * The frame mirrors @tanstack/db's `LoadSubsetOptions` (ADR-0005): a base - * `where` plus a raw `cursor` (whereFrom/whereCurrent, which exclude the base). We compose - * `base AND whereCurrent` (ties, unbounded) and `base AND whereFrom` (next - * page, bounded by `limit`) as TWO SELECTs in ONE handler turn: synchronous - * SQLite, no `await` between them, so both observe the same database at one - * `seq`. The page therefore slots into the delta stream at a single position - * — a concurrent mutation is either reflected in it or arrives as a delta - * AFTER it, never split across the two reads (ADR-0003). */ - private handleFetch(ws: WebSocket, frame: Extract): void { - const coll = this.registry.collections.get(frame.collection) - if (!coll) { - this.send(ws, { t: "page", fetchId: frame.fetchId, rows: [], seq: "0" }) - return - } - // A cursor must carry BOTH halves (TanStack's CursorExpressions always - // does). A missing `whereCurrent` would otherwise compose to an empty - // predicate and run the ties SELECT unbounded — a silent full-table scan, - // which the operator floor exists to forbid. Reject loudly instead. - if (frame.cursor != null && (frame.cursor.whereCurrent == null || frame.cursor.whereFrom == null)) { - console.error(`fetch '${frame.fetchId}' on '${frame.collection}' rejected: malformed cursor`) - this.send(ws, { t: "page", fetchId: frame.fetchId, rows: [], seq: String(currentSeq(this.sql)) }) - return - } - const seq = String(currentSeq(this.sql)) - try { - const rows: Array = [] - // Cursor present: ties first (base AND whereCurrent, unbounded boundary - // set), then the bounded next page (base AND whereFrom). The cursor - // expressions arrive raw — excluding the base — so we compose them here. - // No cursor: a plain bounded `where` read. - if (frame.cursor != null) { - const tq = compileSubsetQuery(frame.collection, { - where: andPredicates(frame.where, frame.cursor.whereCurrent), - orderBy: frame.orderBy, - }) - rows.push(...Array.from(this.sql.exec(tq.sql, ...tq.params))) - } - const nextWhere = frame.cursor != null ? andPredicates(frame.where, frame.cursor.whereFrom) : frame.where - const nq = compileSubsetQuery(frame.collection, { - where: nextWhere, - orderBy: frame.orderBy, - limit: frame.limit, - }) - rows.push(...Array.from(this.sql.exec(nq.sql, ...nq.params))) - this.send(ws, { t: "page", fetchId: frame.fetchId, rows, seq }) - } catch (e) { - if (e instanceof UnsupportedPredicateError) { - console.error(`fetch '${frame.fetchId}' on '${frame.collection}' rejected: ${e.message}`) - this.send(ws, { t: "page", fetchId: frame.fetchId, rows: [], seq }) - return - } - throw e - } + // The base facade is generics-erased (SyncApi); the runtime + // is type-agnostic, so re-narrowing the schema here is sound. + this.sync.registerSync(schema as SyncSchema) } /** - * Apply a mutation atomically and confirm on the single ordered stream. - * - * Order is the load-bearing invariant (ADR-0002 C1): this connection's - * matched deltas are flushed BEFORE its `committed` frame, so the client's - * single cursor only ever advances over a contiguous applied prefix and the - * optimistic overlay is never dropped before the authoritative row lands. - * With no egress coalescer yet (M4), `drainAndBroadcast` sends deltas - * synchronously here; M4 must preserve this by flushing the originating - * socket before `committed`. + * Apply a server-originated write and broadcast it (ADR-0006). Legacy alias for + * `this.sync.runSyncedWrite`. */ - private async handleMut(ws: WebSocket, f: Extract): Promise { - // Inbound limit: reject over-length batches without applying anything - // (ADR-0012). Reject-don't-truncate: a partial apply silently drops writes. - if (f.ops.length > this.maxOpsPerMutation) { - return this.rejectTx(ws, f.txId, `mutation exceeds maxOpsPerMutation (${this.maxOpsPerMutation})`, "LIMIT_EXCEEDED") - } - - const seen = lookupTx(this.sql, f.txId) - if (seen) return this.replayReceipt(ws, f.txId, seen) - - const user = this.userFor(ws) - - // Authorize every op BEFORE the transaction (may be async). - try { - for (const op of f.ops) { - const def = this.registry.mutations.get(`${f.collection}:${op.type}`) - if (!def) throw new Error(`no mutation handler for '${f.collection}:${op.type}'`) - if (def.authorize) await def.authorize({ user, op, sql: this.sql, env: this.env }) - } - } catch (e) { - // authorize and validation errors surface to the client: a schema failure - // carries a VALIDATION code, an authz "throw to deny" keeps its message. The - // execute catch below stays sanitized. - if (e instanceof ValidationError) return this.rejectTx(ws, f.txId, e.message, "VALIDATION") - return this.rejectTx(ws, f.txId, errorMessage(e)) - } - - // Apply all ops in one synchronous transaction (atomic with the trigger - // rows). A handler that returns a Promise is a programming error. - let commitSeq: string - try { - this.ctx.storage.transactionSync(() => { - for (const op of f.ops) { - const def = this.registry.mutations.get(`${f.collection}:${op.type}`)! - const result = def.execute({ user, op, sql: this.sql, env: this.env }) as unknown - if (result !== undefined && typeof (result as PromiseLike).then === "function") { - // `execute` runs inside transactionSync, which cannot await; an async - // execute also can't be atomic with its CDC rows. Do async work in - // `authorize` (pre-tx), `afterCommit` (post-commit), or a command. - throw new Error( - `mutation '${f.collection}:${op.type}' execute must be synchronous — do async work in authorize, afterCommit, or a command`, - ) - } - } - }) - commitSeq = String(currentSeq(this.sql)) - } catch (e) { - // Log full detail server-side; send only a generic message to the client - // (ADR-0012). SQLite constraint strings, column names, and programming- - // error text are internal detail — not client API surface. The authorize - // catch above is intentionally kept user-facing (README: "throw to deny"). - console.error(`mutation '${f.collection}' execute failed: ${errorMessage(e)}`) - return this.rejectTx(ws, f.txId, "mutation failed", "EXECUTE_FAILED") - } - - recordTx(this.sql, f.txId, true, commitSeq, null, null) - // Enqueue deltas for all subscribers, then flush THIS socket before its - // receipt (C1) so its deltas land first. Other subscribers flush on the - // coalescer tick. - this.drainAndBroadcast() - this.broadcaster.flushOne(ws) - this.send(ws, { t: "committed", txId: f.txId, seq: commitSeq }) - - // Fire-and-forget post-commit hooks AFTER the receipt — never on the - // client's critical path. Each runs under `waitUntil` (keeps the DO alive - // until it settles) and is isolated: a throw is logged and dropped, leaving - // the committed mutation untouched. The hook owns its own idempotency - // (ADR-0004); the library guarantees only "runs once per commit, off-path". - for (const op of f.ops) { - const after = this.registry.mutations.get(`${f.collection}:${op.type}`)?.afterCommit - if (!after) continue - this.ctx.waitUntil( - (async () => { - try { - await after({ user, op, sql: this.sql, env: this.env }) - } catch (e) { - console.error(`afterCommit '${f.collection}:${op.type}' failed: ${errorMessage(e)}`) - } - })(), - ) - } - } - - /** Run a named command (outside any transaction) and confirm with its result. */ - private async handleCall(ws: WebSocket, f: Extract): Promise { - const seen = lookupTx(this.sql, f.txId) - if (seen) return this.replayReceipt(ws, f.txId, seen) - - const def = this.registry.commands.get(f.name) - if (!def) return this.rejectTx(ws, f.txId, `unknown command '${f.name}'`, "UNKNOWN_COMMAND") - - const user = this.userFor(ws) - // authorize and validation errors surface like a mutation's: a schema failure - // carries a VALIDATION code, an authz "throw to deny" keeps its message. This - // matches mutation surfacing, revising ADR-0012 D3 (which sanitized a - // command's authorize too). - try { - if (def.authorize) await def.authorize({ user, args: f.args, sql: this.sql, env: this.env }) - } catch (e) { - if (e instanceof ValidationError) return this.rejectTx(ws, f.txId, e.message, "VALIDATION") - return this.rejectTx(ws, f.txId, errorMessage(e)) - } - // execute runs arbitrary, often async code, so its errors are sanitized like a - // mutation's execute — internal detail never leaks. - let result: unknown - try { - result = await def.execute({ user, args: f.args, sql: this.sql, env: this.env }) - } catch (e) { - console.error(`command '${f.name}' execute failed: ${errorMessage(e)}`) - return this.rejectTx(ws, f.txId, "command failed", "EXECUTE_FAILED") - } - - // Serialize the result for dedup replay BEFORE recording success. A - // non-serializable result can't be replayed, so record an error rather - // than risk re-running the command's side effects on retry. - let stored: string | null - try { - stored = encodeResult(result) - } catch (e) { - return this.rejectTx(ws, f.txId, `non-serializable command result: ${errorMessage(e)}`, "NON_SERIALIZABLE") - } - - const commitSeq = String(currentSeq(this.sql)) - recordTx(this.sql, f.txId, true, commitSeq, null, stored) - this.drainAndBroadcast() - this.broadcaster.flushOne(ws) - this.send(ws, { t: "committed", txId: f.txId, seq: commitSeq, result }) - } - - private rejectTx(ws: WebSocket, txId: string, message: string, code?: string): void { - recordTx(this.sql, txId, false, null, message, null) - this.send(ws, { t: "rejected", txId, error: code ? { code, message } : { message } }) - } - - private replayReceipt(ws: WebSocket, txId: string, seen: SeenTx): void { - if (seen.ok) { - this.send(ws, { t: "committed", txId, seq: seen.cursor ?? "0", result: decodeResult(seen.result) }) - } else { - this.send(ws, { t: "rejected", txId, error: { message: seen.error ?? "unknown" } }) - } + protected runSyncedWrite(fn: (sql: SqlStorage) => T): T { + return this.sync.runSyncedWrite(fn) } - /** - * Apply a SERVER-ORIGINATED write and broadcast it to connected clients - * (ADR-0006). The home for writes outside the client mutation flow — an agent - * inserting a row, a webhook, a cron/`alarm` job, an admin edit, a bulk seed. - * - * `fn` runs inside `transactionSync` (atomic, and the same synchronous - * constraint mutations live under) and may return a value (e.g. an inserted - * count); its CDC is then drained and broadcast on the next coalescer tick. - * Unlike a mutation there is no `txId`, no `committed` receipt, and no - * dedup — a server write has no client to confirm to. Idempotency is the - * caller's job via the collection's mandated stable keys (`INSERT OR IGNORE`). - * - * `registerSync` (in your constructor) has already created the CDC triggers, - * so a write here reaches connected clients with no extra ceremony (ADR-0007). - * - * A thenable return is rejected (and rolls back): any async work belongs - * BEFORE the call, not inside the transaction. - */ - protected runSyncedWrite(fn: (sql: SqlStorage) => T): T { - let result: T - this.ctx.storage.transactionSync(() => { - result = fn(this.sql) - if (result != null && typeof (result as unknown as PromiseLike).then === "function") { - throw new Error("runSyncedWrite fn must be synchronous (it returned a thenable)") - } - }) - this.drainAndBroadcast() - return result! + /** The compiled schema; throws (ADR-0007) if `registerSync` hasn't run yet. + * Legacy alias for `this.sync.registry`. */ + protected get registry(): CompiledSync { + return this.sync.registry as CompiledSync } - /** - * Drain `_sync_changes` from the last broadcast watermark, fan out one `d` - * per changed key to each subscriber of the affected collection, then a - * single `uptodate` boundary per touched socket. Multiple changes to a key - * within the drain collapse to the latest op. - */ + /** Drain the CDC log and broadcast pending deltas (ADR-0006). Legacy alias for + * `this.sync.drainAndBroadcast`. */ protected drainAndBroadcast(): void { - const sql = this.sql - const last = getDrainCursor(sql) - const changes = readChangesSince(sql, last) - if (changes.length === 0) return - const cursor = String(changes[changes.length - 1]!.seq) - - const byTable = new Map>() - for (const c of changes) { - let arr = byTable.get(c.tbl) - if (!arr) { - arr = [] - byTable.set(c.tbl, arr) - } - arr.push(c) - } - - for (const [tbl, tableChanges] of byTable) { - const coll = this.registry.collections.get(tbl) - if (!coll) continue - const latest = new Map() - for (const c of tableChanges) latest.set(c.key, c) - const liveKeys = [...latest.values()].filter((c) => c.op !== "delete").map((c) => c.key) - const hydrated = hydrateRows(sql, tbl, coll.pk, liveKeys) - - // Enqueue into the coalescer; it flushes one `d` per surviving key plus a - // single `uptodate` boundary per socket (on the tick, or via flushOne). - for (const { ws, sub } of this.subs.forCollection(tbl)) { - for (const [key, change] of latest) { - const row = hydrated.get(key) - // Always-emit rule (no before-image, ADR-0002 C4): a key that is - // deleted, gone, or no longer matches this sub's predicate -> a - // synthetic delete (idempotent; move-out). A matching live row -> - // its current state with the actual op (move-in via update upserts - // on the client — verified). Predicate is always-true when unfiltered. - if (change.op === "delete" || !row || !sub.predicate(row)) { - this.broadcaster.enqueue(ws, { subId: sub.subId, key, op: "delete" }, cursor) - } else { - // Full row as the partial patch; column-level diffs arrive later. - this.broadcaster.enqueue(ws, { subId: sub.subId, key, op: change.op, cols: row }, cursor) - } - } - } - } - - setDrainCursor(sql, changes[changes.length - 1]!.seq) - this.maybeCompact() + this.sync.drainAndBroadcast() } /** - * Opportunistic GC: every `compactionEvery` drained mutations, collapse the - * change log to latest-op-per-key and sweep expired dedup entries. Deferred - * via `ctx.waitUntil` so it rides just after a burst of work — it never - * blocks a mutation's response, and (unlike an alarm) never wakes an idle DO. - * `waitUntil` keeps the DO alive until it completes. + * Validate the upgrade and produce the attachment bound to the WebSocket + * (read via `deserializeAttachment` in handlers). Override to read a + * Worker-forged claims header and/or reject by throwing a `Response`. + * Default: no identity. */ - private maybeCompact(): void { - if (++this.writesSinceCompaction < this.compactionEvery) return - this.writesSinceCompaction = 0 - this.ctx.waitUntil( - (async (): Promise => { - compactChanges(this.sql) - pruneChanges(this.sql, this.changelogRetentionMs, Date.now()) - sweepDedup(this.sql, this.dedupRetentionMs, Date.now()) - })(), - ) - } - - /** Full-collection subscribe: emit every current row as a snapshot, then a - * boundary. Predicate/subset shaping arrives in M5/M6. */ - private handleSub(ws: WebSocket, frame: Extract): void { - const coll = this.registry.collections.get(frame.collection) - if (!coll) { - // Unknown collection: drop the subscriber's view. Richer sub-error - // signalling is deferred; for now reset is the honest minimum. - this.send(ws, { t: "reset", sub: frame.subId }) - return - } - - // Per-socket subscription cap (ADR-0012). A re-sub on an existing subId - // replaces the old entry (SubscriptionRegistry.add semantics) — count - // only new subIds against the cap. - const existingCount = this.subs.countFor(ws) - const existingSub = this.subs.forWs(ws).find((s) => s.subId === frame.subId) - if (!existingSub && existingCount >= this.maxSubsPerSocket) { - console.error(`sub '${frame.subId}' refused: maxSubsPerSocket (${this.maxSubsPerSocket}) reached`) - this.send(ws, { t: "reset", sub: frame.subId }) - return - } - // Lower where/orderBy/limit/offset into SQLite. An un-lowerable predicate - // (outside the supported floor) is rejected, not silently full-scanned. - let query: { sql: string; params: Array } - try { - query = compileSubsetQuery(frame.collection, { - where: frame.where, - orderBy: frame.orderBy, - limit: frame.limit, - offset: frame.offset, - }) - } catch (e) { - if (e instanceof UnsupportedPredicateError) { - console.error(`sub '${frame.subId}' on '${frame.collection}' rejected: ${e.message}`) - this.send(ws, { t: "reset", sub: frame.subId }) - return - } - throw e - } - - // C1′ (ADR-0011, generalizing ADR-0002 C1): what follows is a synchronous - // cursor-advancing emission — a snapshot's `snap-end` or a catch-up's - // `uptodate` carries the CURRENT seq, which may include changes whose - // deltas are still buffered in the coalescer for this socket. Flush them - // first, or the client's cursor claims a seq it never applied and a drop - // before the tick loses the write (reconnect resumes past it). - this.broadcaster.flushOne(ws) - - // Registering compiles the predicate in @tanstack/db's evaluator. If the - // predicate is outside the JS floor (e.g. an operator the SQL floor somehow - // let through), that throws UnsupportedPredicateError — reject with `reset` - // rather than letting it escape uncaught and hang the client (ADR-0013). - // With the floors aligned this is belt-and-suspenders, but it must hold for - // any future operator added to one floor and not the other. - let sub: Sub - try { - sub = this.subs.add(ws, frame.subId, frame.collection, frame.where) - } catch (e) { - if (e instanceof UnsupportedPredicateError) { - console.error(`sub '${frame.subId}' on '${frame.collection}' rejected: ${e.message}`) - this.send(ws, { t: "reset", sub: frame.subId }) - return - } - throw e - } - const seq = String(currentSeq(this.sql)) - - // Reconnect catch-up: a `since` cursor asks for changes after that point - // rather than a fresh snapshot. Serve a windowed delta while the change log - // still reaches back that far; otherwise fall back to reset + snapshot. - // - // The floor is `minChangeSeq` — no persisted watermark needed (ADR-0009). - // Retention prunes a seq-prefix (ts is monotone in seq), so every pruned - // change is below the floor: a client at `since >= floor-1` is missing - // nothing, one below it may be, and must reset. An EMPTY log (floor 0) with - // a real `since` means history was pruned away (or this is a storage-reset - // incarnation) — also a reset, never a silent "up to date". - const since = frame.since != null ? Number(frame.since) : 0 - if (since > 0) { - const floor = minChangeSeq(this.sql) - if (floor !== 0 && since >= floor - 1) { - this.emitCatchUp(ws, sub, coll, since, seq) - return - } - this.send(ws, { t: "reset", sub: frame.subId }) - } - - const rows = Array.from(this.sql.exec(query.sql, ...query.params)) as Array> - for (const row of rows) { - this.send(ws, { t: "snap", sub: frame.subId, key: row[coll.pk], row, seq }) - } - this.send(ws, { t: "snap-end", sub: frame.subId, seq }) - } - - /** Windowed catch-up: the latest op per changed key since `since`, resolved - * through the sub's predicate (move-in/out), then an `uptodate` boundary. */ - private emitCatchUp( - ws: WebSocket, - sub: { subId: string; predicate: (row: Record) => boolean }, - coll: { table: string; pk: string }, - since: number, - seq: string, - ): void { - const changes = readChangesSinceFor(this.sql, coll.table, since) - const latest = new Map() - for (const c of changes) latest.set(c.key, c) - const liveKeys = [...latest.values()].filter((c) => c.op !== "delete").map((c) => c.key) - const hydrated = hydrateRows(this.sql, coll.table, coll.pk, liveKeys) - for (const [key, change] of latest) { - const row = hydrated.get(key) - if (change.op === "delete" || !row || !sub.predicate(row)) { - this.send(ws, { t: "d", sub: sub.subId, key, op: "delete", seq }) - } else { - this.send(ws, { t: "d", sub: sub.subId, key, op: change.op, cols: row, seq }) - } - } - this.send(ws, { t: "uptodate", seq }) - } - - /** Encode and send a server frame on one socket. */ - protected send(ws: WebSocket, frame: ServerFrame): void { - ws.send(this.codec.encode(frame)) - } - - /** The attachment bound at upgrade, surviving hibernation. */ - protected userFor(ws: WebSocket): TUser { - return ws.deserializeAttachment() as TUser - } - - /** Live sockets — used by the broadcaster (M4) for fan-out. */ - protected getLiveWs(): Iterable { - return this.liveWs + protected parseAttachment(_req: Request): TUser | Promise { + return undefined as TUser } } - -function errorMessage(e: unknown): string { - return e instanceof Error ? e.message : String(e) -} diff --git a/tests/env.d.ts b/tests/env.d.ts index d6b9710..8713478 100644 --- a/tests/env.d.ts +++ b/tests/env.d.ts @@ -8,5 +8,7 @@ declare module "cloudflare:test" { MAINT_DO: DurableObjectNamespace SLOW_DO: DurableObjectNamespace LIMITS_DO: DurableObjectNamespace + HOST_DO: DurableObjectNamespace + HOST_OPTIN_DO: DurableObjectNamespace } } diff --git a/tests/host-matrix.test.ts b/tests/host-matrix.test.ts new file mode 100644 index 0000000..6b93c04 --- /dev/null +++ b/tests/host-matrix.test.ts @@ -0,0 +1,292 @@ +import { env, runInDurableObject, SELF } from "cloudflare:test" +import { describe, expect, it } from "vitest" +import { createFrameCodec } from "../src/wire/frame-codec.ts" +import type { ClientFrame, ServerFrame } from "../src/wire/frames.ts" +import { SYNC_TAG } from "../src/server/mixin.ts" +import { type FakeHost, testSchema } from "./test-worker.ts" + +// WHY: the mixin's entire reason to exist is cohosting — one DO serving both its +// framework's WebSocket surface AND tddc's sync protocol with no cross-talk +// (ADR-0015). These drive the mixin over a FAKE partyserver-like base (owns +// `__pk` sockets, filters foreign sockets, exposes a `sql` tagged template, +// upgrades on `/_host`) and pin the safety properties the cohosting proof rests +// on: tag-partitioned sockets, path-partitioned upgrades, `super` delegation of +// foreign traffic, no `sql` shadow, and trigger safety on a host with its own +// tables. The fake stands in for the real `agents` package so CI never carries a +// ~13 MB pre-1.0 dependency; a real-`agents` smoke gates version bumps. + +const codec = createFrameCodec() +const HOST_TAG = "__host" + +async function openWs(path: string, headers: Record = {}): Promise { + const res = await SELF.fetch(`https://example.com${path}`, { headers: { Upgrade: "websocket", ...headers } }) + expect(res.status).toBe(101) + const ws = res.webSocket + if (!ws) throw new Error("no webSocket on 101 response") + ws.accept() + return ws +} + +function collectUntil(ws: WebSocket, done: (f: ServerFrame) => boolean, timeoutMs = 2000): Promise> { + return new Promise((resolve, reject) => { + const out: Array = [] + const timer = setTimeout(() => reject(new Error(`timeout; got [${out.map((f) => f.t).join(",")}]`)), timeoutMs) + const onMsg = (e: MessageEvent): void => { + out.push(codec.decode(e.data as ArrayBuffer) as ServerFrame) + if (done(out[out.length - 1]!)) { + clearTimeout(timer) + ws.removeEventListener("message", onMsg) + resolve(out) + } + } + ws.addEventListener("message", onMsg) + }) +} + +/** Resolve true iff NO message arrives within `ms` — the honest way to assert a + * frame was NOT delivered to a socket (used for cross-delivery and auto-response + * negatives). */ +function noMessageWithin(ws: WebSocket, ms: number): Promise { + return new Promise((resolve) => { + const onMsg = (): void => { + clearTimeout(timer) + ws.removeEventListener("message", onMsg) + resolve(false) + } + const timer = setTimeout(() => { + ws.removeEventListener("message", onMsg) + resolve(true) + }, ms) + ws.addEventListener("message", onMsg) + }) +} + +const sub = (subId: string, collection: string): ClientFrame => ({ t: "sub", subId, collection }) +const insert = (txId: string, collection: string, id: string, body: string): ClientFrame => ({ + t: "mut", + txId, + collection, + ops: [{ type: "insert", key: id, cols: { id, body } }], +}) + +describe("Syncable over a partyserver-like host (ADR-0015)", () => { + it("sync and host sockets coexist; each protocol's frames reach only its own sockets", async () => { + const room = "coexist" + const hostWs = await openWs(`/host/${room}/_host`) + const syncWs = await openWs(`/host/${room}/_sync`, { "x-user": "alice" }) + + // A sync frame on the sync socket is handled by the mixin (snap-end back)… + syncWs.send(codec.encode(sub("s1", "messages"))) + const frames = await collectUntil(syncWs, (f) => f.t === "snap-end") + expect(frames.at(-1)?.t).toBe("snap-end") + + // …and a plain string on the host socket is delegated to the host handler. + hostWs.send("hello-host") + await new Promise((r) => setTimeout(r, 100)) + + const stub = env.HOST_DO.get(env.HOST_DO.idFromName(room)) + await runInDurableObject(stub, (instance, state) => { + const host = instance as unknown as FakeHost + // The host saw ONLY its own frame — never the sync `sub`. + expect(host.hostInbox).toEqual(["hello-host"]) + // Sockets are tag-partitioned: exactly one each, the discriminator the + // wake-time restore keys on. + expect(state.getWebSockets(SYNC_TAG).length).toBe(1) + expect(state.getWebSockets(HOST_TAG).length).toBe(1) + expect(state.getWebSockets().length).toBe(2) + }) + hostWs.close() + syncWs.close() + }) + + it("a broadcast reaches only the sync socket, never the host socket", async () => { + const room = "broadcast-iso" + const hostWs = await openWs(`/host/${room}/_host`) + const syncWs = await openWs(`/host/${room}/_sync`, { "x-user": "bob" }) + + syncWs.send(codec.encode(sub("s1", "messages"))) + await collectUntil(syncWs, (f) => f.t === "snap-end") + + // The host socket must receive nothing while the sync write fans out. + const hostSilent = noMessageWithin(hostWs, 500) + syncWs.send(codec.encode(insert("t1", "messages", "m1", "hi"))) + const out = await collectUntil(syncWs, (f) => f.t === "committed") + + // The sync client got its delta + receipt… + expect(out.some((f) => f.t === "d" || f.t === "snap")).toBe(true) + expect(out.some((f) => f.t === "committed")).toBe(true) + // …and the host socket stayed silent (the broadcaster never touched it). + expect(await hostSilent).toBe(true) + hostWs.close() + syncWs.close() + }) + + it("a non-sync upgrade reaches super.fetch (host claims /_host); /_sync never reaches the host", async () => { + const room = "fetch-split" + // /_host is claimed by the delegated host fetch. + const hostWs = await openWs(`/host/${room}/_host`) + // A non-upgrade request falls through to the host, which 404s its own way. + const res = await SELF.fetch(`https://example.com/host/${room}/status`) + expect(res.status).toBe(404) + expect(await res.text()).toBe("host: not found") + // /_sync is claimed by the mixin — the host never records a socket for it. + const syncWs = await openWs(`/host/${room}/_sync`, { "x-user": "carol" }) + const stub = env.HOST_DO.get(env.HOST_DO.idFromName(room)) + await runInDurableObject(stub, (_i, state) => { + expect(state.getWebSockets(HOST_TAG).length).toBe(1) // only the /_host socket + expect(state.getWebSockets(SYNC_TAG).length).toBe(1) // only the /_sync socket + }) + hostWs.close() + syncWs.close() + }) + + it("a host socket's message is delegated to super.webSocketMessage and handled by the host", async () => { + const room = "delegate-msg" + const hostWs = await openWs(`/host/${room}/_host`) + hostWs.send("host-frame") + await new Promise((r) => setTimeout(r, 100)) + const stub = env.HOST_DO.get(env.HOST_DO.idFromName(room)) + await runInDurableObject(stub, (instance) => { + expect((instance as unknown as FakeHost).hostInbox).toContain("host-frame") + }) + hostWs.close() + }) + + it("does not shadow the host's `sql` tagged-template method", async () => { + const room = "no-sql-shadow" + await openWs(`/host/${room}/_sync`) // force construction + const stub = env.HOST_DO.get(env.HOST_DO.idFromName(room)) + await runInDurableObject(stub, (instance) => { + const host = instance as unknown as FakeHost + // `this.sql` is still the host's tagged-template FUNCTION, not a SqlStorage + // getter — if the mixin defined a `sql` getter this would throw. + expect(typeof host.sql).toBe("function") + expect(host.sql<{ one: number }>`SELECT 1 AS one`).toEqual([{ one: 1 }]) + }) + }) + + it("installs triggers only on registered tables; a host-owned table gets none and the reaper spares host triggers", async () => { + const room = "trigger-safety" + await openWs(`/host/${room}/_sync`) // registerSync runs in the constructor + const stub = env.HOST_DO.get(env.HOST_DO.idFromName(room)) + await runInDurableObject(stub, (instance, state) => { + const sql = state.storage.sql + const triggers = new Set( + Array.from(sql.exec("SELECT name FROM sqlite_master WHERE type='trigger'")).map((r) => r.name as string), + ) + // Registered table: triggers present. Host-owned table: none. + expect(triggers.has("_sync_changes_messages_ai")).toBe(true) + expect([...triggers].some((n) => n.includes("cf_agents_state"))).toBe(false) + + // A host write emits no CDC row (no trigger on the unregistered table). + const before = Array.from(sql.exec("SELECT count(*) AS c FROM _sync_changes"))[0]!.c as number + sql.exec("INSERT INTO cf_agents_state(key, value) VALUES ('k', 'v')") + const after = Array.from(sql.exec("SELECT count(*) AS c FROM _sync_changes"))[0]!.c as number + expect(after).toBe(before) + + // The reaper only drops `_sync_changes_*`; a host-named trigger survives a + // re-register (GLOB treats `_` literally — ADR-0008). Re-registering the + // same schema is exactly what a real author's constructor does on wake. + sql.exec(`CREATE TRIGGER cf_agents_guard AFTER INSERT ON cf_agents_state BEGIN SELECT 1; END`) + ;(instance as unknown as { sync: { registerSync(s: typeof testSchema): void } }).sync.registerSync(testSchema) + const stillThere = Array.from( + sql.exec("SELECT name FROM sqlite_master WHERE type='trigger' AND name = 'cf_agents_guard'"), + ) + expect(stillThere.length).toBe(1) + // And the registered-table triggers are still present after the reap. + const msgTrig = Array.from( + sql.exec("SELECT name FROM sqlite_master WHERE type='trigger' AND name = '_sync_changes_messages_ai'"), + ) + expect(msgTrig.length).toBe(1) + }) + }) + + it("bare DO treats a legacy untagged socket as sync (0.4.0 → mixin migration)", async () => { + // A socket accepted by 0.4.0 carried NO tag. After the mixin upgrade, such a + // socket can wake out of hibernation; the bare DO must still treat it as sync + // (it owns every socket), or a pre-existing client's frames are silently + // ignored. Simulate it by accepting an untagged socket directly. + const room = "legacy-untagged" + await openWs(`/sync/${room}`) // construct the DO + its schema + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + await runInDurableObject(stub, async (instance, state) => { + const pair = new WebSocketPair() + const server = pair[1] + state.acceptWebSocket(server) // NO tag — exactly what 0.4.0 did + const client = pair[0] + client.accept() + const got = new Promise((resolve, reject) => { + const t = setTimeout(() => reject(new Error("untagged socket frame ignored")), 2000) + client.addEventListener( + "message", + (e) => { clearTimeout(t); resolve(codec.decode(e.data as ArrayBuffer) as ServerFrame) }, + { once: true }, + ) + }) + // The mixin must PROCESS the frame (a snap-end back), not delegate/ignore it. + const handler = instance as unknown as { webSocketMessage(ws: WebSocket, m: ArrayBuffer): Promise } + await handler.webSocketMessage(server, codec.encode(sub("s1", "messages")) as unknown as ArrayBuffer) + expect((await got).t).toBe("snap-end") + }) + }) + + it("configure({ caseSensitiveLike: false }) is a real toggle, not a dead option", async () => { + const room = "toggle-pragma" + await openWs(`/sync/${room}`) + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + await runInDurableObject(stub, (instance, state) => { + const like = (): unknown => Array.from(state.storage.sql.exec("SELECT ('A' LIKE 'a') AS m"))[0]!.m + expect(like()).toBe(0) // bare-DO default ON → case-sensitive + ;(instance as unknown as { sync: { configure(o: { caseSensitiveLike: boolean }): void } }).sync.configure({ + caseSensitiveLike: false, + }) + expect(like()).toBe(1) // toggled OFF → case-insensitive + }) + }) + + it("SyncDurableObject (bare DO base) keeps 0.4.0 defaults: auto-response ON, case-sensitive LIKE ON", async () => { + const room = "bare-defaults" + const ws = await openWs(`/sync/${room}`) + // Auto-response ON: a literal "ping" pongs without reaching a handler. + ws.send("ping") + const pong = await new Promise((resolve, reject) => { + const t = setTimeout(() => reject(new Error("no pong")), 2000) + ws.addEventListener("message", (e) => { clearTimeout(t); resolve(e.data as string) }, { once: true }) + }) + expect(pong).toBe("pong") + // Pragma ON: LIKE is case-sensitive. + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)) + await runInDurableObject(stub, (_i, state) => { + expect(Array.from(state.storage.sql.exec("SELECT ('A' LIKE 'a') AS m"))[0]!.m).toBe(0) + }) + ws.close() + }) + + it("over a non-DO base the two DO-global side effects default OFF, and `configure` opts them back on", async () => { + // Default OFF: pragma off → LIKE is case-insensitive; auto-response off → a + // literal "ping" is NOT pong'd (it is decoded as a frame and dropped). + const offRoom = "sidefx-off" + const offWs = await openWs(`/host/${offRoom}/_sync`) + const noPong = noMessageWithin(offWs, 500) + offWs.send("ping") + expect(await noPong).toBe(true) + await runInDurableObject(env.HOST_DO.get(env.HOST_DO.idFromName(offRoom)), (_i, state) => { + expect(Array.from(state.storage.sql.exec("SELECT ('A' LIKE 'a') AS m"))[0]!.m).toBe(1) + }) + offWs.close() + + // Opted ON via configure(): pragma on → case-sensitive; auto-response on → pong. + const onRoom = "sidefx-on" + const onWs = await openWs(`/host-optin/${onRoom}/_sync`) + onWs.send("ping") + const pong = await new Promise((resolve, reject) => { + const t = setTimeout(() => reject(new Error("no pong")), 2000) + onWs.addEventListener("message", (e) => { clearTimeout(t); resolve(e.data as string) }, { once: true }) + }) + expect(pong).toBe("pong") + await runInDurableObject(env.HOST_OPTIN_DO.get(env.HOST_OPTIN_DO.idFromName(onRoom)), (_i, state) => { + expect(Array.from(state.storage.sql.exec("SELECT ('A' LIKE 'a') AS m"))[0]!.m).toBe(0) + }) + onWs.close() + }) +}) diff --git a/tests/sql-compiler.test.ts b/tests/sql-compiler.test.ts index a081487..eb1ef2e 100644 --- a/tests/sql-compiler.test.ts +++ b/tests/sql-compiler.test.ts @@ -81,10 +81,19 @@ describe("IR -> SQL compiler (M6)", () => { it("emits LIMIT -1 when offset is given without a limit (SQLite requirement)", () => { const q = compileSubsetQuery("t", { offset: 5 }) - expect(q.sql).toBe(`SELECT * FROM "t" LIMIT -1 OFFSET ?`) + expect(q.sql).toBe(`SELECT * FROM "t" ORDER BY rowid LIMIT -1 OFFSET ?`) expect(q.params).toEqual([5]) }) + it("defaults to ORDER BY rowid when the client sends no orderBy (deterministic snapshot order)", () => { + // Field-verified regression: a bare SELECT with no ORDER BY leaves row order + // as an accident of SQLite's query plan. `rowid` is always available (D9 + // forbids an INTEGER PRIMARY KEY pk) and matches insertion order among + // currently-live rows. + const q = compileSubsetQuery("t", {}) + expect(q.sql).toBe(`SELECT * FROM "t" ORDER BY rowid`) + }) + it("rejects a negative limit/offset and an invalid orderBy column", () => { expect(() => compileSubsetQuery("t", { limit: -1 })).toThrow(/non-negative/) expect(() => compileSubsetQuery("t", { orderBy: [{ col: "a; DROP" }] })).toThrow(/orderBy column/) diff --git a/tests/subset-query.test.ts b/tests/subset-query.test.ts index d3c2f1f..500c7fa 100644 --- a/tests/subset-query.test.ts +++ b/tests/subset-query.test.ts @@ -65,6 +65,27 @@ describe("subset shaping pushed into SQLite (M6)", () => { ws.close() }) + it("cold snapshot preserves insertion order when the client sends no orderBy (field-verified regression)", async () => { + const room = "ss-cold-order" + const ws = await openWs(room) + // Insertion order deliberately NOT pk-lexicographic: a query plan that uses + // the pk's autoindex (SQLite may pick it once a WHERE clause touches `id`) + // would return sorted-by-id order ("a","m","z") instead of insertion order + // — exactly the divergence the field-verified bug exposed. All three + // inserts land in one synchronous block (same millisecond), so a `ts`-based + // tiebreak could never disambiguate them even if one were used. + await seed(room, [ + ["z", "1"], + ["a", "2"], + ["m", "3"], + ]) + const where = { type: "func", name: "gt", args: [{ type: "ref", path: ["id"] }, { type: "val", value: "" }] } + send(ws, { t: "sub", subId: "s1", collection: "messages", where } as never) + const frames = await collectUntil(ws, (f) => f.t === "snap-end") + expect(snapKeys(frames)).toEqual(["z", "a", "m"]) + ws.close() + }) + it("rejects a subscription whose predicate cannot be lowered (reset)", async () => { const ws = await openWs("ss-reject") // ilike is outside the supported floor. diff --git a/tests/test-worker.ts b/tests/test-worker.ts index 02a4ec7..a1b5da4 100644 --- a/tests/test-worker.ts +++ b/tests/test-worker.ts @@ -2,6 +2,7 @@ // miniflare.durableObjects, and routes WebSocket upgrades to the sync DO. import { DurableObject } from "cloudflare:workers" +import { Syncable } from "../src/server/mixin.ts" import { defineSync, type StandardSchemaV1 } from "../src/server/registry.ts" import { SyncDurableObject } from "../src/server/sync-do.ts" @@ -71,8 +72,9 @@ function upcasingBody(): StandardSchemaV1 { const sync = defineSync() // The same collections/mutations/commands as before, authored via the -// object-schema API. The schema VALUE is registered on the DO below. -const testSchema = sync.schema({ +// object-schema API. The schema VALUE is registered on the DO below. Exported so +// the host-matrix test can re-register to drive the trigger reaper (ADR-0008). +export const testSchema = sync.schema({ collections: { messages: sync.collection({ pk: "id", @@ -244,12 +246,122 @@ export class LimitsTestDO extends SyncTestDO { protected override readonly maxSubsPerSocket = 2 } +// ---- Host-matrix fixtures (host-matrix.test.ts) --------------------------- +// +// A fake partyserver-like host base and the Syncable mixin applied over it. The +// fake mimics partyserver's OBSERVABLE contract without pulling the ~13 MB agents +// dependency (ADR-0015 test plan): it owns sockets stamped with a `__pk` +// attachment, filters foreign sockets in every hibernation handler, exposes a +// `sql` tagged-template method (the member the mixin must NOT shadow), and +// upgrades on its own `/_host` path — everything else falls through. + +const HOST_TAG = "__host" + +export class FakeHost extends DurableObject { + /** Host-delivered frames, for cross-delivery assertions. */ + hostInbox: Array = [] + /** Count of host-owned socket closes seen by the host handler. */ + hostClosed = 0 + + /** partyserver-style tagged-template query (partyserver dist/index.js:557). + * The mixin must never define a `sql` property that shadows this. */ + sql>(strings: TemplateStringsArray, ...values: Array): Array { + let query = "" + strings.forEach((s, i) => { + query += s + (i < values.length ? "?" : "") + }) + return Array.from(this.ctx.storage.sql.exec(query, ...values)) as Array + } + + /** partyserver's `__pk` discriminator (dist/index.js:14 to 35): a socket is the + * host's iff its attachment carries a `__pk` key. */ + #isHostSocket(ws: WebSocket): boolean { + const a = ws.deserializeAttachment() as { __pk?: unknown } | null + return a != null && typeof a === "object" && "__pk" in a + } + + override async fetch(req: Request): Promise { + const url = new URL(req.url) + if (req.headers.get("Upgrade") === "websocket" && url.pathname.endsWith("/_host")) { + const pair = new WebSocketPair() + pair[1].serializeAttachment({ __pk: crypto.randomUUID() }) + this.ctx.acceptWebSocket(pair[1], [HOST_TAG]) + return new Response(null, { status: 101, webSocket: pair[0] }) + } + return new Response("host: not found", { status: 404 }) + } + + override webSocketMessage(ws: WebSocket, message: string | ArrayBuffer): void { + if (!this.#isHostSocket(ws)) return // partyserver short-circuits foreign sockets + this.hostInbox.push(typeof message === "string" ? message : "") + } + + override webSocketClose(ws: WebSocket): void { + if (!this.#isHostSocket(ws)) return + this.hostClosed++ + } + + override webSocketError(_ws: WebSocket): void {} + + /** partyserver-style broadcast: touches only the host's own sockets. */ + hostBroadcast(msg: string): void { + for (const ws of this.ctx.getWebSockets(HOST_TAG)) ws.send(msg) + } +} + +/** The mixin over the fake host. Defaults: auto-response OFF, pragma OFF (base is + * not DurableObject). Registers the same collections as SyncTestDO, plus a + * host-owned `cf_agents_state` table that is deliberately NOT registered. */ +export class SyncOverHostDO extends Syncable()(FakeHost) { + constructor(ctx: DurableObjectState, env: unknown) { + super(ctx, env) + // Author-owned auth hook via the facade (the mixed base has no legacy + // `parseAttachment` override). + this.sync.configure({ parseAttachment: (req) => ({ userId: req.headers.get("x-user") ?? "anon" }) }) + ctx.blockConcurrencyWhile(async () => { + // NOTE: reach `ctx.storage.sql` directly — `this.sql` here is the HOST's + // tagged-template method (ADR-0015), the whole point of dropping the getter. + const sql = ctx.storage.sql + sql.exec(`CREATE TABLE IF NOT EXISTS messages (id TEXT PRIMARY KEY, body TEXT)`) + sql.exec(`CREATE TABLE IF NOT EXISTS files (id TEXT PRIMARY KEY, name TEXT)`) + sql.exec(`CREATE TABLE IF NOT EXISTS validated (id TEXT PRIMARY KEY, body TEXT)`) + sql.exec(`CREATE TABLE IF NOT EXISTS transformed (id TEXT PRIMARY KEY, body TEXT)`) + // A host-owned table the author never registers (cf. cf_agents_state). + sql.exec(`CREATE TABLE IF NOT EXISTS cf_agents_state (key TEXT PRIMARY KEY, value TEXT)`) + this.sync.registerSync(testSchema) + }) + } +} + +/** Same base, but opts the two DO-global side effects ON — proves `configure` + * restores 0.4.0-style auto-response + case-sensitive LIKE over a non-DO base. */ +export class SyncOverHostOptInDO extends Syncable()(FakeHost) { + constructor(ctx: DurableObjectState, env: unknown) { + super(ctx, env) + this.sync.configure({ + autoResponse: true, + caseSensitiveLike: true, + parseAttachment: (req) => ({ userId: req.headers.get("x-user") ?? "anon" }), + }) + ctx.blockConcurrencyWhile(async () => { + const sql = ctx.storage.sql + sql.exec(`CREATE TABLE IF NOT EXISTS messages (id TEXT PRIMARY KEY, body TEXT)`) + sql.exec(`CREATE TABLE IF NOT EXISTS files (id TEXT PRIMARY KEY, name TEXT)`) + sql.exec(`CREATE TABLE IF NOT EXISTS validated (id TEXT PRIMARY KEY, body TEXT)`) + sql.exec(`CREATE TABLE IF NOT EXISTS transformed (id TEXT PRIMARY KEY, body TEXT)`) + this.sync.registerSync(testSchema) + }) + } +} + interface Env { TEST_DO: DurableObjectNamespace SYNC_DO: DurableObjectNamespace MAINT_DO: DurableObjectNamespace SLOW_DO: DurableObjectNamespace LIMITS_DO: DurableObjectNamespace + HOST_DO: DurableObjectNamespace + HOST_OPTIN_DO: DurableObjectNamespace } export default { @@ -271,6 +383,16 @@ export default { const name = url.pathname.slice("/limits/".length) || "default" return env.LIMITS_DO.get(env.LIMITS_DO.idFromName(name)).fetch(req) } + // Host-matrix routes: /host//... and /host-optin//... forward the + // WHOLE request so the DO sees the trailing /_sync or /_host discriminator. + if (url.pathname.startsWith("/host/")) { + const name = url.pathname.slice("/host/".length).split("/")[0] || "default" + return env.HOST_DO.get(env.HOST_DO.idFromName(name)).fetch(req) + } + if (url.pathname.startsWith("/host-optin/")) { + const name = url.pathname.slice("/host-optin/".length).split("/")[0] || "default" + return env.HOST_OPTIN_DO.get(env.HOST_OPTIN_DO.idFromName(name)).fetch(req) + } return new Response("test-worker") }, } diff --git a/vitest.config.ts b/vitest.config.ts index 73e9251..11f12dc 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -25,6 +25,8 @@ export default defineWorkersProject({ MAINT_DO: { className: "MaintTestDO", useSQLite: true }, SLOW_DO: { className: "SlowTickDO", useSQLite: true }, LIMITS_DO: { className: "LimitsTestDO", useSQLite: true }, + HOST_DO: { className: "SyncOverHostDO", useSQLite: true }, + HOST_OPTIN_DO: { className: "SyncOverHostOptInDO", useSQLite: true }, }, }, },