server: onMutationCommitted, a post-commit hook with net row changes — 0.2.1

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

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.

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
 

docs/reference/server.md

@@ -85,6 +85,36 @@ logger: (level, message, { workspaceId }, ...detail) => {
 }
 ```
 
+### onMutationCommitted
+
+`(event: MutationCommitted, env: Env) => void | Promise<void>`
+
+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<Response>` · `…Promise<Response | null>`

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`.

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": {

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<string, unknown> | null
+  after: Record<string, unknown> | 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<S extends AnySyncSchema = AnySyncSchema, Env = unknown> {
   /**
@@ -186,4 +217,22 @@ export interface WorkspaceEngineConfig<S extends AnySyncSchema = AnySyncSchema,
    * carries an {@link EngineLogContext} naming the workspace it came from.
    */
   logger?: EngineLogger
+  /**
+   * 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 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 {@link WorkspaceEngineConfig.logger} and
+   * affects nothing else. Delivery is at-most-once: a workspace evicted with
+   * the promise in flight does not replay it, so consumers that must not
+   * miss an event make their effects idempotent on `version` and keep a way
+   * to rebuild from an admin `export`.
+   */
+  onMutationCommitted?: (event: MutationCommitted, env: Env) => void | Promise<void>
 }

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<S extends AnySyncSchema, Env = unknown>(
         // 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<S extends AnySyncSchema, Env = unknown>(
      * 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<S extends AnySyncSchema, Env = unknown>(
       }
       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<S extends AnySyncSchema, Env = unknown>(
           // 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<S extends AnySyncSchema, Env = unknown>(
               }
               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<S extends AnySyncSchema, Env = unknown>(
 
       // 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, ?)