feat(plugins): atomic multi-document storage batch (ctx.storage.batch)

.changeset/plugin-storage-atomic-batch.md

@@ -0,0 +1,9 @@
+---
+"emdash": minor
+---
+
+Adds `ctx.storage.batch([...])` — apply several conditional writes (`insert` / `updateIf`) across multiple documents and collections all-or-nothing. The batch commits only if every op's guard passes; otherwise it rolls back the whole batch and reports which op failed (`{ applied: false, failedIndex, reason, conflictField? }`). This is the atomic primitive behind coupled writes like claim ∧ decrement ∧ flip.
+
+Atomic on Postgres and SQLite (a real transaction) and on Cloudflare D1 (raw `env.DB.batch()` with interleaved zero-rows guard assertions). Guard and uniqueness outcomes are reported, never thrown; malformed ops (float delta, a field in both `set` and `delta`, unknown op, empty ops array, `updateIf` with neither `set` nor `delta`, `insert` without `data`) throw. A batch is capped at **50 ops** (aligned with D1's per-batch statement / bound-parameter limits — each guarded op compiles to up to two D1 statements); an over-limit batch throws. A collection named `batch` is now rejected at declaration time (it would collide with the new method).
+
+Caveat (inherited from the numeric-guard support): numeric guards inside batch ops fall back to a sequential scan on Postgres. On D1's failure path, `failedIndex` / `reason` are best-effort under concurrent writers (the committed state is always correct — resolve by durable state).

.changeset/plugin-storage-batch-cloudflare.md

@@ -0,0 +1,5 @@
+---
+"@emdash-cms/cloudflare": minor
+---
+
+Adds the `storage/batch` bridge method so sandboxed plugins on Cloudflare get the same atomic multi-document batch (`ctx.storage.batch`) as in-process plugins. On D1 the batch runs through `env.DB.batch()` with interleaved zero-rows guard assertions, so a failed guard rolls back every coupled write. Returns the identical `BatchResult` shape as the in-process and workerd paths.

.changeset/plugin-storage-batch-workerd.md

@@ -0,0 +1,5 @@
+---
+"@emdash-cms/sandbox-workerd": minor
+---
+
+Adds the `storage/batch` bridge method so sandboxed plugins on the workerd-on-Node runtime get the same atomic multi-document batch (`ctx.storage.batch`) as in-process plugins. Every op's declared collection is validated before execution (a batch cannot smuggle a write to an undeclared collection), and the batch runs in a real transaction against the host database with the identical `BatchResult` shape as the Cloudflare D1 path.

packages/cloudflare/package.json

@@ -115,12 +115,14 @@
   "devDependencies": {
     "@arethetypeswrong/cli": "catalog:",
     "@astrojs/cloudflare": "catalog:",
+    "@cloudflare/vitest-pool-workers": "catalog:",
     "@cloudflare/workers-types": "catalog:",
     "@types/pg": "^8.16.0",
     "publint": "catalog:",
     "tsdown": "catalog:",
     "typescript": "catalog:",
-    "vitest": "catalog:"
+    "vitest": "catalog:",
+    "wrangler": "catalog:"
   },
   "repository": {
     "type": "git",

packages/cloudflare/src/sandbox/bridge.ts

@@ -9,8 +9,8 @@
 
 import type { D1Database } from "@cloudflare/workers-types";
 import { WorkerEntrypoint } from "cloudflare:workers";
-import type { SandboxEmailSendCallback } from "emdash";
-import { ulid, PluginStorageRepository } from "emdash";
+import type { SandboxEmailSendCallback, BatchOp } from "emdash";
+import { ulid, PluginStorageRepository, applyPluginStorageBatchD1 } from "emdash";
 import { Kysely } from "kysely";
 import { D1Dialect } from "kysely-d1";
 
@@ -487,6 +487,49 @@ export class PluginBridge extends WorkerEntrypoint<PluginBridgeEnv, PluginBridge
 		});
 	}
 
+	/**
+	 * Atomic multi-document batch. This is the ONE place that uses the raw
+	 * `env.DB.batch()` mechanism (D1 has no interactive transactions):
+	 * `applyPluginStorageBatchD1` compiles each op via a SQLite-dialect Kysely and
+	 * interleaves zero-rows assertions so a failed guard rolls back the WHOLE
+	 * batch. Validates every op's declared collection (anti-smuggling) and per-op
+	 * where/set/delta shapes up front — symmetric with `storageUpdateIf` — so the
+	 * `BatchResult` shape and error messages match the workerd bridge exactly.
+	 */
+	async storageBatch(ops: unknown): Promise<unknown> {
+		const { pluginId, storageCollections } = this.ctx.props;
+		if (!Array.isArray(ops)) {
+			throw new Error("storage/batch requires an array of ops");
+		}
+		for (const op of ops) {
+			if (!isJsonObject(op)) throw new Error("batch op must be an object");
+			if (op.op !== "insert" && op.op !== "updateIf") {
+				throw new Error(`batch op has unknown op: ${String(op.op)}`);
+			}
+			if (typeof op.collection !== "string") {
+				throw new Error("batch op requires a string collection");
+			}
+			if (typeof op.id !== "string") throw new Error("batch op requires a string id");
+			if (!storageCollections.includes(op.collection)) {
+				throw new Error(`Storage collection not declared: ${op.collection}`);
+			}
+			if (op.op === "updateIf") {
+				if (!isJsonObject(op.where)) {
+					throw new Error("storage/updateIf requires an object `where`");
+				}
+				if (op.set !== undefined && !isJsonObject(op.set)) {
+					throw new Error("storage/updateIf `set` must be an object when provided");
+				}
+				if (op.delta !== undefined && !isJsonObject(op.delta)) {
+					throw new Error("storage/updateIf `delta` must be an object when provided");
+				}
+			}
+		}
+		// env.DB (D1Database) is structurally a D1BatchBinding (prepare/bind/first/batch).
+		// eslint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- each op validated above; applyPluginStorageBatchD1 re-validates set/delta and reports guard outcomes.
+		return applyPluginStorageBatchD1(this.env.DB, pluginId, ops as BatchOp[]);
+	}
+
 	// =========================================================================
 	// Content Operations - capability-gated
 	// =========================================================================

packages/cloudflare/src/sandbox/wrapper.ts

@@ -92,6 +92,9 @@ function createContext(env) {
 	const storage = new Proxy({}, {
 		get(_, collectionName) {
 			if (typeof collectionName !== "string") return undefined;
+			// "batch" is the cross-collection atomic primitive on the storage
+			// access object, not a per-collection accessor.
+			if (collectionName === "batch") return (ops) => bridge.storageBatch(ops);
 			return createStorageCollection(collectionName);
 		}
 	});

packages/cloudflare/tests/d1/migrations/0001_plugin_storage.sql

@@ -0,0 +1,36 @@
+-- Real `_plugin_storage` / `_plugin_indexes` schema for the D1 batch feature
+-- tests. Mirrors core migration 004_plugins (SQLite/D1 forms) so the tests
+-- exercise `applyPluginStorageBatchD1` against the ACTUAL production schema, not
+-- a scratch `probe` table. Plus a partial UNIQUE expression index on
+-- `reservations.idempotency_key` (the shape createStorageIndexes materializes),
+-- and its `_plugin_indexes` tracking row so `conflictField` recovery resolves.
+CREATE TABLE _plugin_storage (
+	plugin_id TEXT NOT NULL,
+	collection TEXT NOT NULL,
+	id TEXT NOT NULL,
+	data TEXT NOT NULL,
+	created_at TEXT,
+	updated_at TEXT,
+	PRIMARY KEY (plugin_id, collection, id)
+);
+
+CREATE TABLE _plugin_indexes (
+	plugin_id TEXT NOT NULL,
+	collection TEXT NOT NULL,
+	index_name TEXT NOT NULL,
+	fields TEXT NOT NULL,
+	created_at TEXT,
+	PRIMARY KEY (plugin_id, collection, index_name)
+);
+
+CREATE UNIQUE INDEX uidx_plugin_shop_reservations_idempotency_key
+	ON _plugin_storage(json_extract(data, '$.idempotency_key'))
+	WHERE plugin_id = 'shop' AND collection = 'reservations';
+
+INSERT INTO _plugin_indexes (plugin_id, collection, index_name, fields)
+VALUES (
+	'shop',
+	'reservations',
+	'uidx_plugin_shop_reservations_idempotency_key',
+	'["idempotency_key"]'
+);