From 607b6a3bd45bd0ba217cac2eebb81b0e37288030 Mon Sep 17 00:00:00 2001 From: Jacob Maynard Date: Mon, 7 Sep 2026 15:32:46 -0500 Subject: [PATCH] =?UTF-8?q?server:=20onMutationCommitted,=20a=20post-commi?= =?UTF-8?q?t=20hook=20with=20net=20row=20changes=20=E2=80=94=200.2.1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WorkspaceEngineConfig gains onMutationCommitted(event, env), invoked once per mutation that wrote rows, after its transaction committed, with { workspaceId, name, args (parsed), principal, clientId, version, changes }. Each change is the row's stored image at the mutation's first touch against what it left, so a row put twice reports once and a create-then-delete of a new row reports nothing. The hook observes and never participates: it is never awaited (the push handler stays synchronous), a returned promise goes to waitUntil, and a throw or rejection lands on the engine logger with the mutation name and version. Migrations, admin import and reset bypass the mutation path and never fire it; rejected mutations and no-op writes are silent too. Delivery is at-most-once. Before-image capture is opt-in on the WriteSet and the DO enables it only when a hook is configured, so apps without one pay nothing. A tx.get that reached storage doubles as the image, so read-modify-write mutators add no read; blind puts pay one point lookup. Migrations never track. createTestEngine().mutate() returns the same changes list, so the logic behind a hook is unit-testable in node. Additive API, hence a patch. Claude-Session: https://claude.ai/code/session_012Xd62wDxf41DjJK1T3Wh3Y --- ARCHITECTURE.md | 20 ++ docs/guide/mutations.md | 17 ++ docs/guide/testing.md | 1 + docs/reference/server.md | 30 +++ docs/reference/testing.md | 6 +- packages/server/package.json | 2 +- packages/server/src/config.ts | 51 +++- packages/server/src/do.ts | 58 ++++- packages/server/src/engine-core.ts | 61 ++++- packages/server/src/index.ts | 2 + packages/server/src/testing.ts | 19 +- packages/server/test/commit-hook.test.ts | 218 ++++++++++++++++++ packages/server/test/fixture/worker.ts | 14 +- .../server/test/node/write-changes.test.ts | 113 +++++++++ 14 files changed, 586 insertions(+), 26 deletions(-) create mode 100644 packages/server/test/commit-hook.test.ts create mode 100644 packages/server/test/node/write-changes.test.ts diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 2f110cb..d5e5c73 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -189,6 +189,26 @@ storage — `tx.put`, shared by mutators, schema migrations, and admin import: - Validation must be synchronous (mutations commit inside `transactionSync`); a validator returning a Promise is rejected as a permanent error. +**Post-commit hook.** `onMutationCommitted` on the engine config is the one +seam for effects outside the workspace's rows (notifications, projections +into D1, activity logs). With a hook configured — and only then — the +`WriteSet` records each row's stored image at the mutation's first touch (a +`tx.get` that reached storage doubles as the image, so read-modify-write +mutators pay no extra read; blind puts pay one point lookup), and `flush` +returns the net `{ tbl, id, before, after }` list — a row put twice reports +once; a create-then-delete of a new row reports nothing. Migrations never +track. The DO invokes the hook after `transactionSync` returns, +once per mutation that wrote rows, with the parsed args, the principal stamp, +and the committed version; it never awaits it (invariant 3: the push handler +stays synchronous, and a slow consumer cannot delay the poke), hands a +returned promise to `waitUntil`, and routes a throw or rejection to the +engine logger. It is an observer, not a participant: no veto, no writes into +the transaction. Delivery is at-most-once — an eviction with the promise in +flight loses it — so consumers must be idempotent on `version` and +rebuildable from `export`. Migrations, admin import, and reset bypass +`#applyMutation` and therefore never fire it; rejected mutations and no-op +writes do not either. + ## Client adapter The adapter is a TanStack DB collection options creator implementing diff --git a/docs/guide/mutations.md b/docs/guide/mutations.md index 682d5bc..128bada 100644 --- a/docs/guide/mutations.md +++ b/docs/guide/mutations.md @@ -110,3 +110,20 @@ const status = useSyncStatus(client) ``` Status is about the *pipe*, not about individual mutations — individual outcomes arrive through each `mutate` promise. + +## Reacting after the commit + +Mutators are pure over the workspace's rows; anything that reaches outside them — a notification, a projection into your primary database, an activity log — belongs in [`onMutationCommitted`](/reference/server#onmutationcommitted) on the server config. It runs after the mutation's transaction committed with the net before/after image of every row it touched, plus the mutator name, its parsed args, and the connection's principal: + +```ts +export class WorkspaceDO extends createWorkspaceDO({ + app, + onMutationCommitted: async ({ workspaceId, name, principal, changes }, env) => { + await env.DB.prepare('UPDATE projects SET last_activity_at = ?, last_activity_by = ? WHERE id = ?') + .bind(Date.now(), principal, workspaceId) + .run() + }, +}) {} +``` + +The hook observes; it cannot reject or alter the mutation, and the client's confirmation never waits on it. It also never fires for a rejected mutation, a schema migration, or an admin import — only for a mutation that changed rows. Delivery is at-most-once, so a consumer that must not miss an event keys its effects on `version` and can rebuild from an admin export. The same change list comes back from the [test engine](/reference/testing#testmutationresult), so the logic you put behind the hook is unit-testable in node. diff --git a/docs/guide/testing.md b/docs/guide/testing.md index 4ef1895..af0078c 100644 --- a/docs/guide/testing.md +++ b/docs/guide/testing.md @@ -60,6 +60,7 @@ The engine honors the [engine invariants](https://github.com/InfinityBowman/cf-s - An `AppError` from a mutator (or invalid args) reports as `result.error` — **permanent**, no data written, and `engine.lastMutationId()` still advances. Assert on both when testing rejection paths. - Any other throw is **transient**: rethrown, nothing committed. - Auth-dependent mutators can be exercised by passing a principal and auth context, so `ctx.authoritative` permission checks are testable without a socket in sight. +- `result.changes` is the before/after row list the Durable Object would hand to [`onMutationCommitted`](/reference/server#onmutationcommitted), so a hook's logic — which changes warrant a notification, what a projection row should look like — is unit-testable here by feeding it a real mutation's changes. ## Testing the full stack diff --git a/docs/reference/server.md b/docs/reference/server.md index 81ac5e0..6353ef5 100644 --- a/docs/reference/server.md +++ b/docs/reference/server.md @@ -85,6 +85,36 @@ logger: (level, message, { workspaceId }, ...detail) => { } ``` +### onMutationCommitted + +`(event: MutationCommitted, env: Env) => void | Promise` + +Runs after a mutation's data effects commit, with the rows it wrote or deleted — the seam for notifications, projections into another store, and activity logging. Fires once per mutation that changed rows; a rejected mutation, one whose writes net to nothing, a schema migration, and an admin `import` or `reset` all emit nothing. The worker env rides along so the hook can reach its own bindings. + +```ts +onMutationCommitted: async ({ workspaceId, name, principal, changes }, env) => { + for (const { tbl, before, after } of changes) { + if (tbl === 'tasks' && after?.assignee && after.assignee !== before?.assignee) { + await env.NOTIFY.send({ to: after.assignee, workspaceId, by: principal }) + } + } +} +``` + +The hook is an observer, never a participant: it cannot veto or amend the mutation, and the engine never waits for it — the client's confirmation is sent regardless. A returned promise is held with `waitUntil` so the object stays alive until it settles; a rejection or a synchronous throw goes to [`logger`](#logger) at `error` level and affects nothing else. Delivery is at-most-once: a workspace evicted with the promise in flight does not replay it, so a consumer that must not miss an event makes its effects idempotent on `version` and keeps a way to rebuild from an admin [`export`](#createadminfetch-createadminroute). + +`MutationCommitted` carries: + +- `workspaceId` — the workspace the mutation was applied to. +- `name` — the mutator name, as registered in the app definition. +- `args` — the args the mutator ran with: validated and parsed, defaults applied. +- `principal` — the connection's principal stamp, when [`authorize`](#authorize) set one. +- `clientId` — the client that pushed the mutation. +- `version` — the data version the mutation committed as; every `after` row is stamped with it. +- `changes` — `RowChange[]`, the rows the mutation wrote or deleted in the order it first touched them, never empty. Each is `{ tbl, id, before, after }`: `before` is the row as stored when the mutation began (`null` for an insert), `after` what it left (`null` for a delete). A row the mutation touched more than once appears once, spanning its net effect. + +The [test engine](/reference/testing#testmutationresult) returns the same `changes` from `mutate`, so the logic behind a hook can be unit-tested in node without a Durable Object. + ## createSyncFetch · createSyncRoute `(opts: SyncFetchOptions) => (request, env) => Promise` · `…Promise` diff --git a/docs/reference/testing.md b/docs/reference/testing.md index 968b520..53ed7c6 100644 --- a/docs/reference/testing.md +++ b/docs/reference/testing.md @@ -64,6 +64,8 @@ Applies a named mutation authoritatively as the engine's default client, with th - An `AppError` thrown by the mutator — or invalid args — is a **permanent** rejection: writes are discarded, the result carries [`error`](#testmutationresult), and [`lastMutationId`](#lastmutationid) still advances. Assert on both when testing rejection paths. - Any other throw is **transient**: it rethrows out of `mutate`, nothing commits, and the LMID does not advance — the real client would retry the push. +A successful result also carries [`changes`](#testmutationresult) — the rows the mutation wrote or deleted as before/after pairs, exactly what the Durable Object hands to [`onMutationCommitted`](/reference/server#onmutationcommitted) — so the logic behind a hook is testable here too. + ### mutateAs `(clientId, name, args?) => TestMutationResult` @@ -102,10 +104,12 @@ Read-only getters: the default clientId, and the current data version — bumps ### TestMutationResult -`{ error?: { code: string; message: string } }` +`{ error?: { code: string; message: string }; changes: RowChange[] }` The outcome of one authoritative mutation. `error` is present only for permanent rejections: `code` is an engine built-in (`InvalidArgs`, `UnknownMutator`, …) or an app-defined `AppError` code passed through verbatim — the same vocabulary the client's [`MutationError`](/reference/sync-client#mutationerror) carries. Transient failures never produce a result; they throw. +`changes` lists the rows the mutation wrote or deleted, in the order it first touched them, as `{ tbl, id, before, after }` — `before` the stored row when the mutation began (`null` for an insert), `after` what it left (`null` for a delete); a row touched more than once appears once with its net effect. Empty on a permanent error and when the writes net to nothing (then `version` does not move either). This is the same list the Durable Object passes to [`onMutationCommitted`](/reference/server#onmutationcommitted). + ## Schema-drift helpers The CI side of [drift detection](/guide/schema-evolution#drift-detection) — catching the one schema mistake the type system cannot: changing a table schema without bumping `version`. diff --git a/packages/server/package.json b/packages/server/package.json index b49f88c..302d81b 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -1,6 +1,6 @@ { "name": "@cf-sync/server", - "version": "0.2.0", + "version": "0.2.1", "description": "Server-authoritative sync engine on Cloudflare Durable Objects: createWorkspaceDO, worker routers, admin surface, and an in-memory test engine", "license": "MIT", "repository": { diff --git a/packages/server/src/config.ts b/packages/server/src/config.ts index e46234c..a8a26c2 100644 --- a/packages/server/src/config.ts +++ b/packages/server/src/config.ts @@ -139,9 +139,40 @@ export type EngineLogger = ( ...detail: unknown[] ) => void +/** + * One row a committed mutation wrote or deleted. `before` is the row as + * stored when the mutation began (null for an insert); `after` is what the + * mutation left (null for a delete). A row the mutation touched more than + * once appears once, spanning its net effect. + */ +export interface RowChange { + tbl: string + id: string + before: Record | null + after: Record | null +} + +/** What {@link WorkspaceEngineConfig.onMutationCommitted} receives: one committed mutation and its net row effects. */ +export interface MutationCommitted { + /** The workspace the mutation was applied to. */ + workspaceId: string + /** The mutator name, as registered in the app definition. */ + name: string + /** The args the mutator ran with — validated and parsed, defaults applied. */ + args: unknown + /** The connection's principal stamp, when `authorize` set one. */ + principal?: string + /** The client that pushed the mutation. */ + clientId: string + /** The data version the mutation committed as; every `after` row is stamped with it. */ + version: number + /** The rows the mutation wrote or deleted, in the order it first touched them. Never empty. */ + changes: RowChange[] +} + /** * What {@link createWorkspaceDO} takes: the shared app definition, plus - * optional compaction, R2-export, and extension settings. + * optional compaction, R2-export, extension, logging, and post-commit settings. */ export interface WorkspaceEngineConfig { /** @@ -186,4 +217,22 @@ export interface WorkspaceEngineConfig void | Promise } diff --git a/packages/server/src/do.ts b/packages/server/src/do.ts index 5e3b334..b94cc14 100644 --- a/packages/server/src/do.ts +++ b/packages/server/src/do.ts @@ -45,6 +45,8 @@ import { type EngineExtension, type EngineExtensionMessageContext, type EngineLogContext, + type MutationCommitted, + type RowChange, type WorkspaceEngineConfig, } from './config' import { WriteSet, validateRow } from './engine-core' @@ -365,10 +367,10 @@ export function createWorkspaceDO( // One write buffer across the chain: later steps read earlier steps' // writes, and the net result is validated against the current schema // at flush (intermediate shapes are transient). - const writes = new WriteSet(this.#rows, config.app.schema, true) + const writes = new WriteSet(this.#rows, config.app.schema, { validateAtFlush: true }) for (const step of steps) step.migrate?.(writes.tx) const candidate = this.#meta.currentVersion + 1 - if (writes.flush(candidate) > 0) { + if (writes.flush(candidate).written > 0) { migratedVersion = candidate // Rewritten rows are a new data version, and no cursor issued // before the migration may catch up from it — force bootstrap. @@ -1092,7 +1094,9 @@ export function createWorkspaceDO( * Applies one mutation. Returns the permanent app error, if any. The LMID * advance, the mutation-log append, and the data effects commit in one * SQLite transaction; permanent errors advance the LMID with no data - * effects; transient errors throw and roll everything back. + * effects; transient errors throw and roll everything back. The + * post-commit hook runs after the transaction returned, so what it + * observes is what every client will be poked with. */ #applyMutation(attachment: Attachment, mutation: Mutation): { code: string; message: string } | undefined { const { clientId } = attachment @@ -1105,6 +1109,8 @@ export function createWorkspaceDO( } let appError: { code: string; message: string } | undefined let committedVersion: number | null = null + let parsedArgs: unknown = mutation.args + let changes: RowChange[] = [] this.ctx.storage.transactionSync(() => { let wroteVersion: number | null = null @@ -1113,7 +1119,11 @@ export function createWorkspaceDO( // hello, so this is a registry bug, and retrying can never succeed. appError = { code: 'UnknownMutator', message: `no mutator named "${mutation.name}"` } } else { - const writes = new WriteSet(this.#rows, config.app.schema) + // Before-images cost a read per blind write; only pay when a hook + // will see them. + const writes = new WriteSet(this.#rows, config.app.schema, { + trackChanges: config.onMutationCommitted !== undefined, + }) try { // Args are validated (and parsed: defaults applied) before apply // runs; invalid args are permanent — retrying identical args can @@ -1130,9 +1140,12 @@ export function createWorkspaceDO( } args = result.value } + parsedArgs = args mutator.apply(writes.tx, args, ctx) const candidate = this.#meta.currentVersion + 1 - if (writes.flush(candidate) > 0) wroteVersion = candidate + const flushed = writes.flush(candidate) + changes = flushed.changes + if (flushed.written > 0) wroteVersion = candidate } catch (err) { if (err instanceof AppError) { appError = { code: err.code, message: err.message } @@ -1166,10 +1179,43 @@ export function createWorkspaceDO( // In-memory meta updates only after the transaction commits, so a // rollback can never leave memory ahead of storage. - if (committedVersion !== null) this.#meta.currentVersion = committedVersion + if (committedVersion !== null) { + this.#meta.currentVersion = committedVersion + this.#notifyCommitted({ + workspaceId: this.#meta.workspaceId, + name: mutation.name, + args: parsedArgs, + principal: attachment.principal, + clientId, + version: committedVersion, + changes, + }) + } return appError } + /** + * Post-commit fan-out (ARCHITECTURE.md#mutation-processing). Never awaited: + * the push handler stays synchronous (invariant 3 of ARCHITECTURE.md#invariants) + * and a slow or failing consumer cannot hold up confirmation. The promise + * goes to waitUntil so the runtime keeps the object alive until it settles. + */ + #notifyCommitted(event: MutationCommitted): void { + const hook = config.onMutationCommitted + if (!hook) return + const detail = { name: event.name, version: event.version } + try { + const result = hook(event, this.env) + if (result instanceof Promise) { + this.ctx.waitUntil( + result.catch((err: unknown) => this.#log('error', '[cf-sync] onMutationCommitted rejected', err, detail)), + ) + } + } catch (err) { + this.#log('error', '[cf-sync] onMutationCommitted threw', err, detail) + } + } + #touchClient(clientId: string): number { this.#sql.exec( `INSERT INTO clients (client_id, last_mutation_id, last_seen_at) VALUES (?, 0, ?) diff --git a/packages/server/src/engine-core.ts b/packages/server/src/engine-core.ts index d003f0a..2415241 100644 --- a/packages/server/src/engine-core.ts +++ b/packages/server/src/engine-core.ts @@ -1,5 +1,6 @@ import { AppError, MAX_ROW_BYTES, type AnySyncSchema, type MutatorTx, type StandardSchemaV1 } from '@cf-sync/protocol' import { MAX_ID_LENGTH, TABLE_NAME_RE, formatIssues, jsonByteSize } from '@cf-sync/protocol/internal' +import type { RowChange } from './config' /** * The storage-agnostic core of the workspace engine: row validation and the @@ -79,16 +80,30 @@ interface RowWrite { * the chain's net result must parse. Mutations keep validating at `put` so a * mutator reading back its own write sees the parsed output (defaults * applied), exactly what a poke will carry. + * + * `trackChanges` makes `flush` report each written row's before/after pair + * (the post-commit hook's payload). It costs one stored-row read per row the + * mutation writes blind — a `tx.get` that reached storage doubles as the + * image — so it stays off unless something consumes the list. */ export class WriteSet { #puts = new Map() #dels = new Map() + // The stored row at the mutation's first touch of it, keyed like the + // buffers, so the change list reports the net effect (before -> after) + // rather than intermediate overlay states. Insertion order is touch order. + #before = new Map | null>() + readonly #validateAtFlush: boolean + readonly #trackChanges: boolean constructor( private readonly rows: EngineRowStore, private readonly schema: AnySyncSchema, - private readonly validateAtFlush = false, - ) {} + opts: { validateAtFlush?: boolean; trackChanges?: boolean } = {}, + ) { + this.#validateAtFlush = opts.validateAtFlush ?? false + this.#trackChanges = opts.trackChanges ?? false + } readonly tx: MutatorTx = { get: (tbl, id) => { @@ -97,7 +112,13 @@ export class WriteSet { if (this.#dels.has(k)) return null const buffered = this.#puts.get(k) if (buffered) return structuredClone(buffered.data) - return this.rows.get(tbl, id) + const stored = this.rows.get(tbl, id) + if (!this.#trackChanges || this.#before.has(k)) return stored + // Read-modify-write is the common mutator shape: keep this read as + // the before-image so `put` need not fetch the row a second time. The + // caller may mutate what it gets back, so the kept copy is private. + this.#before.set(k, structuredClone(stored)) + return stored }, list: (tbl) => { if (!TABLE_NAME_RE.test(tbl)) throw new AppError('InvalidArgs', `invalid table name "${tbl}"`) @@ -109,7 +130,7 @@ export class WriteSet { }, put: (tbl, id, data) => { validateTarget(tbl, id) - const stored = this.validateAtFlush + const stored = this.#validateAtFlush ? (data as Record) : validateRow(this.schema, tbl, id, data) const bytes = jsonByteSize(stored) @@ -117,36 +138,52 @@ export class WriteSet { throw new AppError('RowTooLarge', `row ${tbl}/${id} is ${bytes} bytes (max ${MAX_ROW_BYTES})`) } const k = rowKey(tbl, id) + this.#rememberBefore(k, tbl, id) this.#dels.delete(k) this.#puts.set(k, { tbl, id, data: structuredClone(stored) }) }, del: (tbl, id) => { validateTarget(tbl, id) const k = rowKey(tbl, id) + this.#rememberBefore(k, tbl, id) this.#puts.delete(k) this.#dels.set(k, { tbl, id }) }, } - /** Flushes buffered writes stamped with `version`. Returns rows actually written. */ - flush(version: number): number { - let written = 0 + #rememberBefore(k: string, tbl: string, id: string): void { + if (this.#trackChanges && !this.#before.has(k)) this.#before.set(k, this.rows.get(tbl, id)) + } + + /** + * Flushes buffered writes stamped with `version`. `written` counts rows + * actually written or deleted; `changes` lists them as before/after pairs + * in first-touch order when `trackChanges` is on, and is empty otherwise. + */ + flush(version: number): { written: number; changes: RowChange[] } { + const after = new Map | null>() for (const { tbl, id, data } of this.#puts.values()) { - const stored = this.validateAtFlush ? validateRow(this.schema, tbl, id, data) : data - if (this.validateAtFlush) { + const stored = this.#validateAtFlush ? validateRow(this.schema, tbl, id, data) : data + if (this.#validateAtFlush) { const bytes = jsonByteSize(stored) if (bytes > MAX_ROW_BYTES) { throw new AppError('RowTooLarge', `row ${tbl}/${id} is ${bytes} bytes (max ${MAX_ROW_BYTES})`) } } this.rows.put(tbl, id, stored, version) - written++ + after.set(rowKey(tbl, id), stored) } for (const { tbl, id } of this.#dels.values()) { // Deleting a row that never existed is a no-op, not a tombstone: no // client can hold a row the server never had. - written += this.rows.del(tbl, id, version) + if (this.rows.del(tbl, id, version) > 0) after.set(rowKey(tbl, id), null) + } + const changes: RowChange[] = [] + for (const [k, before] of this.#before) { + if (!after.has(k)) continue + const target = this.#puts.get(k) ?? this.#dels.get(k)! + changes.push({ tbl: target.tbl, id: target.id, before, after: after.get(k) as Record | null }) } - return written + return { written: after.size, changes } } } diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index 0ba98d1..d1aa419 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -6,6 +6,8 @@ export { type EngineLogContext, type EngineLogger, type ExportConfig, + type MutationCommitted, + type RowChange, type WorkspaceEngineConfig, } from './config' export { createWorkspaceDO, type WorkspaceDOClass } from './do' diff --git a/packages/server/src/testing.ts b/packages/server/src/testing.ts index d0d2768..eaf5ff7 100644 --- a/packages/server/src/testing.ts +++ b/packages/server/src/testing.ts @@ -11,6 +11,7 @@ import { type TableName, } from '@cf-sync/protocol' import { formatIssues, migrationPath } from '@cf-sync/protocol/internal' +import type { RowChange } from './config' import { WriteSet, rowKey, validateRow, type EngineRowStore } from './engine-core' import { schemaFingerprint, unfingerprintableTables } from './fingerprint' @@ -21,6 +22,7 @@ export { schemaFingerprint } // and dies outside workerd with an error that never names this fix. export { AppError, crudMutators, defineApp, defineMutators, defineSchema } from '@cf-sync/protocol' export type { AppDefinition, MigrationTx, MutatorContext, MutatorTx } from '@cf-sync/protocol' +export type { RowChange } from './config' /** Options for {@link createTestEngine}: initial state, stored schema version, and the identity mutators observe. */ export interface TestEngineOptions { @@ -49,6 +51,12 @@ export interface TestEngineOptions { export interface TestMutationResult { /** `code` is an {@link EngineErrorCode} built-in or an app-defined `AppError` code. */ error?: { code: EngineErrorCode | (string & {}); message: string } + /** + * The rows the mutation wrote or deleted as before/after pairs — what the + * Durable Object hands to `onMutationCommitted`. Empty on a permanent + * error and when the writes net to nothing. + */ + changes: RowChange[] } interface StoredRow { @@ -165,7 +173,7 @@ export class TestEngine 0) this.#version = candidate + const flushed = writes.flush(candidate) + changes = flushed.changes + if (flushed.written > 0) this.#version = candidate } catch (err) { if (err instanceof AppError) { appError = { code: err.code, message: err.message } @@ -233,7 +244,7 @@ export class TestEngine `hook-${++n}-${Date.now()}` + +function connect(workspace: string, clientId: string, principal?: string): Promise { + return TestClient.connect(workspace, clientId, '/rollout', principal ? { 'x-test-principal': principal } : {}) +} + +async function evict(workspaceId: string): Promise { + const stub = env.ROLLOUT.get(env.ROLLOUT.idFromName(workspaceId)) + await runInDurableObject(stub, async (_instance, state) => { + state.abort() + }).catch(() => { + // abort() kills the object; the call itself is expected to fail + }) +} + +function admin(workspaceId: string, op: string, body?: unknown): Promise { + return SELF.fetch(`https://test/rollout-admin/${workspaceId}/${op}`, { + method: 'POST', + headers: { 'x-test-admin': 'yes', 'content-type': 'application/json' }, + body: body === undefined ? null : JSON.stringify(body), + }) +} + +afterEach(() => { + delete rolloutConfig.onMutationCommitted + delete rolloutConfig.logger + rolloutConfig.app = rolloutApp + vi.restoreAllMocks() +}) + +describe('onMutationCommitted', () => { + it('fires once per committed mutation with net before/after images and the committed version', async () => { + const workspace = ws() + const hook = vi.fn() + rolloutConfig.onMutationCommitted = hook + + const c1 = await connect(workspace, 'c1', 'alice') + await c1.syncOnce() + c1.push([ + { id: 1, name: 'sync.put', args: { tbl: 'todos', id: 't1', data: { title: 'a' } } }, + { id: 2, name: 'sync.put', args: { tbl: 'todos', id: 't1', data: { title: 'b' } } }, + { id: 3, name: 'sync.del', args: { tbl: 'todos', id: 't1' } }, + ]) + await c1.pokeUntilLmid(3) + c1.close() + + expect(hook).toHaveBeenCalledTimes(3) + const events = hook.mock.calls.map((call) => call[0] as MutationCommitted) + expect(events[0]).toEqual({ + workspaceId: workspace, + name: 'sync.put', + args: { tbl: 'todos', id: 't1', data: { title: 'a' } }, + principal: 'alice', + clientId: 'c1', + version: 1, + changes: [{ tbl: 'todos', id: 't1', before: null, after: { title: 'a' } }], + }) + expect(events[1]).toMatchObject({ + name: 'sync.put', + version: 2, + changes: [{ tbl: 'todos', id: 't1', before: { title: 'a' }, after: { title: 'b' } }], + }) + expect(events[2]).toMatchObject({ + name: 'sync.del', + version: 3, + changes: [{ tbl: 'todos', id: 't1', before: { title: 'b' }, after: null }], + }) + // The worker env rides along so a hook can reach its own bindings. + expect(hook.mock.calls[0]![1]).toHaveProperty('ROLLOUT') + }) + + it('passes args as the mutator received them: parsed, with defaults applied', async () => { + const workspace = ws() + const hook = vi.fn() + rolloutConfig.onMutationCommitted = hook + + const c1 = await connect(workspace, 'c1') + await c1.syncOnce() + c1.push([{ id: 1, name: 'sync.put', args: { tbl: 'typed', id: 'x', data: { name: 'n' } } }]) + await c1.pokeUntilLmid(1) + c1.close() + + const event = hook.mock.calls[0]![0] as MutationCommitted + expect(event.principal).toBeUndefined() + expect(event.changes).toEqual([{ tbl: 'typed', id: 'x', before: null, after: { name: 'n', n: 1 } }]) + }) + + it('stays silent for rejected mutations and for writes that net to nothing', async () => { + const workspace = ws() + const hook = vi.fn() + rolloutConfig.onMutationCommitted = hook + + const c1 = await connect(workspace, 'c1') + await c1.syncOnce() + c1.push([ + // Permanent error: the LMID advances, nothing is written. + { id: 1, name: 'sync.put', args: { tbl: 'not-a-table', id: 't1', data: { title: 'a' } } }, + // Deleting a row that never existed writes no tombstone. + { id: 2, name: 'sync.del', args: { tbl: 'todos', id: 'absent' } }, + { id: 3, name: 'sync.put', args: { tbl: 'todos', id: 't1', data: { title: 'a' } } }, + ]) + await c1.pokeUntilLmid(3) + c1.close() + + expect(hook).toHaveBeenCalledTimes(1) + expect(hook.mock.calls[0]![0]).toMatchObject({ name: 'sync.put', version: 1 }) + }) + + it('a throwing or rejecting hook is reported to the logger and never blocks confirmation', async () => { + const workspace = ws() + const logger = vi.fn() + rolloutConfig.logger = logger + rolloutConfig.onMutationCommitted = (event) => { + if (event.version === 1) throw new Error('sync boom') + return Promise.reject(new Error('async boom')) + } + + const c1 = await connect(workspace, 'c1') + await c1.syncOnce() + c1.push([ + { id: 1, name: 'sync.put', args: { tbl: 'todos', id: 't1', data: { title: 'a' } } }, + { id: 2, name: 'sync.put', args: { tbl: 'todos', id: 't2', data: { title: 'b' } } }, + ]) + await c1.pokeUntilLmid(2) + expect(c1.rows.get('todos/t2')).toEqual({ title: 'b' }) + c1.close() + + await vi.waitFor(() => expect(logger).toHaveBeenCalledTimes(2)) + const [thrown, rejected] = logger.mock.calls.map((call) => ({ + level: call[0], + message: String(call[1]), + context: call[2], + error: call[3], + detail: call[4], + })) + expect(thrown).toMatchObject({ + level: 'error', + message: '[cf-sync] onMutationCommitted threw', + context: { workspaceId: workspace }, + detail: { name: 'sync.put', version: 1 }, + }) + expect((thrown!.error as Error).message).toBe('sync boom') + expect(rejected).toMatchObject({ + level: 'error', + message: '[cf-sync] onMutationCommitted rejected', + detail: { name: 'sync.put', version: 2 }, + }) + expect((rejected!.error as Error).message).toBe('async boom') + }) + + it('does not fire for schema migrations', async () => { + const workspace = ws() + const hook = vi.fn() + rolloutConfig.onMutationCommitted = hook + + const c1 = await connect(workspace, 'c1') + await c1.syncOnce() + c1.push([{ id: 1, name: 'sync.put', args: { tbl: 'todos', id: 't1', data: { title: 'a' } } }]) + await c1.pokeUntilLmid(1) + c1.close() + expect(hook).toHaveBeenCalledTimes(1) + + // "Deploy" v2 with a migration that rewrites every row, then wake the DO. + rolloutConfig.app = defineApp({ + version: 2, + schema: testSchema, + mutators: { ...crudMutators(testSchema) }, + migrations: { + 2: (tx) => { + for (const { id, data } of tx.list('todos')) tx.put('todos', id, { ...data, migrated: true }) + }, + }, + }) + await evict(workspace) + const c2 = await connect(workspace, 'c2') + c2.schemaVersion = 2 + const poke = await c2.syncOnce() + expect(c2.rows.get('todos/t1')).toEqual({ title: 'a', migrated: true }) + expect(poke.cursor.version).toBe(2) + c2.close() + + expect(hook).toHaveBeenCalledTimes(1) + }) + + it('does not fire for admin import or reset', async () => { + const workspace = ws() + const hook = vi.fn() + rolloutConfig.onMutationCommitted = hook + + const c1 = await connect(workspace, 'c1') + await c1.syncOnce() + const imported = await admin(workspace, 'import', { + formatVersion: 1, + schemaVersion: 1, + rows: [{ tbl: 'todos', id: 't1', data: { title: 'imported' } }], + }) + expect(imported.status).toBe(200) + const poke = await c1.nextPoke() + expect(poke.patch).toContainEqual({ op: 'put', tbl: 'todos', id: 't1', value: { title: 'imported' } }) + expect((await admin(workspace, 'reset')).status).toBe(200) + await c1.nextPoke() + c1.close() + + expect(hook).not.toHaveBeenCalled() + }) +}) diff --git a/packages/server/test/fixture/worker.ts b/packages/server/test/fixture/worker.ts index 41bfd88..51777e8 100644 --- a/packages/server/test/fixture/worker.ts +++ b/packages/server/test/fixture/worker.ts @@ -147,17 +147,29 @@ const verdictRoute = createSyncRoute({ }, }) const compactRoute = createSyncRoute({ namespace: (env) => env.COMPACT, pathPrefix: '/compact', authorize: 'public' }) -const rolloutRoute = createSyncRoute({ namespace: (env) => env.ROLLOUT, pathPrefix: '/rollout', authorize: 'public' }) +// The rollout DO also hosts the post-commit hook drills (its config is the +// mutable one), so its route stamps a principal when a test asks for one. +const rolloutRoute = createSyncRoute({ + namespace: (env) => env.ROLLOUT, + pathPrefix: '/rollout', + authorize: (request) => ({ ok: true, principal: request.headers.get('x-test-principal') ?? undefined }), +}) const adminRoute = createAdminRoute({ namespace: (env) => env.WORKSPACE, authorize: (request) => request.headers.get('x-test-admin') === 'yes', }) +const rolloutAdminRoute = createAdminRoute({ + namespace: (env) => env.ROLLOUT, + pathPrefix: '/rollout-admin', + authorize: (request) => request.headers.get('x-test-admin') === 'yes', +}) // Routes compose with ?? (null = "not mine"); the sync fetch is the terminal // handler with its own 404 fallback. export default { fetch: async (request: Request, env: Env) => (await adminRoute(request, env)) ?? + (await rolloutAdminRoute(request, env)) ?? (await verdictRoute(request, env)) ?? (await compactRoute(request, env)) ?? (await rolloutRoute(request, env)) ?? diff --git a/packages/server/test/node/write-changes.test.ts b/packages/server/test/node/write-changes.test.ts new file mode 100644 index 0000000..36077fb --- /dev/null +++ b/packages/server/test/node/write-changes.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, it } from 'vitest' +import { z } from 'zod' +import { AppError, createTestEngine, crudMutators, defineApp, defineMutators, defineSchema } from '../../src/testing' + +// The change list a mutation reports — what onMutationCommitted receives — +// is the net effect per row: the stored image at first touch against what +// the mutation left, whatever happened in between. + +const schema = defineSchema({ + todos: z.object({ title: z.string(), done: z.boolean().default(false) }), +}) + +const mutators = defineMutators(schema, { + ...crudMutators(schema), + 'todo.toggleTwice': { + args: z.object({ id: z.string() }), + apply: (tx, { id }) => { + const row = tx.get('todos', id) + if (!row) throw new AppError('NotFound', id) + tx.put('todos', id, { ...row, done: !row.done }) + // Reads see the overlay, so this flips it back: the net effect is a + // rewrite of the same image. + const flipped = tx.get('todos', id)! + tx.put('todos', id, { ...flipped, done: !flipped.done }) + }, + }, + 'todo.mutateInPlace': { + args: z.object({ id: z.string() }), + apply: (tx, { id }) => { + const row = tx.get('todos', id)! + row.done = true + tx.put('todos', id, row) + }, + }, + 'todo.createThenDelete': { + args: z.object({ id: z.string() }), + apply: (tx, { id }) => { + tx.put('todos', id, { title: 'ephemeral' }) + tx.del('todos', id) + }, + }, + 'todo.deleteThenCreate': { + args: z.object({ id: z.string() }), + apply: (tx, { id }) => { + tx.del('todos', id) + tx.put('todos', id, { title: 'reborn' }) + }, + }, + 'todo.writeThenFail': { + args: z.object({ id: z.string() }), + apply: (tx, { id }) => { + tx.put('todos', id, { title: 'ghost' }) + throw new AppError('Nope', 'wrote then failed') + }, + }, +}) + +const app = defineApp({ version: 1, schema, mutators }) + +describe('mutation change list', () => { + it('reports inserts, updates, and deletes as before/after pairs, in touch order', () => { + const engine = createTestEngine(app) + + expect(engine.mutate('sync.put', { tbl: 'todos', id: 'a', data: { title: 'a' } }).changes).toEqual([ + { tbl: 'todos', id: 'a', before: null, after: { title: 'a', done: false } }, + ]) + expect(engine.mutate('sync.put', { tbl: 'todos', id: 'a', data: { title: 'a2' } }).changes).toEqual([ + { tbl: 'todos', id: 'a', before: { title: 'a', done: false }, after: { title: 'a2', done: false } }, + ]) + expect(engine.mutate('sync.del', { tbl: 'todos', id: 'a' }).changes).toEqual([ + { tbl: 'todos', id: 'a', before: { title: 'a2', done: false }, after: null }, + ]) + }) + + it('collapses repeated touches of one row to its net effect', () => { + const engine = createTestEngine(app) + engine.seed('todos', 'a', { title: 'a' }) + + expect(engine.mutate('todo.toggleTwice', { id: 'a' }).changes).toEqual([ + { tbl: 'todos', id: 'a', before: { title: 'a', done: false }, after: { title: 'a', done: false } }, + ]) + expect(engine.mutate('todo.deleteThenCreate', { id: 'a' }).changes).toEqual([ + { tbl: 'todos', id: 'a', before: { title: 'a', done: false }, after: { title: 'reborn', done: false } }, + ]) + }) + + it('keeps the before-image private when a mutator edits the row it read in place', () => { + const engine = createTestEngine(app) + engine.seed('todos', 'a', { title: 'a' }) + + expect(engine.mutate('todo.mutateInPlace', { id: 'a' }).changes).toEqual([ + { tbl: 'todos', id: 'a', before: { title: 'a', done: false }, after: { title: 'a', done: true } }, + ]) + }) + + it('is empty when nothing was written, and the version does not move', () => { + const engine = createTestEngine(app) + const before = engine.version + + expect(engine.mutate('sync.del', { tbl: 'todos', id: 'absent' })).toEqual({ changes: [] }) + expect(engine.mutate('todo.createThenDelete', { id: 'x' })).toEqual({ changes: [] }) + expect(engine.version).toBe(before) + }) + + it('is empty on a permanent error, with the write discarded', () => { + const engine = createTestEngine(app) + + const result = engine.mutate('todo.writeThenFail', { id: 'g' }) + expect(result.error).toMatchObject({ code: 'Nope' }) + expect(result.changes).toEqual([]) + expect(engine.get('todos', 'g')).toBeNull() + }) +})